use std::net::SocketAddr;
use std::sync::Arc;
use axum::Router;
use axum::extract::{ConnectInfo, Request, State};
use axum::http::{HeaderMap, StatusCode};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Json, Response};
use axum::routing::get;
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
use serde_json::json;
use tokio::sync::Mutex as TokioMutex;
use super::auth_extractor::extract_request_credentials;
use super::localhost_detector::is_localhost;
use super::metrics;
use super::types::HttpServerConfig;
use crate::core::component_registry::{ComponentRegistry, ComponentStatus};
use crate::core::mcp_server::McpifyServer;
#[derive(Clone)]
struct AppState {
registry: Arc<TokioMutex<ComponentRegistry>>,
header_location: &'static str,
header_name: String,
}
async fn healthz(State(state): State<AppState>) -> impl IntoResponse {
let registry = state.registry.lock().await;
let status = registry.overall_status();
let code = if status == ComponentStatus::Unhealthy {
StatusCode::SERVICE_UNAVAILABLE
} else {
StatusCode::OK
};
(
code,
Json(json!({
"status": format!("{status:?}"),
"components": registry.snapshot().len(),
})),
)
}
async fn metrics_handler() -> impl IntoResponse {
metrics::increment("http_requests_total");
(StatusCode::OK, metrics::render_metrics())
}
async fn auth_gate(
State(state): State<AppState>,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
request: Request,
next: Next,
) -> Response {
metrics::increment("http_requests_total");
let relaxed = is_localhost(addr.ip());
if !relaxed
&& extract_request_credentials(&headers, state.header_location, &state.header_name).is_err()
{
return (
StatusCode::UNAUTHORIZED,
Json(json!({ "error": "missing or invalid credentials for this API's auth method" })),
)
.into_response();
}
next.run(request).await
}
async fn set_cors_header(cors_allow: Option<String>, request: Request, next: Next) -> Response {
let mut response = next.run(request).await;
if let Some(origin) = cors_allow
&& let Ok(value) = origin.parse()
{
response
.headers_mut()
.insert("Access-Control-Allow-Origin", value);
}
response
}
pub async fn start_http_server(
mcp_service_factory: impl Fn() -> Result<McpifyServer, std::io::Error> + Send + Sync + 'static,
config: &HttpServerConfig,
registry: Arc<TokioMutex<ComponentRegistry>>,
header_location: &'static str,
header_name: String,
) -> anyhow::Result<()> {
let mut allowed_hosts = vec![
"localhost".to_string(),
"127.0.0.1".to_string(),
"::1".to_string(),
];
allowed_hosts.push(config.host.clone());
allowed_hosts.push(format!("{}:{}", config.host, config.port));
let mcp_service = StreamableHttpService::new(
mcp_service_factory,
LocalSessionManager::default().into(),
StreamableHttpServerConfig::default().with_allowed_hosts(allowed_hosts),
);
let state = AppState {
registry,
header_location,
header_name,
};
let cors_allow = config.cors_allow.clone();
let mcp_router = Router::new()
.nest_service("/mcp", mcp_service)
.layer(middleware::from_fn_with_state(state.clone(), auth_gate));
let app = Router::new()
.route("/healthz", get(healthz))
.route("/metrics", get(metrics_handler))
.merge(mcp_router)
.layer(middleware::from_fn(move |req, next| {
set_cors_header(cors_allow.clone(), req, next)
}))
.with_state(state);
let addr: SocketAddr = format!("{}:{}", config.host, config.port).parse()?;
let listener = tokio::net::TcpListener::bind(addr).await?;
axum::serve(
listener,
app.into_make_service_with_connect_info::<SocketAddr>(),
)
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use std::net::{IpAddr, Ipv4Addr};
use crate::auth::auth_manager::AuthManager;
use crate::core::config_schema::{AuthMethod, Config};
use axum::body::Body;
use axum::http::{Request, header};
use tower::ServiceExt;
use super::*;
fn state() -> AppState {
AppState {
registry: Arc::new(TokioMutex::new(ComponentRegistry::new())),
header_location: "header",
header_name: "Authorization".to_string(),
}
}
fn request_from(ip: IpAddr, authorization: Option<&str>) -> Request<Body> {
let mut request = Request::builder().uri("/").body(Body::empty()).unwrap();
request
.extensions_mut()
.insert(ConnectInfo(SocketAddr::new(ip, 12345)));
if let Some(value) = authorization {
request
.headers_mut()
.insert(header::AUTHORIZATION, value.parse().unwrap());
}
request
}
#[tokio::test]
async fn health_metrics_auth_gate_cors_and_server_bootstrap_are_exercised() {
let state = state();
assert_eq!(
healthz(State(state.clone())).await.into_response().status(),
StatusCode::OK
);
{
let mut registry = state.registry.lock().await;
registry.register("database", true);
registry.report(
"database",
ComponentStatus::Unhealthy,
Some("failed".to_string()),
);
}
assert_eq!(
healthz(State(state.clone())).await.into_response().status(),
StatusCode::SERVICE_UNAVAILABLE
);
assert_eq!(
metrics_handler().await.into_response().status(),
StatusCode::OK
);
let gated = Router::new()
.route("/", get(|| async { StatusCode::NO_CONTENT }))
.layer(middleware::from_fn_with_state(state.clone(), auth_gate));
let remote = IpAddr::V4(Ipv4Addr::new(203, 0, 113, 10));
assert_eq!(
gated
.clone()
.oneshot(request_from(remote, None))
.await
.unwrap()
.status(),
StatusCode::UNAUTHORIZED
);
assert_eq!(
gated
.clone()
.oneshot(request_from(remote, Some("Bearer coverage")))
.await
.unwrap()
.status(),
StatusCode::NO_CONTENT
);
assert_eq!(
gated
.oneshot(request_from(IpAddr::V4(Ipv4Addr::LOCALHOST), None))
.await
.unwrap()
.status(),
StatusCode::NO_CONTENT
);
let cors = Router::new()
.route("/", get(|| async { StatusCode::OK }))
.layer(middleware::from_fn(|request, next| {
set_cors_header(Some("https://client.example".to_string()), request, next)
}));
let cors_response = cors
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(
cors_response.headers()[header::ACCESS_CONTROL_ALLOW_ORIGIN],
"https://client.example"
);
let config: Config = serde_json::from_value(serde_json::json!({
"url": "https://api.example.test",
"auth_method": "pat"
}))
.unwrap();
let auth_manager = Arc::new(TokioMutex::new(AuthManager::new(AuthMethod::Pat)));
let server_config = HttpServerConfig {
host: "not a valid socket address".to_string(),
port: 3000,
cors_allow: Some("https://client.example".to_string()),
};
let result = start_http_server(
move || {
Ok(McpifyServer::new(
"gh-2026-03-10".to_string(),
config.clone(),
auth_manager.clone(),
))
},
&server_config,
state.registry,
"header",
"Authorization".to_string(),
)
.await;
assert!(result.is_err());
}
}