Go to Qiita Advent Calendar Top
LoginSignup
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@nohanaga(Nobusuke Hanagasaki)

Microsoft Entra ID による MCP 認証テスト：FastMCP サーバーのセキュアな実装メモ

Last updated at Posted at 2025-11-14

Microsoft Entra ID を使った MCP 認証システムを Frontend、Backend、Downstream API という3つの構造で実現するためのサンプルを作成します。本記事では、Flask Frontend、FastMCP Backend、Azure サービスという実際の構成を例に、3つの認証フェーズに分けて中身を慎重に見ていきます。

構成として以下のような Full Python で実現してみます。

  • Frontend: Flask
  • Backend: FastMCP + Microsoft Agent Framework with MCPStreamableHTTPTool
  • Authorization: Identity for Flask, FastMCP JWTVerifier, MSAL for Python

認証周りの処理は複雑で独自に実装することは避けるべきであり、MSAL for Python およびそのラッパーである Identity for Flask を利用するようにしています。

image.png

Entra ID で取得されるすべてのトークン情報を可視化して、内部理解を促進できるような UI にしています。

アーキテクチャ概要

┌─────────────────────────────────────────────────────────────────────────┐
│                          ユーザー (ブラウザ)                              │
└────────────────────────────┬────────────────────────────────────────────┘
                             │
                  [フェーズ1: ユーザー認証]
                  Authorization Code Flow with PKCE
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                     Microsoft Entra ID (認証基盤)                             │
│  ・ユーザー認証 (MFA 対応)                                                │
│  ・初回トークン発行 (Graph API 用)                                        │
│  ・委任トークン発行 (Backend API 用)                                      │
│  ・OBO トークン交換 (Downstream API 用)                                   │
└────────────────────────────┬────────────────────────────────────────────┘
                             │
                  [フェーズ2: 委任アクセス]
                  Delegated Access Token
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                      Frontend (Flask - Port 5001)                        │
│  ・Authorization Code Flow with PKCE                                     │
│  ・セッション管理とトークンキャッシュ                                     │
│  ・AI チャット UI (Microsoft Agent Framework)                            │
└────────────────────────────┬────────────────────────────────────────────┘
                             │
                  Authorization: Bearer {delegated_token}
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                     Backend (FastMCP - Port 8000)                        │
│  ・JWT 検証 (署名・Audience・Issuer・スコープ)                            │
│  ・MCP ツール提供                                                         │
└────────────────────────────┬────────────────────────────────────────────┘
                             │
                  [フェーズ3: On-Behalf-Of フロー]
                  OBO Token Exchange
                             │
                             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                 Downstream API (Azure サービス)                          │
│  ・Azure AI Search など                                                  │
│  ・Azure サービス群                                                       │
│  ・Microsoft Graph API                                                   │
└─────────────────────────────────────────────────────────────────────────┘

セットアップ

前提条件

  • Python 3.10 以上
  • Azure サブスクリプション
  • 実験用 Microsoft Entra ID テナント

Microsoft Entra ID アプリケーション登録

Backend 用アプリ (API_APP_ID)

  1. Azure Portal → Entra ID → App registrations → New registration
  2. API の公開:
    • Application ID URI: api://{CLIENT_ID}
    • Scope 作成: access_as_user (Admins and users)
  3. API のアクセス許可: ダウンストリーム API 権限を追加（オプション）
    • 例: Azure AI Search → user_impersonation
    • 「<テナント名> に管理者の同意を与えます」ボタンをクリック
  4. Authentication → Federated credentials → Add credential:（オプション）
    • Issuer: https://login.microsoftonline.com/{TENANT_ID}/v2.0
    • Subject: {UMI_CLIENT_ID} (Managed IdentityのクライアントID)

