futu_mcp/

http.rs

use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context, Result};
use futu_auth::KeyStore;

use crate::tools;

/// HTTP 模式：axum + rmcp StreamableHttpService，`/mcp` 路径跑 MCP，
/// `/metrics` 暴露 Prometheus counters（无需 token），
/// `/.well-known/oauth-protected-resource` 暴露 OAuth2 Protected Resource
/// Metadata（RFC 9728，给 MCP 客户端发现鉴权要求用）。
///
/// v1.4+：未带 Bearer token 的 `/mcp` 请求会回 `401 + WWW-Authenticate`
/// 头，指向 resource metadata，配合 rmcp 客户端的自动发现流程。
pub(super) fn render_mcp_metrics_body() -> String {
    let registry = futu_auth::metrics::global();
    render_mcp_metrics_body_for(registry.as_deref())
}

pub(super) fn render_mcp_metrics_body_for(registry: Option<&futu_auth::MetricsRegistry>) -> String {
    registry.map(|r| r.render_prometheus()).unwrap_or_else(|| {
        concat!(
            "# HELP futu_metrics_registry_installed Whether futu_auth metrics registry is installed (1=yes, 0=no)\n",
            "# TYPE futu_metrics_registry_installed gauge\n",
            "futu_metrics_registry_installed{state=\"metrics registry not installed\"} 0\n"
        )
        .to_string()
    })
}

pub(super) async fn serve_http(
    server: tools::FutuServer,
    key_store: Arc<KeyStore>,
    listen: &str,
    tls: Option<(PathBuf, PathBuf)>,
) -> Result<()> {
    use rmcp::transport::streamable_http_server::{
        StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
    };

    // 补齐 `:port` 写法：bind 到 0.0.0.0:port
    let bind_addr = if listen.starts_with(':') {
        format!("0.0.0.0{listen}")
    } else {
        listen.to_string()
    };

    // rmcp StreamableHttpService 要求一个 service factory —— 每个 HTTP 会话要
    // 一份独立的 MCP Service 实例。ServerState 跨会话共享，但 legacy
    // logging/setLevel 阈值属于会话，必须由 FutuServer::new 逐会话创建。
    let session_manager = std::sync::Arc::new(LocalSessionManager::default());
    // One endpoint serves both lifecycles. Legacy versions retain their
    // initialize-created sessions, while rmcp always routes 2026-07-28
    // requests statelessly (SEP-2567). Do not enable the global strict
    // metadata switch: rmcp 1.x/2.x clients do not attach modern per-request
    // metadata. Default loopback Host validation remains active.
    let transport_config = StreamableHttpServerConfig::default()
        .with_legacy_session_mode(true)
        .with_stateless_protocol_metadata_required(false);
    let mcp_svc = StreamableHttpService::new(
        {
            let state = server.state.clone();
            move || Ok::<_, std::io::Error>(crate::tools::FutuServer::new(state.clone()))
        },
        session_manager,
        transport_config,
    );

    // axum 0.8 router：
    // - /mcp        → MCP tower service（带 WWW-Authenticate 401 middleware）
    // - /metrics    → Prometheus 文本
    // - /.well-known/oauth-protected-resource → RFC 9728 元数据
    use axum::routing::get;
    let mcp_with_auth_hint = axum::Router::new()
        .nest_service("/mcp", mcp_svc)
        .layer(axum::middleware::from_fn_with_state(
            key_store,
            authenticate_mcp_transport,
        ))
        .layer(axum::middleware::from_fn(inject_www_authenticate));

    let app = axum::Router::new()
        .route(
            "/metrics",
            get(|| async {
                let body = render_mcp_metrics_body();
                (
                    axum::http::StatusCode::OK,
                    [(
                        axum::http::header::CONTENT_TYPE,
                        "text/plain; version=0.0.4",
                    )],
                    body,
                )
            }),
        )
        .route(
            "/.well-known/oauth-protected-resource",
            get(oauth_protected_resource_metadata),
        )
        .merge(mcp_with_auth_hint);

    let bind_addr_sock: std::net::SocketAddr = bind_addr
        .parse()
        .map_err(|e| anyhow::anyhow!("invalid bind address {bind_addr}: {e}"))?;

    if let Some((cert_path, key_path)) = tls {
        // ---------- HTTPS（graceful shutdown 通过 Handle）----------
        let tls_config =
            axum_server::tls_rustls::RustlsConfig::from_pem_file(&cert_path, &key_path)
                .await
                .with_context(|| {
                    format!(
                        "load TLS cert={} key={}",
                        cert_path.display(),
                        key_path.display()
                    )
                })?;
        let handle = axum_server::Handle::new();
        let shutdown_handle = handle.clone();
        tokio::spawn(async move {
            shutdown_signal().await;
            tracing::info!("graceful shutdown: draining HTTPS connections...");
            shutdown_handle.graceful_shutdown(Some(std::time::Duration::from_secs(10)));
        });
        tracing::info!(
            addr = %bind_addr,
            cert = %cert_path.display(),
            "futu-mcp HTTPS transport started \
             (MCP: /mcp, metrics: /metrics, OAuth metadata: /.well-known/oauth-protected-resource)"
        );
        axum_server::bind_rustls(bind_addr_sock, tls_config)
            .handle(handle)
            .serve(app.into_make_service())
            .await
            .map_err(|e| anyhow::anyhow!("axum-server TLS serve error: {e}"))?;
    } else {
        // ---------- plain HTTP（graceful shutdown 通过 axum::serve）----------
        let listener = tokio::net::TcpListener::bind(&bind_addr)
            .await
            .map_err(|e| anyhow::anyhow!("bind {bind_addr}: {e}"))?;
        tracing::info!(
            addr = %bind_addr,
            "futu-mcp HTTP transport started \
             (MCP: /mcp, metrics: /metrics, OAuth metadata: /.well-known/oauth-protected-resource)"
        );
        axum::serve(listener, app)
            .with_graceful_shutdown(async {
                shutdown_signal().await;
                tracing::info!("graceful shutdown: draining HTTP connections...");
            })
            .await
            .map_err(|e| anyhow::anyhow!("axum serve error: {e}"))?;
    }
    tracing::info!("server stopped");
    Ok(())
}

