From f491d139357763a435a18ba4e06ac06e64e70ae6 Mon Sep 17 00:00:00 2001 From: Cody De Arkland Date: Fri, 24 Jul 2026 23:58:27 -0700 Subject: [PATCH 1/2] feat(mcp): proxy the remote MCP server over stdio using the CLI login Adds `railway mcp proxy`: a stdio bridge to mcp.railway.com that attaches a fresh Bearer token from the stored `railway login` on every request, so MCP-capable editors can use the remote server without the OAuth (DCR + browser consent) flow and without long-lived credentials in editor configs. - Token freshness delegates to ensure_valid_token (lockfile-serialized, so the proxy coexists with the local `railway mcp` server); a proxy started logged-out completes the handshake with a login hint and heals on the next request after `railway login`, no restart needed. - SSE responses are re-framed as newline-delimited JSON-RPC via a single writer task; concurrent tool calls after the handshake; transparent re-initialize + one retry on session loss; bounded 5s shutdown so a stalled upstream can't orphan the process; JSON-RPC batch error paths answer every id. - `railway mcp install` gains a three-way transport: --remote now installs the proxy, --remote --oauth keeps the plain-URL editor OAuth flow, and the config matchers distinguish local/proxy/oauth entries (a bare "args contains mcp" check would misdetect proxy installs as local). - Same choices plumbed through `railway setup agent`. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 16 + Cargo.toml | 1 + src/commands/mcp/install.rs | 464 ++++++++++++++++++++------- src/commands/mcp/mod.rs | 4 + src/commands/mcp/proxy.rs | 612 ++++++++++++++++++++++++++++++++++++ src/commands/setup.rs | 77 ++++- 6 files changed, 1051 insertions(+), 123 deletions(-) create mode 100644 src/commands/mcp/proxy.rs diff --git a/Cargo.lock b/Cargo.lock index 6fa53e8ba..879fefb0b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3775,6 +3775,7 @@ dependencies = [ "base64", "bytes 1.11.1", "futures-core", + "futures-util", "http 1.2.0", "http-body", "http-body-util", @@ -3795,12 +3796,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", "webpki-roots", ] @@ -5378,6 +5381,19 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmparser" version = "0.244.0" diff --git a/Cargo.toml b/Cargo.toml index bc0aa8a7a..da8da2ade 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,6 +31,7 @@ reqwest = { version = "0.12.28", default-features = false, features = [ "rustls-tls", "rustls-tls-native-roots", "json", + "stream", ] } reqwest-websocket = "0.5.1" regex = "1.12.3" diff --git a/src/commands/mcp/install.rs b/src/commands/mcp/install.rs index 60c0f57f3..3e64fbfd6 100644 --- a/src/commands/mcp/install.rs +++ b/src/commands/mcp/install.rs @@ -14,20 +14,67 @@ pub struct Args { #[clap(long)] agent: Vec, - /// Configure the remote HTTP MCP server at mcp.railway.com instead of the local stdio server. + /// Configure the remote MCP server at mcp.railway.com, bridged through `railway mcp proxy` + /// so it authenticates with your existing CLI login (no browser OAuth flow). #[clap(long)] remote: bool, + + /// With --remote: write the plain HTTP server URL instead of the CLI proxy, so the + /// editor runs its own OAuth (browser consent) flow. + #[clap(long, requires = "remote")] + oauth: bool, +} + +/// Which flavor of the Railway MCP server an install writes into a harness config. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub(crate) enum McpTransport { + /// `railway mcp` — local stdio server backed by GraphQL. + Local, + /// `railway mcp proxy` — local stdio bridge to mcp.railway.com that + /// authenticates with the CLI's stored login. + RemoteProxy, + /// Plain mcp.railway.com URL — the editor authenticates via its own OAuth flow. + RemoteOauth, +} + +impl McpTransport { + pub(crate) fn from_flags(remote: bool, oauth: bool) -> Self { + match (remote, oauth) { + (false, _) => Self::Local, + (true, false) => Self::RemoteProxy, + (true, true) => Self::RemoteOauth, + } + } +} + +/// The argv written into harness configs for the stdio transports. +fn stdio_args(transport: McpTransport) -> Vec<&'static str> { + match transport { + McpTransport::Local => vec!["mcp"], + McpTransport::RemoteProxy => vec!["mcp", "proxy"], + // RemoteOauth entries are URL-based; callers never ask for its argv. + McpTransport::RemoteOauth => unreachable!("RemoteOauth has no stdio argv"), + } } pub async fn command(args: Args) -> Result<()> { - install_mcp(&args.agent, args.remote, false).await + install_mcp( + &args.agent, + McpTransport::from_flags(args.remote, args.oauth), + false, + ) + .await } // `quiet` suppresses the section header, the "Installing … to:" line, per-tool // success lines, and the footer/restart notice — used by the embedded // agent-setup flow, which prints its own collapsed one-line summary. Failures // are always surfaced so a silent error can't contradict that summary. -pub(crate) async fn install_mcp(agent_filter: &[String], remote: bool, quiet: bool) -> Result<()> { +pub(crate) async fn install_mcp( + agent_filter: &[String], + transport: McpTransport, + quiet: bool, +) -> Result<()> { let home = dirs::home_dir().context("could not determine home directory")?; let tools = resolve_tools(&home, agent_filter)?; @@ -57,16 +104,18 @@ pub(crate) async fn install_mcp(agent_filter: &[String], remote: bool, quiet: bo } let names: Vec<_> = configurable.iter().map(|t| t.name).collect(); - let transport = if remote { - format!("remote ({})", REMOTE_MCP_URL).cyan() - } else { - "local stdio".cyan() + let transport_desc = match transport { + McpTransport::Local => "local stdio".to_string(), + McpTransport::RemoteProxy => { + format!("remote ({REMOTE_MCP_URL} via CLI proxy, uses `railway login`)") + } + McpTransport::RemoteOauth => format!("remote ({REMOTE_MCP_URL}, editor OAuth)"), }; if !quiet { println!( "{} {} {} {}\n", "Installing".bold(), - transport, + transport_desc.cyan(), "to:".bold(), names.join(", ") ); @@ -74,7 +123,7 @@ pub(crate) async fn install_mcp(agent_filter: &[String], remote: bool, quiet: bo for tool in &configurable { let path = config_path(tool.slug, &home); - match install_for(tool.slug, &path, remote) { + match install_for(tool.slug, &path, transport) { Ok(()) => { if !quiet { println!( @@ -127,7 +176,13 @@ fn config_path(slug: &str, home: &Path) -> PathBuf { } } -/// True when a railway MCP entry is configured for either transport — the +const ALL_TRANSPORTS: [McpTransport; 3] = [ + McpTransport::Local, + McpTransport::RemoteProxy, + McpTransport::RemoteOauth, +]; + +/// True when a railway MCP entry is configured for any transport — the /// help health check doesn't care which one `setup agent` installed. Reads /// each config file once instead of once per transport. pub(crate) fn mcp_configured_any_transport(home: &Path, slug: &str) -> bool { @@ -137,68 +192,95 @@ pub(crate) fn mcp_configured_any_transport(home: &Path, slug: &str) -> bool { .ok() .and_then(|root| root.pointer("/mcpServers/railway").cloned()) .is_some_and(|entry| { - json_mcp_entry_matches(&entry, false) || json_mcp_entry_matches(&entry, true) + ALL_TRANSPORTS + .iter() + .any(|t| json_mcp_entry_matches(&entry, *t)) }), "opencode" => read_json_or_empty(&path) .ok() .and_then(|root| root.pointer("/mcp/railway").cloned()) .is_some_and(|entry| { - opencode_mcp_entry_matches(&entry, false) - || opencode_mcp_entry_matches(&entry, true) + ALL_TRANSPORTS + .iter() + .any(|t| opencode_mcp_entry_matches(&entry, *t)) }), - // Codex keeps its TOML matching in one place at the cost of a second - // read for this one tool. - "codex" => codex_mcp_configured(&path, false) || codex_mcp_configured(&path, true), + // Codex keeps its TOML matching in one place at the cost of extra + // reads for this one tool. + "codex" => ALL_TRANSPORTS + .iter() + .any(|t| codex_mcp_configured(&path, *t)), _ => false, } } -pub(crate) fn mcp_configured_for_slug(home: &Path, slug: &str, remote: bool) -> bool { +pub(crate) fn mcp_configured_for_slug(home: &Path, slug: &str, transport: McpTransport) -> bool { let path = config_path(slug, home); match slug { "claude-code" | "cursor" | "copilot" | "factory-droid" => read_json_or_empty(&path) .ok() .and_then(|root| root.pointer("/mcpServers/railway").cloned()) - .is_some_and(|entry| json_mcp_entry_matches(&entry, remote)), + .is_some_and(|entry| json_mcp_entry_matches(&entry, transport)), "opencode" => read_json_or_empty(&path) .ok() .and_then(|root| root.pointer("/mcp/railway").cloned()) - .is_some_and(|entry| opencode_mcp_entry_matches(&entry, remote)), - "codex" => codex_mcp_configured(&path, remote), + .is_some_and(|entry| opencode_mcp_entry_matches(&entry, transport)), + "codex" => codex_mcp_configured(&path, transport), _ => false, } } -fn json_mcp_entry_matches(entry: &JsonValue, remote: bool) -> bool { - if remote { - entry.get("url").and_then(JsonValue::as_str) == Some(REMOTE_MCP_URL) - } else { - entry.get("command").and_then(JsonValue::as_str) == Some("railway") - && entry - .get("args") - .and_then(JsonValue::as_array) - .is_some_and(|args| args.iter().any(|arg| arg.as_str() == Some("mcp"))) +/// Distinguish `railway mcp` (local) from `railway mcp proxy`: a bare +/// "contains mcp" check would classify a proxy entry as a local install. +fn stdio_argv_matches<'a>( + args: impl Iterator + Clone, + transport: McpTransport, +) -> bool { + let has_mcp = args.clone().any(|a| a == "mcp"); + let has_proxy = args.clone().any(|a| a == "proxy"); + has_mcp && (has_proxy == matches!(transport, McpTransport::RemoteProxy)) +} + +fn json_mcp_entry_matches(entry: &JsonValue, transport: McpTransport) -> bool { + match transport { + McpTransport::RemoteOauth => { + entry.get("url").and_then(JsonValue::as_str) == Some(REMOTE_MCP_URL) + } + stdio => { + entry.get("command").and_then(JsonValue::as_str) == Some("railway") + && entry + .get("args") + .and_then(JsonValue::as_array) + .is_some_and(|args| { + stdio_argv_matches(args.iter().filter_map(JsonValue::as_str), stdio) + }) + } } } -fn opencode_mcp_entry_matches(entry: &JsonValue, remote: bool) -> bool { - if remote { - entry.get("type").and_then(JsonValue::as_str) == Some("remote") - && entry.get("url").and_then(JsonValue::as_str) == Some(REMOTE_MCP_URL) - } else { - entry.get("type").and_then(JsonValue::as_str) == Some("local") - && entry - .get("command") - .and_then(JsonValue::as_array) - .is_some_and(|command| { - command.first().and_then(JsonValue::as_str) == Some("railway") - && command.iter().any(|arg| arg.as_str() == Some("mcp")) - }) +fn opencode_mcp_entry_matches(entry: &JsonValue, transport: McpTransport) -> bool { + match transport { + McpTransport::RemoteOauth => { + entry.get("type").and_then(JsonValue::as_str) == Some("remote") + && entry.get("url").and_then(JsonValue::as_str) == Some(REMOTE_MCP_URL) + } + stdio => { + entry.get("type").and_then(JsonValue::as_str) == Some("local") + && entry + .get("command") + .and_then(JsonValue::as_array) + .is_some_and(|command| { + command.first().and_then(JsonValue::as_str) == Some("railway") + && stdio_argv_matches( + command.iter().skip(1).filter_map(JsonValue::as_str), + stdio, + ) + }) + } } } -fn codex_mcp_configured(path: &Path, remote: bool) -> bool { +fn codex_mcp_configured(path: &Path, transport: McpTransport) -> bool { let Ok(existing) = std::fs::read_to_string(path) else { return false; }; @@ -208,67 +290,70 @@ fn codex_mcp_configured(path: &Path, remote: bool) -> bool { doc.get("mcp_servers") .and_then(|servers| servers.get("railway")) - .is_some_and(|entry| { - if remote { + .is_some_and(|entry| match transport { + McpTransport::RemoteOauth => { entry.get("url").and_then(toml::Value::as_str) == Some(REMOTE_MCP_URL) - } else { + } + stdio => { entry.get("command").and_then(toml::Value::as_str) == Some("railway") && entry .get("args") .and_then(toml::Value::as_array) - .is_some_and(|args| args.iter().any(|arg| arg.as_str() == Some("mcp"))) + .is_some_and(|args| { + stdio_argv_matches(args.iter().filter_map(toml::Value::as_str), stdio) + }) } }) } -fn install_for(slug: &str, path: &Path, remote: bool) -> Result<()> { +fn install_for(slug: &str, path: &Path, transport: McpTransport) -> Result<()> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("Failed to create directory {}", parent.display()))?; } match slug { "claude-code" => { - let entry = if remote { - json!({ "type": "http", "url": REMOTE_MCP_URL }) - } else { - json!({ "command": "railway", "args": ["mcp"] }) + let entry = match transport { + McpTransport::RemoteOauth => json!({ "type": "http", "url": REMOTE_MCP_URL }), + stdio => json!({ "command": "railway", "args": stdio_args(stdio) }), }; write_json_mcp_servers(path, entry) } "cursor" => { // Cursor auto-detects HTTP/SSE from the presence of `url`. - let entry = if remote { - json!({ "url": REMOTE_MCP_URL }) - } else { - json!({ "command": "railway", "args": ["mcp"] }) + let entry = match transport { + McpTransport::RemoteOauth => json!({ "url": REMOTE_MCP_URL }), + stdio => json!({ "command": "railway", "args": stdio_args(stdio) }), }; write_json_mcp_servers(path, entry) } - "opencode" => write_opencode_mcp(path, remote), - "codex" => write_codex_toml(path, remote), + "opencode" => write_opencode_mcp(path, transport), + "codex" => write_codex_toml(path, transport), "copilot" => { - let entry = if remote { - json!({ "type": "http", "url": REMOTE_MCP_URL, "tools": ["*"] }) - } else { - json!({ + let entry = match transport { + McpTransport::RemoteOauth => { + json!({ "type": "http", "url": REMOTE_MCP_URL, "tools": ["*"] }) + } + stdio => json!({ "type": "local", "command": "railway", - "args": ["mcp"], + "args": stdio_args(stdio), "tools": ["*"] - }) + }), }; write_json_mcp_servers(path, entry) } "factory-droid" => { - let entry = if remote { - json!({ "type": "http", "url": REMOTE_MCP_URL, "disabled": false }) - } else { - json!({ + let entry = match transport { + McpTransport::RemoteOauth => { + json!({ "type": "http", "url": REMOTE_MCP_URL, "disabled": false }) + } + stdio => json!({ "type": "stdio", "command": "railway", - "args": ["mcp"], + "args": stdio_args(stdio), "disabled": false - }) + }), }; write_json_mcp_servers(path, entry) } @@ -297,20 +382,23 @@ fn write_json_mcp_servers(path: &Path, entry: JsonValue) -> Result<()> { /// OpenCode uses an `mcp` key with a slightly different per-server schema /// (`type: "local"` with `command` as an argv array, or `type: "remote"` with /// `url`). See docs.opencode.ai for the canonical shape. -fn write_opencode_mcp(path: &Path, remote: bool) -> Result<()> { +fn write_opencode_mcp(path: &Path, transport: McpTransport) -> Result<()> { let mut root = read_json_or_empty(path)?; - let entry = if remote { - json!({ + let entry = match transport { + McpTransport::RemoteOauth => json!({ "type": "remote", "url": REMOTE_MCP_URL, "enabled": true, - }) - } else { - json!({ - "type": "local", - "command": ["railway", "mcp"], - "enabled": true, - }) + }), + stdio => { + let mut command = vec!["railway"]; + command.extend(stdio_args(stdio)); + json!({ + "type": "local", + "command": command, + "enabled": true, + }) + } }; let obj = root @@ -336,7 +424,7 @@ fn write_opencode_mcp(path: &Path, remote: bool) -> Result<()> { write_json_pretty(path, &root) } -fn write_codex_toml(path: &Path, remote: bool) -> Result<()> { +fn write_codex_toml(path: &Path, transport: McpTransport) -> Result<()> { let existing = match std::fs::read_to_string(path) { Ok(s) => s, Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(), @@ -363,20 +451,28 @@ fn write_codex_toml(path: &Path, remote: bool) -> Result<()> { .context("`mcp_servers` is not a TOML table")?; let mut railway = toml::value::Table::new(); - if remote { - railway.insert( - "url".to_string(), - toml::Value::String(REMOTE_MCP_URL.to_string()), - ); - } else { - railway.insert( - "command".to_string(), - toml::Value::String("railway".to_string()), - ); - railway.insert( - "args".to_string(), - toml::Value::Array(vec![toml::Value::String("mcp".to_string())]), - ); + match transport { + McpTransport::RemoteOauth => { + railway.insert( + "url".to_string(), + toml::Value::String(REMOTE_MCP_URL.to_string()), + ); + } + stdio => { + railway.insert( + "command".to_string(), + toml::Value::String("railway".to_string()), + ); + railway.insert( + "args".to_string(), + toml::Value::Array( + stdio_args(stdio) + .into_iter() + .map(|a| toml::Value::String(a.to_string())) + .collect(), + ), + ); + } } servers.insert("railway".to_string(), toml::Value::Table(railway)); @@ -491,8 +587,55 @@ mod tests { ) .unwrap(); - assert!(mcp_configured_for_slug(home.path(), "cursor", false)); - assert!(!mcp_configured_for_slug(home.path(), "cursor", true)); + assert!(mcp_configured_for_slug( + home.path(), + "cursor", + McpTransport::Local + )); + assert!(!mcp_configured_for_slug( + home.path(), + "cursor", + McpTransport::RemoteProxy + )); + assert!(!mcp_configured_for_slug( + home.path(), + "cursor", + McpTransport::RemoteOauth + )); + } + + #[test] + fn proxy_entry_is_not_mistaken_for_local() { + let home = tempfile::tempdir().unwrap(); + let path = home.path().join(".cursor").join("mcp.json"); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + + install_for("cursor", &path, McpTransport::RemoteProxy).unwrap(); + + let written = std::fs::read_to_string(&path).unwrap(); + let root: JsonValue = serde_json::from_str(&written).unwrap(); + let railway = root.pointer("/mcpServers/railway").unwrap(); + assert_eq!( + railway.get("args").unwrap(), + &serde_json::json!(["mcp", "proxy"]) + ); + + assert!(mcp_configured_for_slug( + home.path(), + "cursor", + McpTransport::RemoteProxy + )); + assert!(!mcp_configured_for_slug( + home.path(), + "cursor", + McpTransport::Local + )); + assert!(!mcp_configured_for_slug( + home.path(), + "cursor", + McpTransport::RemoteOauth + )); + assert!(mcp_configured_any_transport(home.path(), "cursor")); } #[test] @@ -518,8 +661,53 @@ mod tests { ) .unwrap(); - assert!(mcp_configured_for_slug(home.path(), "opencode", true)); - assert!(!mcp_configured_for_slug(home.path(), "opencode", false)); + assert!(mcp_configured_for_slug( + home.path(), + "opencode", + McpTransport::RemoteOauth + )); + assert!(!mcp_configured_for_slug( + home.path(), + "opencode", + McpTransport::Local + )); + assert!(!mcp_configured_for_slug( + home.path(), + "opencode", + McpTransport::RemoteProxy + )); + } + + #[test] + fn writes_and_detects_opencode_proxy_mcp() { + let home = tempfile::tempdir().unwrap(); + let path = home + .path() + .join(".config") + .join("opencode") + .join("opencode.json"); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + + install_for("opencode", &path, McpTransport::RemoteProxy).unwrap(); + + let written = std::fs::read_to_string(&path).unwrap(); + let root: JsonValue = serde_json::from_str(&written).unwrap(); + let railway = root.pointer("/mcp/railway").unwrap(); + assert_eq!( + railway.get("command").unwrap(), + &serde_json::json!(["railway", "mcp", "proxy"]) + ); + + assert!(mcp_configured_for_slug( + home.path(), + "opencode", + McpTransport::RemoteProxy + )); + assert!(!mcp_configured_for_slug( + home.path(), + "opencode", + McpTransport::Local + )); } #[test] @@ -537,8 +725,52 @@ mod tests { ) .unwrap(); - assert!(mcp_configured_for_slug(home.path(), "codex", false)); - assert!(!mcp_configured_for_slug(home.path(), "codex", true)); + assert!(mcp_configured_for_slug( + home.path(), + "codex", + McpTransport::Local + )); + assert!(!mcp_configured_for_slug( + home.path(), + "codex", + McpTransport::RemoteProxy + )); + assert!(!mcp_configured_for_slug( + home.path(), + "codex", + McpTransport::RemoteOauth + )); + } + + #[test] + fn writes_and_detects_codex_proxy_mcp() { + let home = tempfile::tempdir().unwrap(); + let path = home.path().join(".codex").join("config.toml"); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + + write_codex_toml(&path, McpTransport::RemoteProxy).unwrap(); + + let written = std::fs::read_to_string(&path).unwrap(); + let doc = written.parse::().unwrap(); + let args = doc + .get("mcp_servers") + .and_then(|servers| servers.get("railway")) + .and_then(|railway| railway.get("args")) + .and_then(toml::Value::as_array) + .unwrap(); + assert_eq!(args.len(), 2); + assert_eq!(args[1].as_str(), Some("proxy")); + + assert!(mcp_configured_for_slug( + home.path(), + "codex", + McpTransport::RemoteProxy + )); + assert!(!mcp_configured_for_slug( + home.path(), + "codex", + McpTransport::Local + )); } #[test] @@ -555,8 +787,16 @@ mod tests { ) .unwrap(); - assert!(mcp_configured_for_slug(home.path(), "codex", true)); - assert!(!mcp_configured_for_slug(home.path(), "codex", false)); + assert!(mcp_configured_for_slug( + home.path(), + "codex", + McpTransport::RemoteOauth + )); + assert!(!mcp_configured_for_slug( + home.path(), + "codex", + McpTransport::Local + )); } #[test] @@ -565,7 +805,7 @@ mod tests { let path = home.path().join(".codex").join("config.toml"); std::fs::create_dir_all(path.parent().unwrap()).unwrap(); - write_codex_toml(&path, true).unwrap(); + write_codex_toml(&path, McpTransport::RemoteOauth).unwrap(); let written = std::fs::read_to_string(&path).unwrap(); let doc = written.parse::().unwrap(); @@ -588,7 +828,7 @@ mod tests { let path = home.path().join(".copilot").join("mcp-config.json"); std::fs::create_dir_all(path.parent().unwrap()).unwrap(); - install_for("copilot", &path, false).unwrap(); + install_for("copilot", &path, McpTransport::Local).unwrap(); let written = std::fs::read_to_string(&path).unwrap(); let root: JsonValue = serde_json::from_str(&written).unwrap(); @@ -602,7 +842,11 @@ mod tests { railway.get("command").and_then(JsonValue::as_str), Some("railway") ); - assert!(mcp_configured_for_slug(home.path(), "copilot", false)); + assert!(mcp_configured_for_slug( + home.path(), + "copilot", + McpTransport::Local + )); } #[test] @@ -611,7 +855,7 @@ mod tests { let path = home.path().join(".factory").join("mcp.json"); std::fs::create_dir_all(path.parent().unwrap()).unwrap(); - install_for("factory-droid", &path, true).unwrap(); + install_for("factory-droid", &path, McpTransport::RemoteOauth).unwrap(); let written = std::fs::read_to_string(&path).unwrap(); let root: JsonValue = serde_json::from_str(&written).unwrap(); @@ -625,6 +869,10 @@ mod tests { railway.get("url").and_then(JsonValue::as_str), Some("https://mcp.railway.com") ); - assert!(mcp_configured_for_slug(home.path(), "factory-droid", true)); + assert!(mcp_configured_for_slug( + home.path(), + "factory-droid", + McpTransport::RemoteOauth + )); } } diff --git a/src/commands/mcp/mod.rs b/src/commands/mcp/mod.rs index c9ed82366..a39c525c4 100644 --- a/src/commands/mcp/mod.rs +++ b/src/commands/mcp/mod.rs @@ -4,6 +4,7 @@ use rmcp::{ServiceExt, transport::stdio}; mod handler; pub(crate) mod install; pub(crate) mod params; +mod proxy; mod tools; use handler::RailwayMcp; @@ -18,12 +19,15 @@ pub struct Args { enum Commands { /// Install Railway's MCP server config into AI coding tools (Claude Code, Cursor, OpenCode, Codex) Install(install::Args), + /// Proxy the remote MCP server (mcp.railway.com) over stdio, authenticating with your CLI login + Proxy, } pub async fn command(args: Args) -> Result<()> { match args.command { None => serve_stdio().await, Some(Commands::Install(install_args)) => install::command(install_args).await, + Some(Commands::Proxy) => proxy::serve_proxy().await, } } diff --git a/src/commands/mcp/proxy.rs b/src/commands/mcp/proxy.rs new file mode 100644 index 000000000..58fb3c6ba --- /dev/null +++ b/src/commands/mcp/proxy.rs @@ -0,0 +1,612 @@ +//! Stdio ↔ streamable-HTTP proxy for the remote Railway MCP server. +//! +//! Bridges a harness speaking MCP over stdio (Claude Code, Cursor, …) to +//! `mcp.railway.com`, attaching a fresh `Authorization: Bearer` from the CLI's +//! stored login on every request. This lets users who have already run +//! `railway login` use the remote MCP server without going through the +//! harness's OAuth (DCR + browser consent) flow, and without ever writing a +//! long-lived credential into the harness config. +//! +//! Auth freshness is delegated to [`crate::client::ensure_valid_token`], which +//! serializes refreshes across concurrent CLI processes via the config +//! lockfile — the proxy can safely run alongside the local `railway mcp` +//! server and any other CLI invocations. +//! +//! The remote server currently runs the streamable-HTTP transport statelessly +//! (no `Mcp-Session-Id` issued), but the proxy tracks a session id anyway and +//! transparently re-initializes + retries once if the server ever reports a +//! missing/expired session. Server-initiated messages (the optional GET SSE +//! stream) are not proxied; nothing in the current tool surface relies on +//! them. + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, Result}; +use futures_util::StreamExt; +use serde_json::{Value as JsonValue, json}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::sync::{Mutex, mpsc}; + +use crate::client::ensure_valid_token; +use crate::commands::Environment; +use crate::config::Configs; +use crate::consts; + +/// JSON-RPC error code for auth failures surfaced by the proxy itself. +const AUTH_ERROR_CODE: i64 = -32001; + +const LOGIN_HINT: &str = "Not logged in to Railway. Run `railway login` in a terminal, then retry \ + — the proxy picks up the new login automatically, no restart needed."; + +struct ProxyState { + http: reqwest::Client, + url: String, + configs: Mutex, + session: Mutex, +} + +#[derive(Default)] +struct SessionMeta { + id: Option, + /// The harness's `initialize` request, kept so the proxy can re-establish + /// an upstream session (expiry, or a degraded logged-out start) without + /// involving the harness. + init_request: Option, +} + +type Out = mpsc::UnboundedSender; + +pub async fn serve_proxy() -> Result<()> { + let configs = Configs::new()?; + let url = resolve_mcp_url(&configs); + + let http = reqwest::Client::builder() + .danger_accept_invalid_certs(matches!(Configs::get_environment_id(), Environment::Dev)) + .user_agent(consts::get_user_agent()) + .connect_timeout(Duration::from_secs(15)) + // No overall timeout: tool calls (e.g. railway-agent) can legitimately + // stream for minutes. + .build() + .context("Failed to build HTTP client")?; + + let state = Arc::new(ProxyState { + http, + url, + configs: Mutex::new(configs), + session: Mutex::new(SessionMeta::default()), + }); + + // All stdout writes go through one task so concurrent responses can't + // interleave within a line. + let (tx, mut rx) = mpsc::unbounded_channel::(); + let writer = tokio::spawn(async move { + let mut stdout = tokio::io::stdout(); + while let Some(line) = rx.recv().await { + let _ = stdout.write_all(line.as_bytes()).await; + let _ = stdout.write_all(b"\n").await; + let _ = stdout.flush().await; + } + }); + + let stdin = BufReader::new(tokio::io::stdin()); + let mut lines = stdin.lines(); + // Process messages inline until the MCP handshake completes so the + // initialize → initialized ordering is preserved upstream, then handle + // messages concurrently (harnesses issue parallel tool calls). + let mut handshake_done = false; + + while let Some(line) = lines.next_line().await? { + if line.trim().is_empty() { + continue; + } + let msg: JsonValue = match serde_json::from_str(&line) { + Ok(v) => v, + Err(e) => { + eprintln!("railway mcp proxy: ignoring unparseable message: {e}"); + continue; + } + }; + + if method_of(&msg) == Some("initialize") { + state.session.lock().await.init_request = Some(msg.clone()); + } + + if handshake_done { + let state = state.clone(); + let tx = tx.clone(); + tokio::spawn(async move { + handle_message(&state, msg, &tx).await; + }); + } else { + let completes_handshake = method_of(&msg) == Some("notifications/initialized"); + handle_message(&state, msg, &tx).await; + if completes_handshake { + handshake_done = true; + } + } + } + + // stdin closed: the harness is gone. Best-effort end of the remote + // session, then a bounded wait for in-flight tasks — a stalled upstream + // stream must not keep an orphaned proxy alive after the harness exits. + end_session(&state).await; + drop(tx); + let _ = tokio::time::timeout(Duration::from_secs(5), writer).await; + Ok(()) +} + +fn method_of(msg: &JsonValue) -> Option<&str> { + msg.get("method").and_then(JsonValue::as_str) +} + +/// Request ids awaiting a response in this message — one for a plain request, +/// several for a JSON-RPC batch (protocol ≤2025-03-26 allows top-level +/// arrays), none for notifications. Error paths must answer every id or the +/// harness waits forever. +fn ids_of(msg: &JsonValue) -> Vec { + match msg { + JsonValue::Array(items) => items + .iter() + .filter_map(|m| m.get("id").cloned().filter(|id| !id.is_null())) + .collect(), + _ => msg + .get("id") + .cloned() + .filter(|id| !id.is_null()) + .into_iter() + .collect(), + } +} + +fn resolve_mcp_url(configs: &Configs) -> String { + if let Ok(url) = std::env::var("RAILWAY_MCP_URL") { + let url = url.trim(); + if !url.is_empty() { + return url.trim_end_matches('/').to_string(); + } + } + format!("https://mcp.{}", configs.get_host()) +} + +async fn handle_message(state: &ProxyState, msg: JsonValue, out: &Out) { + let ids = ids_of(&msg); + + let Some(token) = fresh_token(state).await else { + respond_unauthenticated(&msg, &ids, out); + return; + }; + + if let Err(e) = forward(state, &msg, &token, out).await { + if ids.is_empty() { + eprintln!("railway mcp proxy: {e:#}"); + } + for id in &ids { + send_error(out, id, -32603, &format!("Railway MCP proxy error: {e:#}")); + } + } +} + +/// Get a currently-valid auth token, refreshing the stored OAuth credentials +/// if they have expired. Returns `None` when the user has no usable login. +async fn fresh_token(state: &ProxyState) -> Option { + let mut configs = state.configs.lock().await; + // A proxy started before `railway login` has no token in memory, and + // `ensure_valid_token`'s fast path never re-reads the config in that + // state. Reload from disk so a login that happened after startup is + // picked up — this is what makes the LOGIN_HINT's "no restart needed" + // promise true. + if configs.get_railway_auth_token().is_none() { + if let Err(e) = configs.reload() { + eprintln!("railway mcp proxy: config reload failed: {e:#}"); + } + } + if let Err(e) = ensure_valid_token(&mut configs).await { + eprintln!("railway mcp proxy: token refresh failed: {e:#}"); + } + configs.get_railway_auth_token() +} + +/// Without a login the proxy still completes the MCP handshake (a crashed +/// server renders as an opaque failure in most harnesses) and answers every +/// request with an actionable error. Once the user logs in, the next request +/// heals automatically via the re-initialize path in [`forward`]. +fn respond_unauthenticated(msg: &JsonValue, ids: &[JsonValue], out: &Out) { + let [id] = ids else { + // Batch or notification: answer every id (none for a notification). + for id in ids { + send_error(out, id, AUTH_ERROR_CODE, LOGIN_HINT); + } + return; + }; + + if method_of(msg) == Some("initialize") { + let protocol = msg + .pointer("/params/protocolVersion") + .and_then(JsonValue::as_str) + .unwrap_or("2025-03-26"); + let result = json!({ + "jsonrpc": "2.0", + "id": id, + "result": { + "protocolVersion": protocol, + "capabilities": { "tools": { "listChanged": true } }, + "serverInfo": { + "name": "railway", + "version": env!("CARGO_PKG_VERSION"), + }, + "instructions": LOGIN_HINT, + } + }); + let _ = out.send(result.to_string()); + } else { + send_error(out, id, AUTH_ERROR_CODE, LOGIN_HINT); + } +} + +async fn forward(state: &ProxyState, msg: &JsonValue, token: &str, out: &Out) -> Result<()> { + let is_initialize = method_of(msg) == Some("initialize"); + let session_id = state.session.lock().await.id.clone(); + + let resp = post_message( + state, + msg, + token, + if is_initialize { + None + } else { + session_id.as_deref() + }, + ) + .await?; + + // Session lost or never established upstream (server-side expiry, or the + // proxy started degraded while logged out): re-initialize with the + // captured initialize request and retry once. + let status = resp.status(); + let can_reinit = { state.session.lock().await.init_request.is_some() }; + if !is_initialize && (status == 404 || status == 400) && can_reinit { + let _ = resp.bytes().await; + reinitialize(state, token).await?; + let session_id = state.session.lock().await.id.clone(); + let resp = post_message(state, msg, token, session_id.as_deref()).await?; + return consume_response(state, resp, msg, is_initialize, out).await; + } + + consume_response(state, resp, msg, is_initialize, out).await +} + +async fn post_message( + state: &ProxyState, + msg: &JsonValue, + token: &str, + session_id: Option<&str>, +) -> Result { + let mut req = state + .http + .post(&state.url) + .header("authorization", format!("Bearer {token}")) + .header("accept", "application/json, text/event-stream") + .header("x-source", consts::get_user_agent()); + if let Some(sid) = session_id { + req = req.header("mcp-session-id", sid); + } + req.json(msg) + .send() + .await + .context("failed to reach the remote MCP server") +} + +/// Re-run the MCP handshake upstream using the harness's captured `initialize` +/// request, discarding the result (the harness already completed its own +/// handshake). Serialized behind the session lock so concurrent failures +/// don't stampede. +async fn reinitialize(state: &ProxyState, token: &str) -> Result<()> { + let mut session = state.session.lock().await; + let init = session + .init_request + .clone() + .context("no initialize request captured yet")?; + + let resp = post_message(state, &init, token, None).await?; + anyhow::ensure!( + resp.status().is_success(), + "re-initialize failed with HTTP {}", + resp.status() + ); + session.id = resp + .headers() + .get("mcp-session-id") + .and_then(|v| v.to_str().ok()) + .map(str::to_string); + let _ = resp.bytes().await; + let session_id = session.id.clone(); + drop(session); + + let initialized = json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }); + let resp = post_message(state, &initialized, token, session_id.as_deref()).await?; + let _ = resp.bytes().await; + Ok(()) +} + +async fn consume_response( + state: &ProxyState, + resp: reqwest::Response, + msg: &JsonValue, + is_initialize: bool, + out: &Out, +) -> Result<()> { + let status = resp.status(); + + if is_initialize { + if let Some(sid) = resp + .headers() + .get("mcp-session-id") + .and_then(|v| v.to_str().ok()) + { + state.session.lock().await.id = Some(sid.to_string()); + } + } + + if status == 401 || status == 403 { + let _ = resp.bytes().await; + for id in &ids_of(msg) { + send_error( + out, + id, + AUTH_ERROR_CODE, + "Railway rejected the CLI's credentials. Run `railway login` and try again.", + ); + } + return Ok(()); + } + + // Accepted notification/response with no body. + if status == 202 || status == 204 { + return Ok(()); + } + + if !status.is_success() { + let body = resp.text().await.unwrap_or_default(); + anyhow::bail!( + "remote MCP server returned HTTP {status}: {}", + truncate(&body, 300) + ); + } + + let content_type = resp + .headers() + .get("content-type") + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + + if content_type.starts_with("text/event-stream") { + stream_sse(resp, out).await + } else { + let body = resp.text().await?; + emit_json_line(body.trim(), out); + Ok(()) + } +} + +/// Relay every SSE `data:` payload to stdout as its own JSON-RPC line. The +/// server closes the per-request stream after the final response message. +async fn stream_sse(resp: reqwest::Response, out: &Out) -> Result<()> { + let mut stream = resp.bytes_stream(); + let mut buf: Vec = Vec::new(); + + while let Some(chunk) = stream.next().await { + let chunk = chunk.context("error reading SSE stream from remote MCP server")?; + buf.extend_from_slice(&chunk); + while let Some((event_len, boundary_end)) = find_event_boundary(&buf) { + let event: Vec = buf.drain(..boundary_end).collect(); + emit_sse_event(&event[..event_len], out); + } + } + if !buf.is_empty() { + emit_sse_event(&buf, out); + } + Ok(()) +} + +/// Find the end of the next SSE event: a blank line, i.e. `\n\n` or +/// `\r\n\r\n`. Returns (event bytes length, total length including boundary). +fn find_event_boundary(buf: &[u8]) -> Option<(usize, usize)> { + for i in 0..buf.len() { + if buf[i] != b'\n' { + continue; + } + if buf.get(i + 1) == Some(&b'\n') { + return Some((i, i + 2)); + } + if buf.get(i + 1) == Some(&b'\r') && buf.get(i + 2) == Some(&b'\n') { + return Some((i, i + 3)); + } + } + None +} + +fn emit_sse_event(raw: &[u8], out: &Out) { + let text = String::from_utf8_lossy(raw); + let data_lines: Vec<&str> = text + .lines() + .filter_map(|line| line.strip_prefix("data:")) + .map(|rest| rest.strip_prefix(' ').unwrap_or(rest)) + .collect(); + if data_lines.is_empty() { + return; + } + emit_json_line(&data_lines.join("\n"), out); +} + +/// Write one JSON-RPC message as a single stdout line. Payloads are compacted +/// through serde so an upstream message containing raw newlines can't corrupt +/// the newline-delimited stdio framing. +fn emit_json_line(payload: &str, out: &Out) { + if payload.is_empty() { + return; + } + let line = serde_json::from_str::(payload) + .map(|v| v.to_string()) + .unwrap_or_else(|_| payload.replace(['\n', '\r'], " ")); + let _ = out.send(line); +} + +fn send_error(out: &Out, id: &JsonValue, code: i64, message: &str) { + let err = json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": code, "message": message }, + }); + let _ = out.send(err.to_string()); +} + +/// Best-effort session teardown when the harness disconnects. +async fn end_session(state: &ProxyState) { + let session_id = state.session.lock().await.id.clone(); + let Some(session_id) = session_id else { return }; + let token = { state.configs.lock().await.get_railway_auth_token() }; + let Some(token) = token else { return }; + let _ = state + .http + .delete(&state.url) + .header("authorization", format!("Bearer {token}")) + .header("mcp-session-id", session_id) + .timeout(Duration::from_secs(5)) + .send() + .await; +} + +fn truncate(s: &str, max_chars: usize) -> String { + if s.chars().count() <= max_chars { + s.to_string() + } else { + let mut out: String = s.chars().take(max_chars).collect(); + out.push('…'); + out + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn collect(rx: &mut mpsc::UnboundedReceiver) -> Vec { + let mut out = Vec::new(); + while let Ok(line) = rx.try_recv() { + out.push(line); + } + out + } + + #[test] + fn sse_event_boundary_handles_lf_and_crlf() { + assert_eq!(find_event_boundary(b"data: {}\n\nrest"), Some((8, 10))); + assert_eq!(find_event_boundary(b"data: {}\r\n\r\nrest"), Some((9, 12))); + assert_eq!(find_event_boundary(b"data: {}"), None); + } + + #[test] + fn sse_event_extracts_data_payload() { + let (tx, mut rx) = mpsc::unbounded_channel(); + emit_sse_event(b"event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1}", &tx); + assert_eq!(collect(&mut rx), vec![r#"{"id":1,"jsonrpc":"2.0"}"#]); + } + + #[test] + fn sse_event_joins_multiline_data() { + let (tx, mut rx) = mpsc::unbounded_channel(); + emit_sse_event(b"data: {\"a\":\ndata: 1}", &tx); + assert_eq!(collect(&mut rx), vec![r#"{"a":1}"#]); + } + + #[test] + fn sse_event_without_data_is_dropped() { + let (tx, mut rx) = mpsc::unbounded_channel(); + emit_sse_event(b"event: ping\nid: 4", &tx); + assert!(collect(&mut rx).is_empty()); + } + + #[test] + fn json_lines_are_compacted_to_one_line() { + let (tx, mut rx) = mpsc::unbounded_channel(); + emit_json_line("{\n \"jsonrpc\": \"2.0\",\n \"id\": 7\n}", &tx); + let lines = collect(&mut rx); + assert_eq!(lines.len(), 1); + assert!(!lines[0].contains('\n')); + } + + #[test] + fn unauthenticated_initialize_fabricates_result() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let msg = json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { "protocolVersion": "2025-06-18" }, + }); + respond_unauthenticated(&msg, &ids_of(&msg), &tx); + let lines = collect(&mut rx); + assert_eq!(lines.len(), 1); + let parsed: JsonValue = serde_json::from_str(&lines[0]).unwrap(); + assert_eq!( + parsed.pointer("/result/protocolVersion").unwrap(), + "2025-06-18" + ); + assert!( + parsed + .pointer("/result/instructions") + .unwrap() + .as_str() + .unwrap() + .contains("railway login") + ); + } + + #[test] + fn unauthenticated_request_gets_actionable_error() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let msg = json!({ "jsonrpc": "2.0", "id": 3, "method": "tools/list" }); + respond_unauthenticated(&msg, &ids_of(&msg), &tx); + let lines = collect(&mut rx); + assert_eq!(lines.len(), 1); + let parsed: JsonValue = serde_json::from_str(&lines[0]).unwrap(); + assert_eq!(parsed.pointer("/error/code").unwrap(), AUTH_ERROR_CODE); + assert!( + parsed + .pointer("/error/message") + .unwrap() + .as_str() + .unwrap() + .contains("railway login") + ); + } + + #[test] + fn unauthenticated_notification_is_dropped() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let msg = json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }); + respond_unauthenticated(&msg, &ids_of(&msg), &tx); + assert!(collect(&mut rx).is_empty()); + } + + #[test] + fn unauthenticated_batch_answers_every_id() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let msg = json!([ + { "jsonrpc": "2.0", "id": 1, "method": "tools/list" }, + { "jsonrpc": "2.0", "method": "notifications/progress" }, + { "jsonrpc": "2.0", "id": "two", "method": "tools/call" }, + ]); + let ids = ids_of(&msg); + assert_eq!(ids, vec![json!(1), json!("two")]); + respond_unauthenticated(&msg, &ids, &tx); + let lines = collect(&mut rx); + assert_eq!(lines.len(), 2); + for line in &lines { + let parsed: JsonValue = serde_json::from_str(line).unwrap(); + assert_eq!(parsed.pointer("/error/code").unwrap(), AUTH_ERROR_CODE); + } + } +} diff --git a/src/commands/setup.rs b/src/commands/setup.rs index ae6f2f9d9..d4f392e10 100644 --- a/src/commands/setup.rs +++ b/src/commands/setup.rs @@ -35,9 +35,15 @@ pub struct AgentArgs { #[clap(short = 'y', long)] yes: bool, - /// Configure the remote HTTP MCP server at mcp.railway.com instead of the local stdio server. + /// Configure the remote MCP server (mcp.railway.com) via the CLI proxy — it authenticates + /// with your existing `railway login`, no browser OAuth flow. #[clap(long)] remote: bool, + + /// With --remote: write the plain HTTP server URL instead of the CLI proxy, so the + /// editor runs its own OAuth (browser consent) flow. + #[clap(long, requires = "remote")] + oauth: bool, } pub async fn command(args: Args) -> Result<()> { @@ -66,7 +72,8 @@ impl fmt::Display for ToolChoice { #[derive(Clone, Copy, PartialEq, Eq)] enum McpChoice { Local, - Remote, + RemoteProxy, + RemoteOauth, Skip, } @@ -78,24 +85,42 @@ impl fmt::Display for McpChoice { "Local (default) {}", "— runs `railway mcp` as a stdio server".dimmed() ), - McpChoice::Remote => write!( + McpChoice::RemoteProxy => write!( f, "Remote {}", - "— https://mcp.railway.com (HTTP)".dimmed() + "— mcp.railway.com via the CLI proxy (uses your `railway login`)".dimmed() + ), + McpChoice::RemoteOauth => write!( + f, + "Remote (OAuth) {}", + "— https://mcp.railway.com directly; your editor handles OAuth".dimmed() ), McpChoice::Skip => write!(f, "Skip {}", "— don't configure MCP".dimmed()), } } } -fn pick_mcp_choice(remote_flag: bool, non_interactive: bool) -> Result { +fn pick_mcp_choice( + remote_flag: bool, + oauth_flag: bool, + non_interactive: bool, +) -> Result { if remote_flag { - return Ok(McpChoice::Remote); + return Ok(if oauth_flag { + McpChoice::RemoteOauth + } else { + McpChoice::RemoteProxy + }); } if non_interactive { return Ok(McpChoice::Local); } - let options = vec![McpChoice::Local, McpChoice::Remote, McpChoice::Skip]; + let options = vec![ + McpChoice::Local, + McpChoice::RemoteProxy, + McpChoice::RemoteOauth, + McpChoice::Skip, + ]; inquire::Select::new("Configure MCP server:", options) .with_render_config(Configs::get_render_config()) .prompt() @@ -240,15 +265,37 @@ async fn agent_setup_inner(args: AgentArgs) -> Result> { // Step 2: MCP install (skips universal internally — no MCP convention). // `--remote` short-circuits the prompt; `-y`/non-TTY defaults to local. - let mcp_choice = pick_mcp_choice(args.remote, non_interactive)?; + let mcp_choice = pick_mcp_choice(args.remote, args.oauth, non_interactive)?; let mcp_transport = match mcp_choice { McpChoice::Local => { - install_missing_mcp(&home, &selected_slugs, false, embedded).await?; + install_missing_mcp( + &home, + &selected_slugs, + mcp_install::McpTransport::Local, + embedded, + ) + .await?; Some("local") } - McpChoice::Remote => { - install_missing_mcp(&home, &selected_slugs, true, embedded).await?; - Some("remote") + McpChoice::RemoteProxy => { + install_missing_mcp( + &home, + &selected_slugs, + mcp_install::McpTransport::RemoteProxy, + embedded, + ) + .await?; + Some("remote-proxy") + } + McpChoice::RemoteOauth => { + install_missing_mcp( + &home, + &selected_slugs, + mcp_install::McpTransport::RemoteOauth, + embedded, + ) + .await?; + Some("remote-oauth") } McpChoice::Skip => { if !embedded { @@ -420,13 +467,13 @@ pub(crate) fn print_agent_health_check() { async fn install_missing_mcp( home: &std::path::Path, selected_slugs: &[String], - remote: bool, + transport: mcp_install::McpTransport, quiet: bool, ) -> Result<()> { let missing_mcp: Vec = selected_slugs .iter() .filter(|slug| slug.as_str() != "universal") - .filter(|slug| !mcp_install::mcp_configured_for_slug(home, slug, remote)) + .filter(|slug| !mcp_install::mcp_configured_for_slug(home, slug, transport)) .cloned() .collect(); @@ -441,7 +488,7 @@ async fn install_missing_mcp( return Ok(()); } - mcp_install::install_mcp(&missing_mcp, remote, quiet).await + mcp_install::install_mcp(&missing_mcp, transport, quiet).await } /// Mirrors the logic at the top of `login::command` without invoking the From 3416cc75072ecba8744d58849c7a7cedd77a0f8e Mon Sep 17 00:00:00 2001 From: Cody De Arkland Date: Mon, 27 Jul 2026 01:15:10 -0700 Subject: [PATCH 2/2] harden(mcp): bound proxy response buffers + refuse redirects/plaintext Security review of the credential-injecting proxy. The token-handling core was already sound (destination fixed at startup and not derivable from server/harness input; reqwest strips the Bearer on cross-host redirect, verified against the 0.12.28 source; TLS enforced outside Dev; no path stringifies the credential). Closing the residual robustness/defense-in-depth gaps it found: - Bound every upstream response to MAX_RESPONSE_BYTES (32 MiB). reqwest's text()/bytes() have no size cap and the proxy sets no overall timeout, so a compromised edge or dev/http-override MITM streaming a boundary-less SSE body (or one giant event/body) could OOM a long-lived proxy. stream_sse and a new read_body_capped helper now abort past the ceiling. - redirect(Policy::none()): an MCP JSON-RPC POST is never legitimately redirected. The Bearer is stripped cross-host, but refusing redirects outright also closes the same-host 307/308 body-resend and the redirected-response relay edges. - Validate RAILWAY_MCP_URL: reject non-https outside Dev so the override can't send the Bearer over plaintext (the first hop isn't covered by the cross-host strip). Logic extracted to validate_mcp_override + unit test. Co-Authored-By: Claude Fable 5 --- src/commands/mcp/proxy.rs | 97 +++++++++++++++++++++++++++++++++++---- 1 file changed, 88 insertions(+), 9 deletions(-) diff --git a/src/commands/mcp/proxy.rs b/src/commands/mcp/proxy.rs index 58fb3c6ba..76b213d8a 100644 --- a/src/commands/mcp/proxy.rs +++ b/src/commands/mcp/proxy.rs @@ -36,6 +36,13 @@ use crate::consts; /// JSON-RPC error code for auth failures surfaced by the proxy itself. const AUTH_ERROR_CODE: i64 = -32001; +/// Hard ceiling on a single upstream response (one SSE stream, or a non-SSE +/// body). The proxy runs long-lived and attaches a live credential on every +/// request, so a compromised edge (or a dev/http override MITM) streaming a +/// boundary-less or unbounded body must not be able to grow memory without +/// limit. Generous enough for the largest legitimate tool payloads. +const MAX_RESPONSE_BYTES: usize = 32 * 1024 * 1024; + const LOGIN_HINT: &str = "Not logged in to Railway. Run `railway login` in a terminal, then retry \ — the proxy picks up the new login automatically, no restart needed."; @@ -59,12 +66,18 @@ type Out = mpsc::UnboundedSender; pub async fn serve_proxy() -> Result<()> { let configs = Configs::new()?; - let url = resolve_mcp_url(&configs); + let url = resolve_mcp_url(&configs)?; let http = reqwest::Client::builder() .danger_accept_invalid_certs(matches!(Configs::get_environment_id(), Environment::Dev)) .user_agent(consts::get_user_agent()) .connect_timeout(Duration::from_secs(15)) + // An MCP JSON-RPC POST is never legitimately redirected. Following + // redirects on a request that carries a Bearer is unnecessary attack + // surface — reqwest strips the token cross-host, but a same-host 307/308 + // would re-send the body and the redirected response is relayed blind. + // Refuse redirects outright. + .redirect(reqwest::redirect::Policy::none()) // No overall timeout: tool calls (e.g. railway-agent) can legitimately // stream for minutes. .build() @@ -159,14 +172,35 @@ fn ids_of(msg: &JsonValue) -> Vec { } } -fn resolve_mcp_url(configs: &Configs) -> String { - if let Ok(url) = std::env::var("RAILWAY_MCP_URL") { - let url = url.trim(); - if !url.is_empty() { - return url.trim_end_matches('/').to_string(); +fn resolve_mcp_url(configs: &Configs) -> Result { + let is_dev = matches!(Configs::get_environment_id(), Environment::Dev); + if let Ok(raw) = std::env::var("RAILWAY_MCP_URL") { + if let Some(url) = validate_mcp_override(&raw, is_dev)? { + return Ok(url); } } - format!("https://mcp.{}", configs.get_host()) + Ok(format!("https://mcp.{}", configs.get_host())) +} + +/// Validate a `RAILWAY_MCP_URL` override. Returns the normalized URL, `None` +/// when the value is blank (caller falls back to the default), or an error +/// when it would send the Bearer over a non-TLS connection. +/// +/// The Bearer is attached to every request to this URL, and the cross-host +/// redirect strip does not protect the *first* hop — so a plaintext target +/// leaks the credential outright. Require https except in the local Dev +/// environment, where a plaintext raildev endpoint is expected. +fn validate_mcp_override(raw: &str, is_dev: bool) -> Result> { + let url = raw.trim(); + if url.is_empty() { + return Ok(None); + } + if !url.starts_with("https://") && !is_dev { + anyhow::bail!( + "RAILWAY_MCP_URL must be an https:// URL (got {url:?}); refusing to send credentials over a non-TLS connection." + ); + } + Ok(Some(url.trim_end_matches('/').to_string())) } async fn handle_message(state: &ProxyState, msg: JsonValue, out: &Out) { @@ -367,7 +401,7 @@ async fn consume_response( } if !status.is_success() { - let body = resp.text().await.unwrap_or_default(); + let body = read_body_capped(resp).await.unwrap_or_default(); anyhow::bail!( "remote MCP server returned HTTP {status}: {}", truncate(&body, 300) @@ -384,12 +418,31 @@ async fn consume_response( if content_type.starts_with("text/event-stream") { stream_sse(resp, out).await } else { - let body = resp.text().await?; + let body = read_body_capped(resp).await?; emit_json_line(body.trim(), out); Ok(()) } } +/// Read a full (non-streaming) response body, refusing to buffer more than +/// [`MAX_RESPONSE_BYTES`]. `reqwest`'s `text()`/`bytes()` have no size cap, so +/// a compromised or MITM'd upstream could otherwise stream an unbounded body +/// into a long-lived proxy and exhaust memory. +async fn read_body_capped(resp: reqwest::Response) -> Result { + let mut stream = resp.bytes_stream(); + let mut buf: Vec = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.context("error reading response from remote MCP server")?; + if buf.len() + chunk.len() > MAX_RESPONSE_BYTES { + anyhow::bail!( + "remote MCP server response exceeded {MAX_RESPONSE_BYTES} bytes; aborting." + ); + } + buf.extend_from_slice(&chunk); + } + Ok(String::from_utf8_lossy(&buf).into_owned()) +} + /// Relay every SSE `data:` payload to stdout as its own JSON-RPC line. The /// server closes the per-request stream after the final response message. async fn stream_sse(resp: reqwest::Response, out: &Out) -> Result<()> { @@ -403,6 +456,14 @@ async fn stream_sse(resp: reqwest::Response, out: &Out) -> Result<()> { let event: Vec = buf.drain(..boundary_end).collect(); emit_sse_event(&event[..event_len], out); } + // A boundary-less stream (or one giant event) would otherwise grow buf + // without limit. Cap it: past the ceiling, no legitimate single SSE + // event is pending — abort rather than let a bad upstream OOM us. + if buf.len() > MAX_RESPONSE_BYTES { + anyhow::bail!( + "remote MCP server SSE event exceeded {MAX_RESPONSE_BYTES} bytes; aborting." + ); + } } if !buf.is_empty() { emit_sse_event(&buf, out); @@ -500,6 +561,24 @@ mod tests { out } + #[test] + fn mcp_override_rejects_plaintext_outside_dev() { + // http:// would send the Bearer in the clear — refused in prod/staging. + assert!(validate_mcp_override("http://evil.example/mcp", false).is_err()); + // Blank falls back to the default (Ok(None), not an error). + assert_eq!(validate_mcp_override(" ", false).unwrap(), None); + // https is accepted and trailing slashes normalized. + assert_eq!( + validate_mcp_override("https://mcp.railway.com/", false).unwrap(), + Some("https://mcp.railway.com".to_string()), + ); + // Dev allows plaintext for a local raildev endpoint. + assert_eq!( + validate_mcp_override("http://localhost:8080", true).unwrap(), + Some("http://localhost:8080".to_string()), + ); + } + #[test] fn sse_event_boundary_handles_lf_and_crlf() { assert_eq!(find_event_boundary(b"data: {}\n\nrest"), Some((8, 10)));