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

winx_code_agent/
http_server.rs