Frontend 用アプリ (FLASK_CLIENT_ID)

  1. Azure Portal → Entra ID → App registrations → New registration
  2. Redirect URI: http://localhost:5001/signin-oidc
  3. API のアクセス許可:
    • Microsoft Graph → User.Read (Delegated)
    • api://{API_APP_ID} → access_as_user (Delegated)
    • 「<テナント名> に管理者の同意を与えます」ボタンをクリック
  4. 証明書とシークレット: クライアント シークレットを作成

フェーズ1: ユーザー認証 (Authorization Code Flow with PKCE)

フェーズ1では、エンドユーザーがブラウザを通じて Microsoft Entra ID で認証を行い、Frontendアプリケーション (Flask) がユーザーの初回アクセストークンを取得します。このフェーズでは Authorization Code Flow with PKCE (Proof Key for Code Exchange) を使用し、認可コードの横取り攻撃を防止します。MSAL Python ライブラリのラッパーである Identity for Flask ライブラリが提供するセッションストアとトークンキャッシュを活用して OAuth 2.0 の一連の処理を隠蔽します。

推奨事項

以下のベストプラクティスを推奨 (参考):

  • PKCE の使用: SPA (Single Page Apps) では必須、その他のアプリケーションタイプ (Web アプリ、ネイティブアプリ) では強く推奨
  • code_challenge_method: SHA256 (S256) を使用すべき。plain の使用は避ける
  • State パラメータ: CSRF 攻撃を防ぐため、ランダムな state 値を生成・検証
  • Redirect URI の厳密な検証: 登録された URI と完全一致することを確認
  • Nonce の検証: ID トークンのリプレイ攻撃を防止

フロー詳細図

実装のポイント

Identity for Flask による自動化

本サンプルでは、Microsoft が紹介する Identity for Flask ライブラリを使用して、PKCE を含む認証フローを自動化しています。

from identity.flask import Auth

auth = Auth(
    app,
    authority=f"https://login.microsoftonline.com/{TENANT_ID}",
    client_id=FLASK_CLIENT_ID,
    client_credential=FLASK_CLIENT_SECRET,
)

@app.route("/")
@auth.login_required(scopes=["User.Read"])
def index(*, context):
    """
    認証が必要なエンドポイント
    
    @auth.login_required デコレーターが以下を自動実行:
    - 未ログインの場合: /authorize へリダイレクト (PKCE パラメーター付き)
    - ログイン済みの場合: セッションからトークンを取得
    - トークン期限切れの場合: Refresh Token で自動更新
    """
    user = context['user']
    access_token = context['access_token']
    
    return render_template('index.html', user=user)

取得されるトークンの詳細

Access Token (Graph API 用)

{
  "aud": "https://graph.microsoft.com",
  "iss": "https://sts.windows.net/{TENANT_ID}/",
  "iat": 1699000000,
  "nbf": 1699000000,
  "exp": 1699003600,
  "scp": "User.Read",
  "oid": "{USER_OBJECT_ID}",
  "upn": "user@contoso.com",
  "name": "田中 太郎",
  "tid": "{TENANT_ID}",
  "ver": "2.0"
}

ID Token (ユーザー情報)

{
  "aud": "{FLASK_CLIENT_ID}",
  "iss": "https://sts.windows.net/{TENANT_ID}/",
  "iat": 1699000000,
  "exp": 1699003600,
  "name": "田中 太郎",
  "oid": "{USER_OBJECT_ID}",
  "preferred_username": "user@contoso.com",
  "tid": "{TENANT_ID}",
  "nonce": "abc123",
  "ver": "2.0"
}

image.png

フェーズ2: ユーザー委任のアクセストークン (User+App Access Token)

フェーズ2では、Frontendアプリが Backend API 専用のユーザー委任のアクセストークン を取得します。このトークンは、ユーザーが既にログイン済みであることを前提に、ユーザーに代わってBackend API を呼び出す権限を証明します。

なぜユーザー委任のトークンが必要なのか？

以下の理由によりフェーズ1で取得した Graph API 用トークンをBackend API にそのまま使用できない。