/// Scope-mode transport authentication boundary for every `/mcp` request.
///
/// This runs outside rmcp's [`StreamableHttpService`], so a rejected request
/// cannot create, inspect, stream, or close a session. Tool-level scope,
/// account, and quota checks remain inside the MCP handlers.
pub(super) async fn authenticate_mcp_transport(
    axum::extract::State(key_store): axum::extract::State<Arc<KeyStore>>,
    req: axum::extract::Request,
    next: axum::middleware::Next,
) -> axum::response::Response {
    if !key_store.is_configured() {
        return next.run(req).await;
    }

    let mut values = req
        .headers()
        .get_all(axum::http::header::AUTHORIZATION)
        .iter();
    let Some(value) = values.next() else {
        return mcp_transport_unauthorized();
    };
    if values.next().is_some() {
        return mcp_transport_unauthorized();
    }
    let Some(token) = value
        .to_str()
        .ok()
        .and_then(futu_auth_pipeline::parse_bearer_scheme)
    else {
        return mcp_transport_unauthorized();
    };
    if key_store.verify(token).is_none() {
        return mcp_transport_unauthorized();
    }

    next.run(req).await
}

fn mcp_transport_unauthorized() -> axum::response::Response {
    use axum::response::IntoResponse;

    (
        axum::http::StatusCode::UNAUTHORIZED,
        [(
            axum::http::header::WWW_AUTHENTICATE,
            "Bearer resource_metadata=\"/.well-known/oauth-protected-resource\"",
        )],
    )
        .into_response()
}

