理解 MCP 中的授权

"""MCP 认证服务器的配置设置。"""

import os


class Config:
    """从环境变量加载并提供合理默认值的配置类。"""

    # 服务器设置
    HOST: str = os.getenv("HOST", "localhost")
    PORT: int = int(os.getenv("PORT", "3000"))

    # 认证服务器设置
    AUTH_HOST: str = os.getenv("AUTH_HOST", "localhost")
    AUTH_PORT: int = int(os.getenv("AUTH_PORT", "8080"))
    AUTH_REALM: str = os.getenv("AUTH_REALM", "master")

    # OAuth 客户端设置
    OAUTH_CLIENT_ID: str = os.getenv("OAUTH_CLIENT_ID", "test-client")
    OAUTH_CLIENT_SECRET: str = os.getenv("OAUTH_CLIENT_SECRET", "")

    # 每个令牌都需要的作用域
    MCP_SCOPE: str = os.getenv("MCP_SCOPE", "mcp:tools")

    @property
    def server_url(self) -> str:
        """构建服务器 URL。"""
        return f"http://{self.HOST}:{self.PORT}"

    @property
    def auth_base_url(self) -> str:
        """构建认证服务器基础 URL。"""
        return f"http://{self.AUTH_HOST}:{self.AUTH_PORT}/realms/{self.AUTH_REALM}/"


# 全局配置实例
config = Config()

import datetime
import logging
from typing import Any
from urllib.parse import urljoin

from pydantic import AnyHttpUrl

from mcp.server import MCPServer
from mcp.server.auth.settings import AuthSettings

from .config import config
from .token_verifier import IntrospectionTokenVerifier

logger = logging.getLogger(__name__)


def create_oauth_urls() -> dict[str, str]:
    """根据配置创建 OAuth URL（Keycloak 风格）。"""
    auth_base_url = config.auth_base_url

    return {
        "issuer": auth_base_url,
        "introspection_endpoint": urljoin(auth_base_url, "protocol/openid-connect/token/introspect"),
        "authorization_endpoint": urljoin(auth_base_url, "protocol/openid-connect/auth"),
        "token_endpoint": urljoin(auth_base_url, "protocol/openid-connect/token"),
    }


def create_server() -> MCPServer:
    """创建并配置 MCP 服务器。"""

    oauth_urls = create_oauth_urls()

    token_verifier = IntrospectionTokenVerifier(
        introspection_endpoint=oauth_urls["introspection_endpoint"],
        server_url=config.server_url,
        client_id=config.OAUTH_CLIENT_ID,
        client_secret=config.OAUTH_CLIENT_SECRET,
    )

    app = MCPServer(
        name="MCP Resource Server",
        instructions="通过授权服务器自省验证令牌的资源服务器",
        debug=True,
        token_verifier=token_verifier,
        auth=AuthSettings(
            issuer_url=AnyHttpUrl(oauth_urls["issuer"]),
            required_scopes=[config.MCP_SCOPE],
            resource_server_url=AnyHttpUrl(config.server_url),
        ),
    )

    @app.tool()
    async def add_numbers(a: float, b: float) -> dict[str, Any]:
        """
        将两个数字相加。
        此工具演示了使用 OAuth 认证的基本算术运算。

        Args:
            a: 要相加的第一个数字
            b: 要相加的第二个数字
        """
        result = a + b
        return {
            "operation": "addition",
            "operand_a": a,
            "operand_b": b,
            "result": result,
            "timestamp": datetime.datetime.now().isoformat(),
        }

    @app.tool()
    async def multiply_numbers(x: float, y: float) -> dict[str, Any]:
        """
        将两个数字相乘。
        此工具演示了使用 OAuth 认证的基本算术运算。

        Args:
            x: 要相乘的第一个数字
            y: 要相乘的第二个数字
        """
        result = x * y
        return {
            "operation": "multiplication",
            "operand_x": x,
            "operand_y": y,
            "result": result,
            "timestamp": datetime.datetime.now().isoformat(),
        }

    return app


