winx_code_agent/
http_server.rs

1//! Optional remote MCP transport over Streamable HTTP.
2//!
3//! The endpoint exposes shell and filesystem capabilities, so remote mode is
4//! deliberately fail-closed: loopback bind by default, strong file-backed or
5//! explicit bearer credentials, bounded request concurrency/rate, and isolated
6//! daemon namespaces for independently authenticated principals.
7
8#![allow(clippy::doc_markdown)]
9
10use std::collections::{HashMap, HashSet};
11use std::net::{IpAddr, SocketAddr};
12use std::sync::Arc;
13use std::time::{Duration, Instant};
14
15use axum::{
16    extract::{ConnectInfo, Request, State},
17    http::{header, HeaderValue, StatusCode},
18    middleware::{self, Next},
19    response::{IntoResponse, Response},
20    Router,
21};
22use rand::RngExt as _;
23use rmcp::transport::streamable_http_server::{
24    session::local::LocalSessionManager, StreamableHttpServerConfig, StreamableHttpService,
25};
26use tokio::sync::{Mutex, Semaphore};
27use tower_http::limit::RequestBodyLimitLayer;
28use tower_http::timeout::TimeoutLayer;
29
30use crate::config::{HttpPrincipal, HttpSessionAffinity};
31use crate::runtime::{configured_shell_runtime, ShellRuntime};
32use crate::server::{SessionIsolation, WinxService};
33
34pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
35
36const MAX_BODY_BYTES: usize = 64 * 1024 * 1024;
37const REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
38const RATE_WINDOW: Duration = Duration::from_secs(60);
39const MAX_RATE_TRACKED_IPS: usize = 4_096;
40const INVALID_AUTH_DELAY: Duration = Duration::from_millis(100);
41pub const DEFAULT_MAX_CONCURRENCY: usize = 32;
42pub const DEFAULT_REQUESTS_PER_MINUTE: u32 = 120;
43
44/// Opaque per-request correlation shared by the HTTP middleware and MCP handler.
45/// It contains no client input and is safe to include in metadata-only usage logs.
46#[derive(Clone, Debug)]
47pub(crate) struct RequestCorrelation(String);
48
49impl RequestCorrelation {
50    fn new() -> Self {
51        Self(format!("r_{:032x}", rand::rng().random::<u128>()))
52    }
53
54    pub(crate) fn as_str(&self) -> &str {
55        &self.0
56    }
57}
58
59#[derive(Clone, Debug)]
60pub struct HttpServerOptions {
61    pub bind: String,
62    pub principals: Vec<HttpPrincipal>,
63    pub extra_hosts: Vec<String>,
64    pub allow_query_token: bool,
65    pub allow_non_loopback: bool,
66    pub session_affinity: HttpSessionAffinity,
67    pub max_concurrency: usize,
68    pub requests_per_minute: u32,
69}
70
71impl HttpServerOptions {
72    pub fn single(
73        bind: impl Into<String>,
74        token: impl Into<String>,
75        extra_hosts: Vec<String>,
76        allow_query_token: bool,
77    ) -> Result<Self, BoxError> {
78        let token = token.into();
79        let principal = HttpPrincipal::new("default", token.trim(), false)?;
80        Ok(Self {
81            bind: bind.into(),
82            principals: vec![principal],
83            extra_hosts,
84            allow_query_token,
85            allow_non_loopback: false,
86            session_affinity: HttpSessionAffinity::Workspace,
87            max_concurrency: DEFAULT_MAX_CONCURRENCY,
88            requests_per_minute: DEFAULT_REQUESTS_PER_MINUTE,
89        })
90    }
91}
92
93pub async fn start_http_server(
94    bind: &str,
95    token: String,
96    extra_hosts: Vec<String>,
97    allow_query_token: bool,
98) -> Result<(), BoxError> {
99    let options = HttpServerOptions::single(bind, token, extra_hosts, allow_query_token)?;
100    start_http_server_with_options(options).await
101}
102
103pub async fn start_http_server_with_options(options: HttpServerOptions) -> Result<(), BoxError> {
104    let shell_runtime = configured_shell_runtime().await?;
105    start_http_server_with_runtime(options, shell_runtime).await
106}
107
108pub async fn start_http_server_with_runtime(
109    options: HttpServerOptions,
110    shell_runtime: Arc<dyn ShellRuntime>,
111) -> Result<(), BoxError> {
112    let bind = validate_options(&options)?;
113    let principal_names =
114        options.principals.iter().map(|principal| principal.name().to_string()).collect::<Vec<_>>();
115    let shared_concurrency = Arc::new(Semaphore::new(options.max_concurrency));
116    let shared_rate_limiter = Arc::new(RateLimiter::new(options.requests_per_minute));
117
118    let mut config = StreamableHttpServerConfig::default();
119    config.legacy_session_mode = true;
120    config.allowed_hosts.extend(options.extra_hosts);
121
122    let shared = WinxService::with_runtime(SessionIsolation::Strict, shell_runtime);
123    let mcp_service = StreamableHttpService::new(
124        move || Ok(shared.clone()),
125        Arc::new(LocalSessionManager::default()),
126        config,
127    );
128    let security = Arc::new(RouteSecurity {
129        principals: Arc::new(options.principals),
130        allow_query: options.allow_query_token,
131        session_affinity: options.session_affinity,
132        concurrency: shared_concurrency,
133        rate_limiter: shared_rate_limiter,
134    });
135    // The timeout sits INSIDE the security middleware: when it fires, the
136    // middleware still receives the 408 response and records the usage event.
137    // With the previous order the timed-out future was dropped before logging,
138    // so 408s were invisible in the usage log.
139    let protected_mcp = Router::new()
140        .nest_service("/mcp", mcp_service)
141        .layer(TimeoutLayer::with_status_code(StatusCode::REQUEST_TIMEOUT, REQUEST_TIMEOUT))
142        .layer(middleware::from_fn_with_state(security, enforce_route_security))
143        .layer(RequestBodyLimitLayer::new(MAX_BODY_BYTES));
144    // Keep OAuth discovery probes and every unrelated path outside bearer auth.
145    // Winx does not advertise OAuth metadata, so those routes should be ordinary
146    // 404s instead of 401s that encourage clients to start a nonexistent flow.
147    let app = Router::new().merge(protected_mcp);
148
149    let listener = tokio::net::TcpListener::bind(bind).await?;
150    tracing::warn!(
151        bind = %bind,
152        principals = ?principal_names,
153        route = "/mcp",
154        "Winx remote MCP is network-reachable; keep it behind authenticated HTTPS"
155    );
156    if options.allow_query_token {
157        tracing::warn!(
158            "query-token authentication is enabled; URLs may expose credentials in proxy logs/history"
159        );
160    }
161    axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await?;
162    Ok(())
163}
164
165fn validate_options(options: &HttpServerOptions) -> Result<SocketAddr, BoxError> {
166    let bind: SocketAddr = options
167        .bind
168        .parse()
169        .map_err(|error| format!("invalid --bind socket address {:?}: {error}", options.bind))?;
170    if !bind.ip().is_loopback() && !options.allow_non_loopback {
171        return Err(format!(
172            "refusing non-loopback HTTP bind {bind}; pass --allow-non-loopback only behind a trusted firewall/tunnel"
173        )
174        .into());
175    }
176    if options.max_concurrency == 0 {
177        return Err("max_concurrency must be at least 1".into());
178    }
179    if options.requests_per_minute == 0 {
180        return Err("requests_per_minute must be at least 1".into());
181    }
182    if options.principals.is_empty() {
183        return Err("at least one HTTP principal is required".into());
184    }
185    let mut names = HashSet::new();
186    for principal in &options.principals {
187        if !names.insert(principal.name()) {
188            return Err(format!("duplicate HTTP principal: {}", principal.name()).into());
189        }
190    }
191    Ok(bind)
192}
193
194#[derive(Clone)]
195struct RouteSecurity {
196    principals: Arc<Vec<HttpPrincipal>>,
197    allow_query: bool,
198    session_affinity: HttpSessionAffinity,
199    concurrency: Arc<Semaphore>,
200    rate_limiter: Arc<RateLimiter>,
201}
202
203async fn enforce_route_security(
204    State(security): State<Arc<RouteSecurity>>,
205    ConnectInfo(peer): ConnectInfo<SocketAddr>,
206    request: Request,
207    next: Next,
208) -> Response {
209    let path = request.uri().path();
210    if path != "/mcp" && !path.starts_with("/mcp/") {
211        return next.run(request).await;
212    }
213    let started = Instant::now();
214    if !security.rate_limiter.allow(peer.ip()).await {
215        return retry_response(StatusCode::TOO_MANY_REQUESTS, "request rate limit exceeded\n");
216    }
217    let Ok(permit) = security.concurrency.clone().try_acquire_owned() else {
218        return retry_response(StatusCode::SERVICE_UNAVAILABLE, "too many concurrent requests\n");
219    };
220
221    let Some(principal) =
222        authenticate_request(&request, &security.principals, security.allow_query)
223    else {
224        tokio::time::sleep(INVALID_AUTH_DELAY).await;
225        tracing::warn!(
226            peer = %peer,
227            path = %request.uri().path(),
228            "rejected request with missing or invalid HTTP token"
229        );
230        return (StatusCode::UNAUTHORIZED, "missing or invalid token\n").into_response();
231    };
232
233    let principal_name = principal.name().to_string();
234    let method = request.method().clone();
235    let correlation = RequestCorrelation::new();
236    let mut request = request;
237    request.extensions_mut().insert(principal);
238    request.extensions_mut().insert(security.session_affinity);
239    request.extensions_mut().insert(correlation.clone());
240    let response = next.run(request).await;
241    drop(permit);
242    // One structured `winx::usage` event per authenticated request. Metadata
243    // only — the body (commands, file contents) is never logged.
244    tracing::info!(
245        target: "winx::usage",
246        event = "http_request",
247        principal = %principal_name,
248        request_id = %correlation.as_str(),
249        peer = %peer,
250        method = %method,
251        status = response.status().as_u16(),
252        duration_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
253        "http request"
254    );
255    response
256}
257
258fn retry_response(status: StatusCode, body: &'static str) -> Response {
259    let mut response = (status, body).into_response();
260    response.headers_mut().insert(header::RETRY_AFTER, HeaderValue::from_static("1"));
261    response
262}
263
264#[derive(Debug)]
265struct RateWindow {
266    started_at: Instant,
267    last_seen: Instant,
268    count: u32,
269}
270
271#[derive(Debug)]
272struct RateLimiter {
273    requests_per_window: u32,
274    windows: Mutex<HashMap<IpAddr, RateWindow>>,
275}
276
277impl RateLimiter {
278    fn new(requests_per_window: u32) -> Self {
279        Self { requests_per_window, windows: Mutex::new(HashMap::new()) }
280    }
281
282    async fn allow(&self, peer: IpAddr) -> bool {
283        let now = Instant::now();
284        let mut windows = self.windows.lock().await;
285        windows.retain(|_, window| now.duration_since(window.last_seen) < RATE_WINDOW * 2);
286        if !windows.contains_key(&peer) && windows.len() >= MAX_RATE_TRACKED_IPS {
287            if let Some(oldest) =
288                windows.iter().min_by_key(|(_, window)| window.last_seen).map(|(ip, _)| *ip)
289            {
290                windows.remove(&oldest);
291            }
292        }
293        let window =
294            windows.entry(peer).or_insert(RateWindow { started_at: now, last_seen: now, count: 0 });
295        if now.duration_since(window.started_at) >= RATE_WINDOW {
296            window.started_at = now;
297            window.count = 0;
298        }
299        window.last_seen = now;
300        if window.count >= self.requests_per_window {
301            return false;
302        }
303        window.count = window.count.saturating_add(1);
304        true
305    }
306}
307
308fn authenticate_request(
309    request: &Request,
310    principals: &[HttpPrincipal],
311    allow_query: bool,
312) -> Option<HttpPrincipal> {
313    let header = bearer_token(request);
314    let query = allow_query.then(|| query_token(request)).flatten();
315    [header, query].into_iter().flatten().find_map(|presented| {
316        principals.iter().find(|principal| principal.matches_token(presented)).cloned()
317    })
318}
319
320fn bearer_token(request: &Request) -> Option<&str> {
321    let value = request.headers().get(header::AUTHORIZATION)?.to_str().ok()?;
322    let (scheme, token) = value.split_once(' ')?;
323    scheme.eq_ignore_ascii_case("bearer").then_some(token.trim())
324}
325
326fn query_token(request: &Request) -> Option<&str> {
327    request.uri().query()?.split('&').find_map(|pair| pair.strip_prefix("token="))
328}
329
330#[cfg(test)]
331mod tests {
332    #![allow(clippy::expect_used)]
333
334    use std::net::{IpAddr, Ipv4Addr};
335
336    use axum::body::Body;
337    use axum::extract::Request;
338
339    use super::{authenticate_request, validate_options, HttpServerOptions, RateLimiter};
340    use crate::config::HttpPrincipal;
341
342    fn request(uri: &str, authorization: Option<&str>) -> Request {
343        let mut builder = Request::builder().uri(uri);
344        if let Some(authorization) = authorization {
345            builder = builder.header(axum::http::header::AUTHORIZATION, authorization);
346        }
347        builder.body(Body::empty()).expect("test request")
348    }
349
350    fn principal(name: &str, token_byte: char) -> HttpPrincipal {
351        let token = token_byte.to_string().repeat(32);
352        HttpPrincipal::new(name, &token, false).expect("valid principal")
353    }
354
355    fn strong_options(bind: &str) -> HttpServerOptions {
356        HttpServerOptions {
357            bind: bind.to_string(),
358            principals: vec![principal("default", 'x')],
359            extra_hosts: Vec::new(),
360            allow_query_token: false,
361            allow_non_loopback: false,
362            session_affinity: crate::config::HttpSessionAffinity::Workspace,
363            max_concurrency: super::DEFAULT_MAX_CONCURRENCY,
364            requests_per_minute: super::DEFAULT_REQUESTS_PER_MINUTE,
365        }
366    }
367
368    #[test]
369    fn bearer_and_opt_in_query_tokens_select_the_matching_principal() {
370        let principals = vec![principal("chatgpt", 'a'), principal("automation", 'b')];
371        let header = request("/mcp", Some(&format!("Bearer {}", "a".repeat(32))));
372        let matched = authenticate_request(&header, &principals, false).expect("header principal");
373        assert_eq!(matched.name(), "chatgpt");
374
375        let query_uri = format!("/mcp?token={}", "b".repeat(32));
376        assert!(authenticate_request(&request(&query_uri, None), &principals, false).is_none());
377        let matched = authenticate_request(&request(&query_uri, None), &principals, true)
378            .expect("query principal");
379        assert_eq!(matched.name(), "automation");
380    }
381
382    #[test]
383    fn non_loopback_binding_requires_explicit_override() {
384        assert!(validate_options(&strong_options("127.0.0.1:8000")).is_ok());
385        assert!(validate_options(&strong_options("0.0.0.0:8000")).is_err());
386
387        let mut public = strong_options("0.0.0.0:8000");
388        public.allow_non_loopback = true;
389        assert!(validate_options(&public).is_ok());
390    }
391
392    #[tokio::test]
393    async fn rate_limiter_rejects_excess_requests() {
394        let limiter = RateLimiter::new(2);
395        let peer = IpAddr::V4(Ipv4Addr::LOCALHOST);
396        assert!(limiter.allow(peer).await);
397        assert!(limiter.allow(peer).await);
398        assert!(!limiter.allow(peer).await);
399    }
400
401    #[test]
402    fn duplicate_principal_names_are_rejected() {
403        let mut options = strong_options("127.0.0.1:8000");
404        options.principals = vec![principal("same", 'a'), principal("same", 'b')];
405        assert!(validate_options(&options).is_err());
406    }
407}
winx_code_agent/http_server.rs

winx_code_agent/
http_server.rs