/// 监听 SIGTERM / SIGINT，任一到达即返回。
/// 同时兼容 Windows（只有 ctrl_c）和 Unix（SIGTERM + SIGINT）。
async fn shutdown_signal() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{SignalKind, signal};
        let sigterm = match signal(SignalKind::terminate()) {
            Ok(signal) => Some(signal),
            Err(e) => {
                tracing::error!(error = %e, "failed to install SIGTERM handler");
                None
            }
        };
        let sigint = match signal(SignalKind::interrupt()) {
            Ok(signal) => Some(signal),
            Err(e) => {
                tracing::error!(error = %e, "failed to install SIGINT handler");
                None
            }
        };

        match (sigterm, sigint) {
            (Some(mut sigterm), Some(mut sigint)) => {
                tokio::select! {
                    _ = sigterm.recv() => tracing::info!("received SIGTERM"),
                    _ = sigint.recv()  => tracing::info!("received SIGINT"),
                }
            }
            (Some(mut sigterm), None) => {
                sigterm.recv().await;
                tracing::info!("received SIGTERM");
            }
            (None, Some(mut sigint)) => {
                sigint.recv().await;
                tracing::info!("received SIGINT");
            }
            (None, None) => wait_for_ctrl_c_or_pending().await,
        }
    }
    #[cfg(not(unix))]
    {
        wait_for_ctrl_c_or_pending().await;
    }
}

async fn wait_for_ctrl_c_or_pending() {
    match tokio::signal::ctrl_c().await {
        Ok(()) => tracing::info!("received Ctrl-C"),
        Err(e) => {
            tracing::error!(
                error = %e,
                "failed to install ctrl-c handler; graceful shutdown signal unavailable"
            );
            std::future::pending::<()>().await;
        }
    }
}

/// RFC 9728 — OAuth 2.0 Protected Resource Metadata
///
/// 我们不是完整 OAuth 授权服务器（那需要独立的 IdP），只是告诉 MCP 客户端
/// "这个资源要 Bearer token，scope 列表如下"。LLM agent 实际拿到 key 的方式
/// 仍然是运维线下发放 + 写入 MCP client 配置里的 `Authorization` 头。
///
/// 客户端可以 GET `/.well-known/oauth-protected-resource` 来发现：
///   - `resource`:              本 MCP endpoint URI
///   - `bearer_methods_supported`: 我们只支持 `header`（Authorization: Bearer ...）
///   - `scopes_supported`:      可声明的 futu-auth scope 列表
///   - `resource_name` / `resource_documentation`: 给人看的说明
pub(super) async fn oauth_protected_resource_metadata() -> axum::response::Json<serde_json::Value> {
    axum::response::Json(serde_json::json!({
        "resource": "/mcp",
        "bearer_methods_supported": ["header"],
        "scopes_supported": [
            "qot:read",
            "acc:read",
            "trade:simulate",
            "trade:real",
            "trade:unlock"
        ],
        "resource_name": "FutuOpenD-rs MCP",
        "resource_documentation": "https://futuapi.com/reference/mcp/",
    }))
}

/// Tower middleware: 如果下游（MCP service）返回 401/403 又没 `WWW-Authenticate`
/// 头，补一个 `Bearer resource_metadata="..."`，指向 `/.well-known/oauth-protected-resource`。
///
/// 符合 RFC 9728 §5.1：资源服务器应通过 WWW-Authenticate 宣告 metadata URL，
/// 让未配置的客户端能自动发现 scope 和鉴权方式。
pub(super) async fn inject_www_authenticate(
    req: axum::extract::Request,
    next: axum::middleware::Next,
) -> axum::response::Response {
    let mut resp = next.run(req).await;
    let status = resp.status();
    if (status == axum::http::StatusCode::UNAUTHORIZED
        || status == axum::http::StatusCode::FORBIDDEN)
        && !resp
            .headers()
            .contains_key(axum::http::header::WWW_AUTHENTICATE)
    {
        // 相对路径 —— 客户端按 Host header 拼全 URL；也避免 TLS 终止在前置
        // 反代（Caddy / Nginx）时我们误把内网地址写进响应头
        let value = axum::http::HeaderValue::from_static(
            "Bearer resource_metadata=\"/.well-known/oauth-protected-resource\"",
        );
        resp.headers_mut()
            .insert(axum::http::header::WWW_AUTHENTICATE, value);
    }
    resp
}