def main() -> int:
    """
    运行 MCP Resource Server。

    此服务器：
    - 提供 RFC 9728 受保护资源元数据
    - 通过授权服务器自省验证令牌
    - 提供需要认证的 MCP 工具

    配置从 config.py 和环境变量中加载。
    """
    logging.basicConfig(level=logging.INFO)

    oauth_urls = create_oauth_urls()

    try:
        mcp_server = create_server()

        logger.info("正在 %s:%s 上启动 MCP 服务器", config.HOST, config.PORT)
        logger.info("授权服务器：%s", oauth_urls["issuer"])

        mcp_server.run(
            transport="streamable-http",
            host=config.HOST,
            port=config.PORT,
            streamable_http_path="/",
        )
        return 0

    except Exception:
        logger.exception("服务器错误")
        return 1


if __name__ == "__main__":
    exit(main())

"""使用 OAuth 2.0 令牌自省（RFC 7662）的令牌验证器实现。"""

import logging
from typing import Any

import httpx2

from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.shared.auth_utils import check_resource_allowed, resource_url_from_server_url

logger = logging.getLogger(__name__)


class IntrospectionTokenVerifier(TokenVerifier):
    """使用 OAuth 2.0 令牌自省（RFC 7662）的令牌验证器。"""

    def __init__(
        self,
        introspection_endpoint: str,
        server_url: str,
        client_id: str,
        client_secret: str,
    ):
        self.introspection_endpoint = introspection_endpoint
        self.server_url = server_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.resource_url = resource_url_from_server_url(server_url)

    async def verify_token(self, token: str) -> AccessToken | None:
        """通过自省端点验证令牌。"""
        if not self.introspection_endpoint.startswith(("https://", "http://localhost", "http://127.0.0.1")):
            return None

        timeout = httpx2.Timeout(10.0, connect=5.0)
        limits = httpx2.Limits(max_connections=10, max_keepalive_connections=5)

        async with httpx2.AsyncClient(
            timeout=timeout,
            limits=limits,
            verify=True,
        ) as client:
            try:
                form_data = {
                    "token": token,
                    "client_id": self.client_id,
                }
                # 仅在配置了 client_secret 时发送 client_secret
                # 公共客户端仅使用 client_id 进行认证。
                if self.client_secret:
                    form_data["client_secret"] = self.client_secret
                headers = {"Content-Type": "application/x-www-form-urlencoded"}

                response = await client.post(
                    self.introspection_endpoint,
                    data=form_data,
                    headers=headers,
                )

                if response.status_code != 200:
                    return None

                data = response.json()
                if not data.get("active", False):
                    return None

                if not self._validate_resource(data):
                    return None

                return AccessToken(
                    token=token,
                    client_id=data.get("client_id", "unknown"),
                    scopes=data.get("scope", "").split() if data.get("scope") else [],
                    expires_at=data.get("exp"),
                    # AccessToken.resource 是 `str | None`。Keycloak 这里返回的 `aud`
                    # 是一个 *列表*（例如 ["test-client", "http://localhost:3000",
                    # "account"]）；如果直接传入该列表，会触发 pydantic
                    # ValidationError，并被下面的宽泛 `except` 转换成一个
                    # 静默的 401。我们已经在 `_validate_resource` 中确认了
                    # 这个服务器的资源是合法受众，因此在这里记录它。
                    resource=self.resource_url,
                    subject=data.get("sub"),  # RFC 7662 subject（资源所有者）
                    claims=data,
                )

            except Exception:
                logger.exception("令牌自省失败")
                return None

    def _validate_resource(self, token_data: dict[str, Any]) -> bool:
        """验证令牌是否为此资源服务器签发。

        规则：
        - 如果缺少 'aud'，则拒绝。
        - 如果任一受众条目匹配派生出的资源 URL，则接受。
        - 根据 JWT 规范，支持字符串或列表形式。
        """
        if not self.server_url or not self.resource_url:
            return False

        aud: list[str] | str | None = token_data.get("aud")
        if isinstance(aud, list):
            return any(self._is_valid_resource(a) for a in aud)
        if isinstance(aud, str):
            return self._is_valid_resource(aud)
        return False

    def _is_valid_resource(self, resource: str) -> bool:
        """检查给定资源是否与我们的服务器匹配。"""
        return check_resource_allowed(requested_resource=self.resource_url, configured_resource=resource)