課題 従来のアプローチ ユーザー委任のトークンアプローチ
トークンの用途分離 Graph API 用トークンをBackendにも使用 Backend API 専用のトークンを発行
Audience 不一致 aud: https://graph.microsoft.com aud: api://{API_APP_ID}
セキュリティ境界 Frontendが多くの権限を要求 Backendは最小限の権限のみ検証
監査証跡 アプリケーションとしてアクセス ユーザーとして追跡可能

推奨事項

Web API を呼び出す Web アプリに対して、以下を推奨 (参考):

  • 専用の Delegated Permissions を定義: Backend API 用に access_as_user などのスコープを作成。最小権限の原則に従い、必要最小限のスコープを定義
  • Application ID URI の設定: Backend API のスコープは api://{API_APP_ID}/access_as_user の形式で定義
  • 段階的同意 (Incremental Consent): 必要なときに追加スコープを要求
  • Refresh Token の活用: 長時間のセッションでもトークンを自動更新
  • トークンの改ざん防止: JWKS 公開鍵を使用して RS256 署名を検証

フロー詳細図

実装のポイント

Frontend: ユーザー委任のトークン取得

Microsoft Agent Framework から MCP ツールを呼び出す場合、MCPStreamableHTTPTool を使用します。MCP に認証を追加する場合、headers に Bearer トークンをセットすることができます。FastMCP において、Server-Sent Events（SSE）は非推奨になったため、Streamable HTTP を用います。

# API_SCOPES は Backend API 用の Delegated Permissions (スコープ)
API_SCOPES = [f"api://{API_APP_ID}/access_as_user"]

@app.route("/chat", methods=["POST"])
@auth.login_required(scopes=API_SCOPES)
async def chat(*, context):
    """
    @auth.login_required(scopes=API_SCOPES) により:
    1. セッションキャッシュをチェック
    2. Backend API用トークンが存在しない場合、自動的に取得
    3. context['access_token'] に Backend API用トークンが設定される
    """
    access_token = context['access_token']
    
    # Backend への HTTP リクエストに Bearer Token を付与
    auth_mcp = MCPStreamableHTTPTool(
        name="Authenticated User Info",
        url="http://localhost:8000/mcp",
        description="認証されたユーザー情報を取得するためのツールです。ユーザーのプロフィールや設定に関する情報を提供します。",
        headers={"Authorization": f"Bearer {access_token}"},
    )
    tools.append(auth_mcp)
    print(f"  Auth MCP added with token (first 20 chars): {access_token[:20]}...")

Backend: JWT 検証 (FastMCP JWTVerifier)

JWTVerifier は、FastMCP サーバーを保護する際に、クライアントが Bearer トークン（JWT）を付与してリクエストを行った場合に、そのトークンが「有効か」「署名が正しいか」「発行者・対象（audience）が想定どおりか」「有効期限が切れていないか」等を検証するために用いられます。

from fastmcp.server.auth.providers.jwt import JWTVerifier

# JWTVerifier の設定
auth = JWTVerifier(
    jwks_uri=f"https://login.microsoftonline.com/{TENANT_ID}/discovery/v2.0/keys",
    issuer=f"https://sts.windows.net/{TENANT_ID}/",
    audience=f"api://{API_APP_ID}",
    required_scopes=["access_as_user"]
)

mcp = FastMCP("Microsoft Entra ID Protected MCP Server", auth=auth)

@mcp.tool()
def secure_ping() -> dict:
    """
    JWTVerifier が自動的にトークンを検証:
    1. JWKS エンドポイントから公開鍵を取得 (自動キャッシュ)
    2. 署名検証 (RS256 アルゴリズム)
    3. Audience 検証: aud == "api://{API_APP_ID}"
    4. Issuer 検証: iss == "https://sts.windows.net/{TENANT_ID}/"
    5. スコープ検証: "access_as_user" が含まれるか
    6. 有効期限検証: exp, nbf クレーム
    """
    return {
        "ok": True,
        "message": "Authenticated ping successful"
    }

取得されるユーザー委任のアクセストークンの詳細

{
  "aud": "api://12345678-1234-1234-1234-123456789abc",
  "iss": "https://sts.windows.net/{TENANT_ID}/",
  "iat": 1699000000,
  "nbf": 1699000000,
  "exp": 1699003600,
  "scp": "access_as_user",
  "oid": "{USER_OBJECT_ID}",
  "upn": "user@contoso.com",
  "name": "田中 太郎",
  "appid": "{FLASK_CLIENT_ID}",
  "tid": "{TENANT_ID}",
  "ver": "2.0"
}

重要なクレーム:

  • aud: Backend API の Application ID URI (api://{API_APP_ID})
  • scp: Delegated Permissions (委任されたスコープ access_as_user)
  • oid, upn: 元のユーザー情報を保持 (監査証跡)
  • appid: Frontendアプリの Client ID (User+App を表す)

トークンキャッシュとリフレッシュ戦略

初回リクエスト

  1. セッションキャッシュに Backend API用トークンが存在しない
  2. Refresh Token を使用して新しい委任トークンを取得
  3. セッションキャッシュに保存 (有効期限: 60-90分)

2回目以降のリクエスト

  1. セッションキャッシュから Backend API用トークンを取得
  2. Entra ID への追加リクエスト不要
  3. 有効期限切れの場合のみ自動更新

image.png

チャット UI で「あなたの情報を教えて」と入力すると get_user_info ツールが呼ばれ、認証されたユーザー情報を取得し、委任アクセストークンの詳細な検証を行います。

実装した検証項目:

# 検証項目 内容
1 Audience (aud) トークンがこの Backend API (api://{API_APP_ID}) 向けに発行されたか確認
2 Issuer (iss) 信頼できるテナント (https://sts.windows.net/{TENANT_ID}/) から発行されたか確認
3 Tenant ID (tid) 正しいテナント ID と一致するか確認
4 有効期限 (exp) トークンが期限切れでないか確認
5 Not Before (nbf) トークンが有効期間内（まだ早すぎない）か確認
6 Application ID (appid/azp) どのフロントエンドアプリが呼び出したかを確認
7 Object ID (oid) ユーザーの一意識別子が存在するか確認
8 Scope (scp) access_as_user スコープが含まれているか確認

フェーズ3: On-Behalf-Of (OBO) トークン交換

フェーズ3では、Backend API (FastMCP) が ユーザーに代わって ダウンストリーム API (Azure AI Search, Azure サービスなど) にアクセスするためのトークンを取得します。これを On-Behalf-Of (OBO) フロー と呼びます。

なぜ OBO が必要なのか？

フェーズ2で取得した委任トークンをダウンストリーム API にそのまま使用できない理由:

// フェーズ2の委任トークン (Backend API 用)
{
  "aud": "api://12345678-1234-1234-1234-123456789abc",  // Backend API
  "scp": "access_as_user"
}

// Azure AI Search が期待するトークン
{
  "aud": "https://search.azure.com",  // Azure AI Search
  "scp": "user_impersonation"
}

委任トークンの aud は Backend API を指しているため、Azure AI Search に送っても Audience 不一致で拒否されます。

推奨事項

OBO フローについて以下を推奨(参考):

  • Confidential Client として登録: Backend はクライアントシークレットまたは証明書で自身を証明
  • User Assertion の提示: ユーザーの委任トークンを assertion パラメーターで送信
  • grant_type の指定: urn:ietf:params:oauth:grant-type:jwt-bearer を使用
  • requested_token_use の指定: on_behalf_of を設定
  • Managed Identity の活用: 本番環境ではクライアントシークレットを避け、Federated Credential を使用

フロー詳細図

実装のポイント

OBO トークン交換の実装

例として Azure AI Search にアクセスする想定。

from azure.identity import ManagedIdentityCredential
import msal

@mcp.tool()
async def get_azure_ai_search_token(ctx: Context) -> dict:
    """
    Azure AI Search用のOBOトークンを取得
    """
    # リクエストから委任トークンを取得
    request = ctx.request_context.request
    auth_header = request.headers.get("Authorization", "")
    user_token = auth_header.split(" ", 1)[1]
    
    # OBO トークン交換を実行
    token_exchanger = TokenOboExchanger()
    search_token = await token_exchanger.perform_obo_token_exchange(
        user_token=user_token,
        resource_uri="https://search.azure.com"
    )
    
    return {
        "ok": True,
        "access_token": search_token,
        "usage": "Use this token in Authorization header: Bearer <token>"
    }

TokenOboExchanger の内部実装

MCP server for Fabric Real-Time Intelligence（fabric-rti-mcp）の実装から借用。

class TokenOboExchanger:
    async def perform_obo_token_exchange(self, user_token: str, resource_uri: str) -> str:
        """
        OBO フローでダウンストリーム API用トークンを取得
        """
        # 1. 認証方法の分岐
        if self.entra_app_client_secret:
            # ローカル開発: クライアントシークレット
            client_credential = self.entra_app_client_secret
        else:
            # 本番環境: Managed Identity + Federated Credential
            managed_identity_credential = ManagedIdentityCredential(
                client_id=self.umi_client_id
            )
            assertion_token = managed_identity_credential.get_token(
                "api://AzureADTokenExchange/.default"
            ).token
            client_credential = {
                "client_assertion": assertion_token,
                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
            }
        
        # 2. MSAL で OBO トークンを取得
        app = msal.ConfidentialClientApplication(
            client_id=self.entra_app_client_id,
            authority=f"https://login.microsoftonline.com/{self.tenant_id}",
            client_credential=client_credential,
        )
        
        result = app.acquire_token_on_behalf_of(
            user_assertion=user_token,
            scopes=[f"{resource_uri}/.default"]
        )
        
        if "access_token" not in result:
            raise Exception(f"OBO token exchange failed: {result.get('error_description')}")
        
        return result["access_token"]

取得される OBO トークンの詳細

{
  "aud": "https://search.azure.com",
  "iss": "https://sts.windows.net/{TENANT_ID}/",
  "iat": 1699000000,
  "nbf": 1699000000,
  "exp": 1699003600,
  "scp": "user_impersonation",
  "oid": "{USER_OBJECT_ID}",
  "upn": "user@contoso.com",
  "name": "田中 太郎",
  "appid": "{ENTRA_APP_CLIENT_ID}",
  "tid": "{TENANT_ID}",
  "ver": "2.0"
}

重要なポイント:

  • aud: Azure AI Search に変更 (https://search.azure.com)
  • oid, upn: 元のユーザー情報を保持 (監査証跡)
  • appid: Backend App が代理でアクセスしていることを示す
  • scp: Azure AI Search のスコープ (user_impersonation)

image.png

UI では「AI Search トークンを取得」ボタンで OBO トークンを発行できる。

まとめ

本記事では、Microsoft Entra ID を使った認証アーキテクチャを Full Python で実装する方法を、3つの認証フェーズに分けて紹介しました。

フェーズ1: ユーザー認証

  • Authorization Code Flow with PKCE により、セキュアなユーザー認証を実現
  • Identity for Flask ライブラリを活用し、複雑な OAuth 2.0 フローを隠蔽

フェーズ2: ユーザー委任のアクセストークン

  • Backend API を MCP Tool として公開し、　Microsoft Agent Framework からアクセス
  • Backend API 専用の Delegated Permissions を定義し、用途ごとにトークンを分離
  • FastMCP JWTVerifier による厳密なトークン検証 (署名、Audience、Issuer、スコープ、有効期限)
  • セッションキャッシュと Refresh Token による効率的なトークン管理

フェーズ3: On-Behalf-Of トークン交換

  • ユーザーのコンテキストを保持したまま、ダウンストリーム API へのアクセスを実現
  • Managed Identity + Federated Credential による本番環境でのシークレットレス認証（オプション）

GitHub

2
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?