[project]
name = "mcp-simple-auth"
version = "0.1.0"
description = "一个演示 OAuth 认证的简单 MCP 服务器"
requires-python = ">=3.10"
authors = [{ name = "Model Context Protocol a Series of LF Projects, LLC." }]
license = { text = "MIT" }
dependencies = [
  "httpx2>=2.5.0",
  "mcp>=2.0.0rc1",
  "pydantic>=2.0",
]

[project.scripts]
mcp-simple-auth-rs = "mcp_server.server:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["mcp_server"]

[dependency-groups]
dev = ["pyright>=1.1.391", "pytest>=8.3.4", "ruff>=0.8.5"]

uv sync
uv run mcp-simple-auth-rs

using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.IdentityModel.Tokens;
using ModelContextProtocol.AspNetCore.Authentication;
using ProtectedMcpServer.Tools;
using System.Security.Claims;

var builder = WebApplication.CreateBuilder(args);

var serverUrl = "http://localhost:3000/";
var authorizationServerUrl = "http://localhost:8080/realms/master/";

builder.Services.AddAuthentication(options =>
{
    options.DefaultChallengeScheme = McpAuthenticationDefaults.AuthenticationScheme;
    options.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;
})
.AddJwtBearer(options =>
{
    options.Authority = authorizationServerUrl;
    var normalizedServerAudience = serverUrl.TrimEnd('/');
    options.TokenValidationParameters = new TokenValidationParameters
    {
        ValidIssuer = authorizationServerUrl,
        ValidAudiences = new[] { normalizedServerAudience, serverUrl },
        AudienceValidator = (audiences, securityToken, validationParameters) =>
        {
            if (audiences == null) return false;
            foreach (var aud in audiences)
            {
                if (string.Equals(aud.TrimEnd('/'), normalizedServerAudience, StringComparison.OrdinalIgnoreCase))
                {
                    return true;
                }
            }
            return false;
        }
    };

    options.RequireHttpsMetadata = false; // 在生产环境中请设为 true

    options.Events = new JwtBearerEvents
    {
        OnTokenValidated = context =>
        {
            var name = context.Principal?.Identity?.Name ?? "unknown";
            var email = context.Principal?.FindFirstValue("preferred_username") ?? "unknown";
            Console.WriteLine($"令牌已验证：{name} ({email})");
            return Task.CompletedTask;
        },
        OnAuthenticationFailed = context =>
        {
            Console.WriteLine($"认证失败：{context.Exception.Message}");
            return Task.CompletedTask;
        },
    };
})
.AddMcp(options =>
{
    options.ResourceMetadata = new()
    {
        Resource = new Uri(serverUrl),
        ResourceDocumentation = new Uri("https://docs.example.com/api/math"),
        AuthorizationServers = { new Uri(authorizationServerUrl) },
        ScopesSupported = ["mcp:tools"]
    };
});

builder.Services.AddAuthorization();

builder.Services.AddHttpContextAccessor();
builder.Services.AddMcpServer()
    .WithTools<MathTools>()
    .WithHttpTransport();

var app = builder.Build();

app.UseAuthentication();
app.UseAuthorization();

app.MapMcp().RequireAuthorization();

Console.WriteLine($"正在 {serverUrl} 启动带授权的 MCP 服务器");
Console.WriteLine($"使用的 Keycloak 服务器：{authorizationServerUrl}");
Console.WriteLine($"受保护资源元数据 URL：{serverUrl}.well-known/oauth-protected-resource");
Console.WriteLine("公开的数学工具：Add、Multiply");
Console.WriteLine("按 Ctrl+C 停止服务器");

app.Run(serverUrl);