diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index cd86ddb88f..8062d60f4b 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -5082,6 +5082,7 @@ name = "codex-tcp-tunnel" version = "0.0.0" dependencies = [ "anyhow", + "base64 0.22.1", "bytes", "clap", "h3", diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 285c1222ec..a8a5ed7e7b 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -105,6 +105,7 @@ use codex_login::AuthManager; use codex_memories_write::clear_memory_roots_contents; use codex_models_manager::bundled_models_response; use codex_models_manager::manager::RefreshStrategy; +use codex_protocol::auth::AuthMode; use codex_protocol::protocol::AskForApproval; use codex_protocol::user_input::UserInput; use codex_terminal_detection::TerminalName; @@ -146,9 +147,10 @@ enum Subcommand { /// Open the agent command center to browse and switch sessions. Agents(AgentsCommand), - /// Internal: forward a local TCP socket through an HTTP/3 CONNECT proxy. + /// Internal: forward SSH stdio or a local TCP socket through an HTTP/3 CONNECT proxy. #[clap(hide = true)] TcpTunnel(codex_tcp_tunnel::Args), + /// Run Codex non-interactively. #[clap(visible_alias = "e")] Exec(ExecCli), @@ -1150,9 +1152,39 @@ async fn dispatch_cli( arg0_paths, ))); } - Some(Subcommand::TcpTunnel(args)) => { + Some(Subcommand::TcpTunnel(args)) if args.auth_token_stdin => { codex_tcp_tunnel::run(args).await?; } + Some(Subcommand::TcpTunnel(args)) => { + let config = + cloud_config::load_config(&root_config_overrides, LoaderOverrides::default()) + .await?; + let permit = config.application_network_policy.acquire(&args.proxy_url)?; + let manager = AuthManager::shared_from_config_with_auth_modes( + &config, + /*enable_codex_api_key_env*/ false, + vec![AuthMode::Chatgpt, AuthMode::ChatgptAuthTokens], + ) + .await?; + anyhow::ensure!( + manager.auth_mode() == Some(AuthMode::Chatgpt), + "ChatGPT OAuth login required for SSH" + ); + let access_token = async { + let oauth = manager + .auth() + .await + .ok_or_else(|| anyhow::anyhow!("ChatGPT OAuth login required for SSH"))?; + Ok(oauth.get_token_data()?.access_token) + }; + permit + .run(codex_tcp_tunnel::run_ssh_proxy( + args, + access_token, + tokio::io::stdout(), + )) + .await??; + } Some(Subcommand::Exec(mut exec_cli)) => { reject_remote_mode_for_subcommand( root_remote.as_deref(), diff --git a/codex-rs/cli/tests/cloud_config.rs b/codex-rs/cli/tests/cloud_config.rs index 41ce93b756..902b3e181b 100644 --- a/codex-rs/cli/tests/cloud_config.rs +++ b/codex-rs/cli/tests/cloud_config.rs @@ -480,6 +480,48 @@ async fn mock_startup_bundle(server: &MockServer, status: u16, config: &str, req .await; } +#[tokio::test] +async fn tcp_tunnel_ssh_enforces_cloud_requirements_before_gateway_connection() -> Result<()> { + for (config, requirements, expected_error) in [ + ( + "", + "[application.network.domains]\n'example.com' = 'allow'", + "destination denied by application network policy", + ), + ( + "forced_login_method = 'api'", + "", + "ChatGPT OAuth login required for SSH", + ), + ( + "forced_chatgpt_workspace_id = 'different-workspace'", + "", + "ChatGPT OAuth login required for SSH", + ), + ] { + let Some(fixture) = CloudManagedConfigFixture::for_backend().await? else { + return Ok(()); + }; + mock_startup_bundle(&fixture.server, /*status*/ 200, config, requirements).await; + let output = fixture + .command(&[ + "tcp-tunnel", + "--environment-id", + "ccarenv_fixture", + "--port", + "22", + ])? + .output() + .await?; + assert!(!output.status.success()); + assert!(output.stdout.is_empty()); + let stderr = String::from_utf8(output.stderr)?; + assert!(stderr.contains(expected_error), "{stderr}"); + fixture.server.verify().await; + } + Ok(()) +} + async fn successful_json(command: &mut Command) -> Result { let output = command.output().await?; ensure!(output.status.success(), "{output:?}"); diff --git a/codex-rs/tcp-tunnel/Cargo.toml b/codex-rs/tcp-tunnel/Cargo.toml index 617ed7e067..2f2924cb50 100644 --- a/codex-rs/tcp-tunnel/Cargo.toml +++ b/codex-rs/tcp-tunnel/Cargo.toml @@ -9,6 +9,7 @@ doctest = false [dependencies] anyhow = { workspace = true } +base64 = { workspace = true } bytes = { workspace = true } clap = { workspace = true, features = ["derive"] } h3 = { git = "https://github.com/hyperium/h3", rev = "e07e69412876f7e26f026bd75a48b2704d8c8283" } diff --git a/codex-rs/tcp-tunnel/README.md b/codex-rs/tcp-tunnel/README.md index a15406566a..2027e563eb 100644 --- a/codex-rs/tcp-tunnel/README.md +++ b/codex-rs/tcp-tunnel/README.md @@ -1,6 +1,6 @@ # HTTP/3 TCP tunnel -The ordinary Codex binary includes the hidden `codex tcp-tunnel` command. It forwards a loopback TCP listener to an explicitly selected host and port through a TLS-verified HTTP/3 CONNECT proxy. The proxy URL must be an exact HTTPS origin in the supplied policy file; the target is provided separately as `host:port` or `[IPv6]:port`. +The ordinary Codex binary includes the hidden `codex tcp-tunnel` command. In listener mode, it forwards a loopback TCP listener to an explicitly selected host and port through a TLS-verified HTTP/3 CONNECT proxy. The proxy URL must be an exact HTTPS origin in the supplied policy file; the target is provided separately as `host:port` or `[IPv6]:port`. The initial bearer must be provided as a single line on standard input (`--auth-token-stdin`). With `--auth-token-updates-stdin`, later lines replace the bearer for new connections, emit `AUTH_UPDATED`, and closure of the controlling pipe stops the tunnel. With `--connect-headers-stdin`, the first line is instead a JSON list of `["x-name","value"]` pairs; token lines follow. Only non-forwarding `x-` extension headers are accepted. No header values or credentials belong in the command arguments. @@ -23,3 +23,22 @@ Version 1 terminal combinations are `startup/{failed,timeout}`, `control/{failed,invalid_input}`, and `transport/failed`. Nonterminal combinations are `connect/{failed,rejected}` and `transport/{failed,timeout,closed,draining}`. Other combinations are unknown records and must use the generic fallback. + +## OpenSSH ProxyCommand + +With `--environment-id ENV --port PORT`, `codex tcp-tunnel` opens one +HTTP/3 CONNECT and carries SSH bytes on stdin and stdout. +The defaults are the public Portals gateway and port 22. You can also specify +`--proxy-url ORIGIN --target ENV:PORT`. The gateway resolves the environment +and port. +A rejected CONNECT ends the command with its HTTP status on stderr. + +This example uses a verified host key stored under `ccarenv_example` in +`known_hosts`. + +```sshconfig +Host ccarenv_example + User dev + StrictHostKeyChecking yes + ProxyCommand /opt/codex/bin/codex tcp-tunnel --environment-id ccarenv_example --port %p +``` diff --git a/codex-rs/tcp-tunnel/src/control.rs b/codex-rs/tcp-tunnel/src/control.rs index 801658aa59..9dceb412b5 100644 --- a/codex-rs/tcp-tunnel/src/control.rs +++ b/codex-rs/tcp-tunnel/src/control.rs @@ -6,6 +6,8 @@ use anyhow::Context; use anyhow::Result; use anyhow::anyhow; use anyhow::ensure; +use base64::Engine; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; use http::HeaderMap; use http::HeaderValue; use http::header; @@ -57,21 +59,52 @@ pub(super) fn read_connect_headers(reader: &mut impl BufRead) -> Result Result> { - let mut line = Vec::new(); - std::io::Read::take(&mut *reader, (MAX_TOKEN_BYTES + 1) as u64) - .read_until(b'\n', &mut line) - .context("reading MASQUE token")?; - if line.is_empty() { - return Ok(None); - } - ensure!(line.len() <= MAX_TOKEN_BYTES, "MASQUE token is too long"); - let token = std::str::from_utf8(&line) - .context("invalid MASQUE token encoding")? - .trim(); - ensure!(!token.is_empty(), "empty MASQUE token"); - let mut auth = - HeaderValue::from_str(&format!("Bearer {token}")).context("invalid MASQUE token header")?; - auth.set_sensitive(true); - Ok(Some(auth)) +#[derive(Clone, Copy)] +pub(super) enum TunnelCredential { + Bearer, + ErsTunnelJwt, +} + +impl TunnelCredential { + pub(super) fn read(self, reader: &mut impl BufRead) -> Result> { + let mut line = Vec::new(); + std::io::Read::take(&mut *reader, (MAX_TOKEN_BYTES + 1) as u64) + .read_until(b'\n', &mut line) + .context("reading MASQUE token")?; + if line.is_empty() { + return Ok(None); + } + ensure!(line.len() <= MAX_TOKEN_BYTES, "MASQUE token is too long"); + let token = std::str::from_utf8(&line) + .context("invalid MASQUE token encoding")? + .trim(); + Ok(Some(self.header(token)?)) + } + + pub(super) fn header(self, token: &str) -> Result { + ensure!(!token.is_empty(), "empty MASQUE token"); + let mut parts = token.split('.'); + let ers = parts + .next() + .and_then(json_segment) + .is_some_and(|header| header["typ"] == "cca-tunnel+jwt"); + if ers || matches!(self, Self::ErsTunnelJwt) { + // Check the wire version locally. The gateway verifies the signature. + let claims = parts.next().and_then(json_segment); + ensure!( + ers && claims.is_some_and(|claims| claims["ver"].as_u64() == Some(2)) + && parts.next().is_some_and(|signature| !signature.is_empty()) + && parts.next().is_none(), + "Portals requires an ERS-issued v2 tunnel JWT" + ); + } + let mut auth = HeaderValue::from_str(&format!("Bearer {token}")) + .context("invalid MASQUE token header")?; + auth.set_sensitive(true); + Ok(auth) + } +} + +fn json_segment(segment: &str) -> Option { + serde_json::from_slice(&URL_SAFE_NO_PAD.decode(segment).ok()?).ok() } diff --git a/codex-rs/tcp-tunnel/src/lib.rs b/codex-rs/tcp-tunnel/src/lib.rs index e892d3716d..1c25df4d70 100644 --- a/codex-rs/tcp-tunnel/src/lib.rs +++ b/codex-rs/tcp-tunnel/src/lib.rs @@ -1,10 +1,10 @@ -//! Loopback TCP forwarding through an HTTP/3 CONNECT proxy. -//! Bearer credentials and optional proxy-specific metadata enter only via stdin. +//! TCP forwarding through a verified HTTP/3 CONNECT proxy, for listeners and SSH stdio. //! The listener survives transport loss; individual TCP streams are never replayed. use std::io::BufRead; use std::io::Write; use std::net::IpAddr; use std::net::SocketAddr; +use std::num::NonZeroU16; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; @@ -17,6 +17,7 @@ use anyhow::ensure; use bytes::Buf; use bytes::Bytes; use clap::Args as ClapArgs; +use clap::builder::ArgPredicate; use http::HeaderMap; use http::HeaderValue; use http::Method; @@ -25,10 +26,11 @@ use http::Uri; use http::header; use http::uri::Authority; use quinn::crypto::rustls::QuicClientConfig; +use tokio::io::AsyncRead; use tokio::io::AsyncReadExt; +use tokio::io::AsyncWrite; use tokio::io::AsyncWriteExt; use tokio::net::TcpListener; -use tokio::net::TcpStream; use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio::sync::watch; @@ -37,12 +39,15 @@ use url::Url; mod control; mod diagnostics; +mod ssh; + +pub use ssh::run_ssh_proxy; use diagnostics::Code; use diagnostics::Diagnostics; use diagnostics::Phase; -use control::read_auth_token; +use control::TunnelCredential; use control::read_connect_headers; const PROXY_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); @@ -51,20 +56,36 @@ const MAX_FIELD_SECTION_SIZE: u64 = 64 * 1024; #[derive(Debug, ClapArgs)] pub struct Args { - /// HTTPS origin of the HTTP/3 proxy. - #[arg(long)] - proxy_url: Url, - /// File containing one exact approved HTTPS proxy origin per line. - #[arg(long)] - proxy_origins_file: PathBuf, - /// CONNECT target authority, including a nonzero port. - #[arg(long)] - target: String, - #[arg(long, default_value = "127.0.0.1:0")] - listen_addr: SocketAddr, - /// Read the initial bearer from stdin. - #[arg(long, default_value_t = false)] - auth_token_stdin: bool, + /// HTTPS proxy origin. With --environment-id, defaults to the public Portals gateway. + #[arg( + long, + required = false, + required_unless_present = "environment_id", + default_value_if("environment_id", ArgPredicate::IsPresent, ssh::PUBLIC_GATEWAY) + )] + pub proxy_url: Url, + /// Listener mode: file containing one exact approved HTTPS proxy origin per line. + #[arg(long, requires = "auth_token_stdin")] + proxy_origins_file: Option, + /// SSH environment ID and port (ENV:PORT), or the CONNECT authority in listener mode. + #[arg( + long, + required_unless_present = "environment_id", + conflicts_with = "environment_id" + )] + target: Option, + /// Open an SSH stdio tunnel to this canonical Portals environment. + #[arg(long, conflicts_with = "auth_token_stdin")] + environment_id: Option, + /// Published SSH port for --environment-id (default 22). + #[arg(long, requires = "environment_id")] + port: Option, + /// Loopback listener address (default 127.0.0.1:0). + #[arg(long, requires = "auth_token_stdin")] + listen_addr: Option, + /// Start a listener and read its bearer from stdin. Otherwise relay SSH over stdio using the Codex login. + #[arg(long, requires = "proxy_origins_file")] + pub auth_token_stdin: bool, /// Read replacement bearers from stdin and stop when the controlling pipe closes. #[arg(long, requires = "auth_token_stdin")] auth_token_updates_stdin: bool, @@ -72,7 +93,7 @@ pub struct Args { #[arg(long, requires = "auth_token_stdin")] connect_headers_stdin: bool, /// Emit credential-safe JSON diagnostics on stderr instead of human-readable errors. - #[arg(long)] + #[arg(long, requires = "auth_token_stdin")] diagnostics_json: bool, } @@ -101,7 +122,17 @@ enum ProxyClosure { } impl ProxyTarget { - fn parse(proxy_url: &Url, trusted_origins: &[String], target: &str) -> Result { + fn request(&self, mut headers: HeaderMap, bearer: HeaderValue) -> Result> { + let mut request = Request::builder() + .method(Method::CONNECT) + .uri(Uri::builder().authority(self.authority.as_str()).build()?) + .body(())?; + headers.insert(header::AUTHORIZATION, bearer); + *request.headers_mut() = headers; + Ok(request) + } + + fn parse(proxy_url: &Url, trusted_origins: &[impl AsRef], target: &str) -> Result { ensure!(proxy_url.scheme() == "https", "proxy URL must use HTTPS"); ensure!( proxy_url.username().is_empty() @@ -111,8 +142,11 @@ impl ProxyTarget { && proxy_url.path() == "/", "proxy URL must be an HTTPS origin" ); + let origin = proxy_url.origin().ascii_serialization(); ensure!( - trusted_origins.contains(&proxy_url.origin().ascii_serialization()), + trusted_origins + .iter() + .any(|trusted| trusted.as_ref() == origin), "proxy URL origin is not trusted" ); let host = match proxy_url.host().context("proxy URL has no host")? { @@ -174,18 +208,21 @@ pub async fn run(args: Args) -> Result<()> { let mut phase = Phase::Startup; let result = async { ensure!(args.auth_token_stdin, "--auth-token-stdin is required"); + let listen_addr = args.listen_addr.unwrap_or(SocketAddr::from(([127, 0, 0, 1], 0))); ensure!( - args.listen_addr.ip().is_loopback(), + listen_addr.ip().is_loopback(), "listener must be loopback" ); - let origins = std::fs::read_to_string(&args.proxy_origins_file) + let origins = std::fs::read_to_string(args.proxy_origins_file.context("missing proxy origins file")?) .context("reading trusted proxy origins")?; let trusted_origins = parse_trusted_origins(&origins)?; - let target = ProxyTarget::parse(&args.proxy_url, &trusted_origins, &args.target)?; + let authority = args.target.as_deref().context("missing CONNECT target")?; + let target = ProxyTarget::parse(&args.proxy_url, &trusted_origins, authority)?; phase = Phase::Control; let (metadata, mut tokens) = control_input( std::io::BufReader::new(std::io::stdin()), args.connect_headers_stdin, + &args.proxy_url.origin().ascii_serialization(), )?; let connect = metadata.await.context("CONNECT metadata input closed")??; let initial_auth = tokens @@ -197,29 +234,10 @@ pub async fn run(args: Args) -> Result<()> { let (updates, auth) = watch::channel(initial_auth); let headers = TunnelHeaders { auth, connect }; - let listener = TcpListener::bind(args.listen_addr) + let listener = TcpListener::bind(listen_addr) .await .context("binding loopback listener")?; - let native = rustls_native_certs::load_native_certs(); - let mut roots = rustls::RootCertStore::empty(); - roots.add_parsable_certificates(native.certs); - ensure!( - !roots.is_empty(), - "no native TLS roots: {:?}", - native.errors - ); - let mut tls = rustls::ClientConfig::builder_with_provider(Arc::new( - rustls::crypto::ring::default_provider(), - )) - .with_safe_default_protocol_versions()? - .with_root_certificates(roots) - .with_no_client_auth(); - tls.alpn_protocols = vec![b"h3".to_vec()]; - let mut config = quinn::ClientConfig::new(Arc::new(QuicClientConfig::try_from(tls)?)); - let mut transport = quinn::TransportConfig::default(); - transport.max_idle_timeout(Some(Duration::from_secs(30).try_into()?)); - transport.keep_alive_interval(Some(Duration::from_secs(10))); - config.transport_config(Arc::new(transport)); + let config = client_config(quinn::TransportConfig::default())?; let (ready, readiness) = oneshot::channel(); let tunnel = async move { let connected = ProxyConnection::connect(&target, &config).await?; @@ -271,7 +289,9 @@ async fn update_auth_tokens( fn control_input( mut reader: impl BufRead + Send + 'static, connect_headers_stdin: bool, + proxy_origin: &str, ) -> Result<(oneshot::Receiver>, TokenStream)> { + let portals = ssh::GATEWAY_ORIGINS.contains(&proxy_origin); let (metadata, initial_metadata) = oneshot::channel(); let (send, tokens) = mpsc::channel(1); // Tokio's blocking stdin can prevent runtime shutdown after a proxy startup failure. @@ -283,13 +303,25 @@ fn control_input( } else { Ok(HeaderMap::new()) }; - let valid = headers.is_ok(); - let _ = metadata.send(headers); - if !valid { - return; - } + let headers = match headers { + Ok(headers) => headers, + Err(error) => { + let _ = metadata.send(Err(error)); + return; + } + }; + let credential = if portals + && headers + .get("x-portals-auth-mode") + .is_none_or(|mode| mode != "oauth") + { + TunnelCredential::ErsTunnelJwt + } else { + TunnelCredential::Bearer + }; + let _ = metadata.send(Ok(headers)); loop { - let token = read_auth_token(&mut reader); + let token = credential.read(&mut reader); let finished = !matches!(&token, Ok(Some(_))); if send.blocking_send(token).is_err() || finished { break; @@ -300,6 +332,29 @@ fn control_input( Ok((initial_metadata, tokens)) } +fn client_config(mut transport: quinn::TransportConfig) -> Result { + let native = rustls_native_certs::load_native_certs(); + let mut roots = rustls::RootCertStore::empty(); + roots.add_parsable_certificates(native.certs); + ensure!( + !roots.is_empty(), + "no native TLS roots: {:?}", + native.errors + ); + let mut tls = rustls::ClientConfig::builder_with_provider(Arc::new( + rustls::crypto::ring::default_provider(), + )) + .with_safe_default_protocol_versions()? + .with_root_certificates(roots) + .with_no_client_auth(); + tls.alpn_protocols = vec![b"h3".to_vec()]; + let mut config = quinn::ClientConfig::new(Arc::new(QuicClientConfig::try_from(tls)?)); + transport.max_idle_timeout(Some(Duration::from_secs(30).try_into()?)); + transport.keep_alive_interval(Some(Duration::from_secs(10))); + config.transport_config(Arc::new(transport)); + Ok(config) +} + impl ProxyConnection { async fn connect(target: &ProxyTarget, config: &quinn::ClientConfig) -> Result { tokio::time::timeout(PROXY_CONNECT_TIMEOUT, async { @@ -391,13 +446,10 @@ async fn serve( let (socket, _) = incoming?; let mut sender = sender.clone(); let draining = draining.clone(); - let mut request = Request::builder().method(Method::CONNECT) - .uri(Uri::builder().authority(target.authority.as_str()).build()?) - .body(())?; - *request.headers_mut() = headers.connect.clone(); - request.headers_mut().insert(header::AUTHORIZATION, headers.auth.borrow().clone()); + let request = target.request(headers.connect.clone(), headers.auth.borrow().clone())?; tokio::spawn(async move { - if let Err(error) = bridge(socket, &mut sender, request, draining).await { + let (read, write) = socket.into_split(); + if let Err(error) = bridge(read, write, &mut sender, request, BridgeMode::Listener(draining)).await { diagnostics.report( Phase::Connect, &error, @@ -480,16 +532,24 @@ async fn serve( } } +enum BridgeMode { + Listener(mpsc::Sender<()>), + ProxyCommand, +} + async fn bridge( - mut socket: TcpStream, + mut local_read: impl AsyncRead + Unpin, + mut local_write: impl AsyncWrite + Unpin, sender: &mut h3::client::SendRequest, request: Request<()>, - draining: mpsc::Sender<()>, + mode: BridgeMode, ) -> Result<()> { let mut stream = match sender.send_request(request).await { Ok(stream) => stream, Err(error) => { - if matches!(error, h3::error::StreamError::RemoteClosing { .. }) { + if let BridgeMode::Listener(draining) = &mode + && matches!(error, h3::error::StreamError::RemoteClosing { .. }) + { let _ = draining.try_send(()); } return Err(error).context("sending CONNECT"); @@ -504,7 +564,6 @@ async fn bridge( return Err(Code::Rejected(status).into()); } let (mut send, mut recv) = stream.split(); - let (mut local_read, mut local_write) = socket.split(); let upload = async { let mut buffer = vec![0; 64 * 1024]; loop { @@ -524,10 +583,20 @@ async fn bridge( local_write.write_all(chunk).await?; data.advance(chunk.len()); } + local_write.flush().await?; } local_write.shutdown().await?; Ok::<_, anyhow::Error>(()) }; - tokio::try_join!(upload, download)?; - Ok(()) + tokio::pin!(upload, download); + tokio::select! { + result = &mut upload => { result?; download.await } + result = &mut download => { + result?; + match mode { + BridgeMode::Listener(_) => upload.await, + BridgeMode::ProxyCommand => Ok(()), + } + } + } } diff --git a/codex-rs/tcp-tunnel/src/lib_tests.rs b/codex-rs/tcp-tunnel/src/lib_tests.rs index 7243084dd8..03e1654441 100644 --- a/codex-rs/tcp-tunnel/src/lib_tests.rs +++ b/codex-rs/tcp-tunnel/src/lib_tests.rs @@ -110,7 +110,11 @@ async fn control_input_acknowledges_bearer_renewal_with_or_without_metadata() -> true, ), ] { - let (metadata, mut tokens) = control_input(Cursor::new(input), includes_metadata)?; + let (metadata, mut tokens) = control_input( + Cursor::new(input), + includes_metadata, + "https://proxy.example", + )?; let metadata = metadata.await??; if includes_metadata { assert_eq!(metadata["x-test-route"], "private-value"); @@ -133,9 +137,89 @@ async fn control_input_acknowledges_bearer_renewal_with_or_without_metadata() -> Ok(()) } +#[tokio::test] +async fn portals_jwt_listener_requires_v2_at_startup_and_on_renewal() -> Result<()> { + // Inert parser fixtures with a placeholder signature. No signing key or live credential. + const HEADER: &str = "eyJ0eXAiOiJjY2EtdHVubmVsK2p3dCJ9"; + let v1 = format!("{HEADER}.eyJ2ZXIiOjF9.fixture"); + let v2 = format!("{HEADER}.eyJ2ZXIiOjJ9.fixture"); + let origin = super::ssh::PUBLIC_GATEWAY; + for invalid in [ + &v1[..], + "opaque", + "malformed.parts", + &format!("{HEADER}.e30.fixture"), + ] { + let (_, mut tokens) = control_input( + Cursor::new(invalid.to_owned()), + /*connect_headers_stdin*/ false, + origin, + )?; + let error = tokens.recv().await.unwrap().unwrap_err(); + assert_eq!( + error.to_string(), + "Portals requires an ERS-issued v2 tunnel JWT" + ); + } + + let (_, mut tokens) = control_input( + Cursor::new(format!("{v2}\n{v2}\n{v1}\n")), + /*connect_headers_stdin*/ false, + origin, + )?; + let initial = tokens.recv().await.unwrap()?.unwrap(); + assert_eq!(initial.to_str()?, format!("Bearer {v2}")); + let (updates, latest) = watch::channel(initial.clone()); + let (ready, readiness) = oneshot::channel(); + ready.send(()).unwrap(); + let mut output = Vec::new(); + let error = update_auth_tokens(&mut tokens, updates, readiness, &mut output) + .await + .unwrap_err(); + assert_eq!( + error.to_string(), + "Portals requires an ERS-issued v2 tunnel JWT" + ); + assert_eq!(output, b"AUTH_UPDATED\n"); + assert_eq!(*latest.borrow(), initial); + Ok(()) +} + +#[tokio::test] +async fn other_bearer_modes_reject_ers_v1_and_accept_unrelated_jwts() -> Result<()> { + let unrelated = "eyJ0eXAiOiJKV1QifQ.eyJ2ZXIiOjF9.fixture"; + let ers_v1 = "eyJ0eXAiOiJjY2EtdHVubmVsK2p3dCJ9.eyJ2ZXIiOjF9.fixture"; + for (origin, metadata) in [ + ( + super::ssh::PUBLIC_GATEWAY, + r#"[["x-portals-auth-mode","oauth"]]"#, + ), + ("https://proxy.example", "[]"), + ] { + let (headers, mut tokens) = control_input( + Cursor::new(format!("{metadata}\n{unrelated}\n{ers_v1}\n")), + /*connect_headers_stdin*/ true, + origin, + )?; + headers.await??; + let initial = tokens.recv().await.unwrap()?.unwrap(); + assert_eq!(initial.to_str()?, format!("Bearer {unrelated}")); + let error = tokens.recv().await.unwrap().unwrap_err(); + assert_eq!( + error.to_string(), + "Portals requires an ERS-issued v2 tunnel JWT" + ); + } + Ok(()) +} + #[tokio::test] async fn closed_parent_pipe_finishes_even_before_proxy_readiness() -> Result<()> { - let (_, mut tokens) = control_input(Cursor::new([]), /*connect_headers_stdin*/ false)?; + let (_, mut tokens) = control_input( + Cursor::new([]), + /*connect_headers_stdin*/ false, + "https://proxy.example", + )?; let (updates, _) = watch::channel("Bearer initial".parse()?); let (_ready, readiness) = oneshot::channel(); let mut output = Vec::new(); @@ -185,8 +269,11 @@ async fn invalid_control_input_stops_credentials_without_exposing_secrets() -> R for &(case, input) in invalid_metadata { let mut input = input.to_vec(); input.extend_from_slice(b"private-secret\n"); - let (metadata, mut tokens) = - control_input(Cursor::new(input), /*connect_headers_stdin*/ true)?; + let (metadata, mut tokens) = control_input( + Cursor::new(input), + /*connect_headers_stdin*/ true, + "https://proxy.example", + )?; let error = metadata.await?.expect_err(case); assert!(!format!("{error:#}").contains("private-"), "{case}"); assert!(tokens.recv().await.is_none(), "{case}"); @@ -196,8 +283,11 @@ async fn invalid_control_input_stops_credentials_without_exposing_secrets() -> R ("empty bearer", b"\n".to_vec()), ("bearer limit", vec![b'x'; MAX_TOKEN_BYTES + 1]), ] { - let (metadata, mut tokens) = - control_input(Cursor::new(input), /*connect_headers_stdin*/ false)?; + let (metadata, mut tokens) = control_input( + Cursor::new(input), + /*connect_headers_stdin*/ false, + "https://proxy.example", + )?; assert!(metadata.await??.is_empty(), "{case}"); let error = tokens.recv().await.expect(case).expect_err(case); assert!(!format!("{error:#}").contains("private-"), "{case}"); @@ -206,35 +296,127 @@ async fn invalid_control_input_stops_credentials_without_exposing_secrets() -> R Ok(()) } +fn local_proxy() -> Result<(quinn::Endpoint, quinn::ClientConfig)> { + let CertifiedKey { cert, signing_key } = + generate_simple_self_signed(vec!["127.0.0.1".to_string()])?; + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let mut server_tls = rustls::ServerConfig::builder_with_provider(provider.clone()) + .with_safe_default_protocol_versions()? + .with_no_client_auth() + .with_single_cert( + vec![cert.der().clone()], + PrivatePkcs8KeyDer::from(signing_key.serialize_der()).into(), + )?; + server_tls.alpn_protocols = vec![b"h3".to_vec()]; + let server_config = quinn::ServerConfig::with_crypto(Arc::new( + quinn::crypto::rustls::QuicServerConfig::try_from(server_tls)?, + )); + let endpoint = quinn::Endpoint::server(server_config, "127.0.0.1:0".parse()?)?; + + let mut roots = rustls::RootCertStore::empty(); + roots.add(cert.der().clone())?; + let mut client_tls = rustls::ClientConfig::builder_with_provider(provider) + .with_safe_default_protocol_versions()? + .with_root_certificates(roots) + .with_no_client_auth(); + client_tls.alpn_protocols = vec![b"h3".to_vec()]; + let config = quinn::ClientConfig::new(Arc::new( + quinn::crypto::rustls::QuicClientConfig::try_from(client_tls)?, + )); + + Ok((endpoint, config)) +} + +#[tokio::test] +async fn stdio_propagates_half_close_and_exits_on_gateway_eof_or_rejection() -> Result<()> { + let (endpoint, config) = local_proxy()?; + let address = endpoint.local_addr()?; + let target = ProxyTarget { + host: address.ip().to_string(), + port: address.port(), + authority: address.to_string(), + }; + let (done, keep_alive) = oneshot::channel(); + let server = tokio::spawn(async move { + let quic = endpoint.accept().await.unwrap().await?; + let mut http = h3::server::Connection::new(h3_quinn::Connection::new(quic)).await?; + for (index, status) in [200, 200, 401].into_iter().enumerate() { + let (_, mut stream) = http.accept().await?.unwrap().resolve_request().await?; + stream + .send_response(Response::builder().status(status).body(())?) + .await?; + if status == 200 { + stream.send_data(Bytes::from_static(b"\0\xff")).await?; + } + // First request must forward the final bytes and FIN before reading the reply. + // The next request sees remote EOF with its own input still open. + if index == 0 { + let mut data = stream.recv_data().await?.unwrap(); + let size = data.remaining(); + assert_eq!(data.copy_to_bytes(size), Bytes::from_static(b"request")); + assert!(stream.recv_data().await?.is_none()); + stream.send_data(Bytes::from_static(b"after FIN")).await?; + } + stream.finish().await?; + } + let _ = keep_alive.await; + anyhow::Ok(()) + }); + let ProxyConnection { + endpoint: _endpoint, + quic: _quic, + mut connection, + mut sender, + } = ProxyConnection::connect(&target, &config).await?; + let driver = tokio::spawn(async move { connection.wait_idle().await }); + for index in 0..3 { + let request = target.request(http::HeaderMap::new(), "Bearer fixture".parse()?)?; + let (mut input, reader) = tokio::io::duplex(16); + if index == 0 { + input.write_all(b"request").await?; + input.shutdown().await?; + } + let mut output = Vec::new(); + let result = timeout( + Duration::from_secs(3), + super::bridge( + reader, + &mut output, + &mut sender, + request, + super::BridgeMode::ProxyCommand, + ), + ) + .await?; + match index { + 0 => { + result?; + assert_eq!(output, b"\0\xffafter FIN"); + } + 1 => { + result?; + assert_eq!(output, b"\0\xff"); + } + _ => { + assert_eq!( + result.unwrap_err().to_string(), + "MASQUE CONNECT rejected with status 401" + ); + assert!(output.is_empty()); + } + } + } + done.send(()).unwrap(); + server.await??; + driver.abort(); + Ok(()) +} + #[tokio::test] async fn connect_rejects_qpack_expansion_and_keeps_connection_usable() -> Result<()> { timeout(Duration::from_secs(15), async { - let CertifiedKey { cert, signing_key } = - generate_simple_self_signed(vec!["127.0.0.1".to_string()])?; - let provider = Arc::new(rustls::crypto::ring::default_provider()); - let mut server_tls = rustls::ServerConfig::builder_with_provider(provider.clone()) - .with_safe_default_protocol_versions()? - .with_no_client_auth() - .with_single_cert( - vec![cert.der().clone()], - PrivatePkcs8KeyDer::from(signing_key.serialize_der()).into(), - )?; - server_tls.alpn_protocols = vec![b"h3".to_vec()]; - let server_config = quinn::ServerConfig::with_crypto(Arc::new( - quinn::crypto::rustls::QuicServerConfig::try_from(server_tls)?, - )); - let endpoint = quinn::Endpoint::server(server_config, "127.0.0.1:0".parse()?)?; + let (endpoint, config) = local_proxy()?; let address = endpoint.local_addr()?; - let mut roots = rustls::RootCertStore::empty(); - roots.add(cert.der().clone())?; - let mut client_tls = rustls::ClientConfig::builder_with_provider(provider) - .with_safe_default_protocol_versions()? - .with_root_certificates(roots) - .with_no_client_auth(); - client_tls.alpn_protocols = vec![b"h3".to_vec()]; - let config = quinn::ClientConfig::new(Arc::new( - quinn::crypto::rustls::QuicClientConfig::try_from(client_tls)?, - )); let (done_tx, done_rx) = oneshot::channel(); let server = tokio::spawn(async move { @@ -299,34 +481,9 @@ async fn connect_rejects_qpack_expansion_and_keeps_connection_usable() -> Result #[tokio::test] async fn transport_reconnect_keeps_listener_and_does_not_replay_old_tcp_streams() -> Result<()> { - let CertifiedKey { cert, signing_key } = - generate_simple_self_signed(vec!["127.0.0.1".to_string()])?; - let provider = Arc::new(rustls::crypto::ring::default_provider()); - let mut server_tls = rustls::ServerConfig::builder_with_provider(provider.clone()) - .with_safe_default_protocol_versions()? - .with_no_client_auth() - .with_single_cert( - vec![cert.der().clone()], - PrivatePkcs8KeyDer::from(signing_key.serialize_der()).into(), - )?; - server_tls.alpn_protocols = vec![b"h3".to_vec()]; - let server_config = quinn::ServerConfig::with_crypto(Arc::new( - quinn::crypto::rustls::QuicServerConfig::try_from(server_tls)?, - )); - let endpoint = quinn::Endpoint::server(server_config, "127.0.0.1:0".parse()?)?; + let (endpoint, config) = local_proxy()?; let proxy_address = endpoint.local_addr()?; - let mut roots = rustls::RootCertStore::empty(); - roots.add(cert.der().clone())?; - let mut client_tls = rustls::ClientConfig::builder_with_provider(provider) - .with_safe_default_protocol_versions()? - .with_root_certificates(roots) - .with_no_client_auth(); - client_tls.alpn_protocols = vec![b"h3".to_vec()]; - let config = quinn::ClientConfig::new(Arc::new( - quinn::crypto::rustls::QuicClientConfig::try_from(client_tls)?, - )); - let request_count = Arc::new(AtomicUsize::new(0)); let server_request_count = request_count.clone(); let (accepted, mut connections) = mpsc::unbounded_channel(); diff --git a/codex-rs/tcp-tunnel/src/ssh.rs b/codex-rs/tcp-tunnel/src/ssh.rs new file mode 100644 index 0000000000..52ea22abdc --- /dev/null +++ b/codex-rs/tcp-tunnel/src/ssh.rs @@ -0,0 +1,152 @@ +//! Single-stream OpenSSH ProxyCommand. Only the verified gateway receives credentials. +//! OpenSSH owns SSH authentication, host-key verification, and all stdio bytes. +use std::future::Future; +use std::io::Read; +use std::num::NonZeroU16; + +use anyhow::Context; +use anyhow::Result; +use anyhow::ensure; +use http::HeaderMap; +use http::HeaderValue; +use tokio::io::AsyncWrite; +use tokio::io::AsyncWriteExt; + +use super::Args; +use super::BridgeMode; +use super::Code; +use super::ProxyConnection; +use super::ProxyTarget; +use super::bridge; +use super::client_config; +use super::control::TunnelCredential; + +// Only gateways shipped with the CLI may receive the signed-in account's credential. +pub(super) const PUBLIC_GATEWAY: &str = "https://codex-portals.api.openai.com"; +pub(super) const GATEWAY_ORIGINS: &[&str] = &[PUBLIC_GATEWAY]; + +impl Args { + fn ssh_destination(&self) -> Result<(ProxyTarget, HeaderMap)> { + let (environment_id, port) = match self.environment_id.as_deref() { + Some(id) => (id, self.port.map_or(/*default*/ 22, NonZeroU16::get)), + None => { + let (id, port) = self + .target + .as_deref() + .and_then(|target| target.rsplit_once(':')) + .context("SSH target must be ENV:PORT")?; + ( + id, + port.parse::() + .context("invalid SSH target port")? + .get(), + ) + } + }; + ensure!( + (1..=256).contains(&environment_id.len()) + && environment_id + .bytes() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, b'-' | b'_')), + "invalid environment ID" + ); + let host = self.proxy_url.host_str().context("missing gateway host")?; + let authority = format!("{host}:{}", self.proxy_url.port().unwrap_or(443)); + let target = ProxyTarget::parse(&self.proxy_url, GATEWAY_ORIGINS, &authority)?; + let mut headers = HeaderMap::new(); + headers.insert("x-portals-auth-mode", HeaderValue::from_static("oauth")); + headers.insert( + "x-portals-environment-id", + HeaderValue::from_str(environment_id)?, + ); + headers.insert("x-portals-port", port.into()); + Ok((target, headers)) + } +} + +fn oauth_bearer(token: &str) -> Result { + let unpadded = token.trim_end_matches('='); + ensure!( + !unpadded.is_empty() + && token.len() < super::control::MAX_TOKEN_BYTES + && unpadded + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"-._~+/".contains(&b)), + "invalid OAuth access token" + ); + TunnelCredential::Bearer.header(token) +} + +/// Relay one SSH session. Poll the active account's token only after verifying the gateway. +pub async fn run_ssh_proxy( + args: Args, + access_token: impl Future>, + stdout: impl AsyncWrite + Unpin, +) -> Result<()> { + let (target, headers) = args.ssh_destination()?; + // Linux can advertise UDP segmentation on paths where batched packets are lost. + // Single datagrams keep bulk SSH transfers moving on those paths. + let mut transport = quinn::TransportConfig::default(); + transport.enable_segmentation_offload(!cfg!(target_os = "linux")); + let ProxyConnection { + endpoint: _endpoint, + quic: _quic, + mut connection, + mut sender, + } = ProxyConnection::connect(&target, &client_config(transport)?).await?; + let request = target.request(headers, oauth_bearer(&access_token.await?)?)?; + + // OpenSSH needs EOF when the peer finishes, even if it still holds stdin open. + // A detached blocking reader lets the Tokio runtime exit in that case. + let (mut input, reader) = tokio::io::duplex(64 * 1024); + let runtime = tokio::runtime::Handle::current(); + let (outcome, input_done) = tokio::sync::oneshot::channel(); + std::thread::Builder::new() + .name("ssh-stdin".into()) + .spawn(move || { + let mut stdin = std::io::stdin().lock(); + let mut buf = [0; 16 * 1024]; + let result = loop { + match stdin.read(&mut buf) { + Ok(0) => break Ok(()), + Ok(n) => { + if let Err(err) = runtime.block_on(input.write_all(&buf[..n])) { + break Err(err); + } + } + Err(err) if err.kind() == std::io::ErrorKind::Interrupted => continue, + Err(err) => break Err(err), + } + }; + let _ = outcome.send(result); + })?; + let relay = bridge( + reader, + stdout, + &mut sender, + request, + BridgeMode::ProxyCommand, + ); + let forward = async { + tokio::pin!(relay); + tokio::select! { + result = &mut relay => result, + result = input_done => { result??; relay.await } + } + }; + tokio::pin!(forward); + let result = tokio::select! { + result = &mut forward => result, + // A closing connection may still have a complete status or payload buffered. + _ = connection.wait_idle() => forward.await, + }; + // A peer's close reason can contain arbitrary text. Only surface the HTTP status. + result.map_err(|err| match err.downcast_ref::() { + Some(Code::Rejected(status)) => Code::Rejected(*status).into(), + _ => anyhow::anyhow!("SSH tunnel stream failed"), + }) +} + +#[cfg(test)] +#[path = "ssh_tests.rs"] +mod tests; diff --git a/codex-rs/tcp-tunnel/src/ssh_tests.rs b/codex-rs/tcp-tunnel/src/ssh_tests.rs new file mode 100644 index 0000000000..690c2771e6 --- /dev/null +++ b/codex-rs/tcp-tunnel/src/ssh_tests.rs @@ -0,0 +1,135 @@ +#![cfg(test)] + +//! Exercise route selection, credential boundaries, and the native OAuth CONNECT request. +use super::Args; +use super::PUBLIC_GATEWAY; +use super::oauth_bearer; +use super::run_ssh_proxy; +use clap::Parser; +use http::HeaderMap; +use http::Method; +use pretty_assertions::assert_eq; + +#[derive(Parser)] +struct Cli { + #[command(flatten)] + args: Args, +} + +fn tcp_tunnel_args<'a>(gateway: &'a str, target: &'a str) -> [&'a str; 5] { + ["tcp-tunnel", "--proxy-url", gateway, "--target", target] +} + +#[test] +fn listener_options_require_the_separate_stdin_control_protocol() { + let ssh = tcp_tunnel_args(PUBLIC_GATEWAY, "ccarenv_test:22"); + assert!(!Cli::try_parse_from(ssh).unwrap().args.auth_token_stdin); + for incomplete in [ + &["--auth-token-stdin"][..], + &["--proxy-origins-file", "origins"], + &["--listen-addr", "127.0.0.1:0"], + &["--connect-headers-stdin"], + &["--auth-token-updates-stdin"], + &["--diagnostics-json"], + ] { + assert!(Cli::try_parse_from(ssh.into_iter().chain(incomplete.iter().copied())).is_err()); + } +} + +#[test] +fn connect_routes_by_gateway_authority_and_published_port() -> anyhow::Result<()> { + let args = Cli::try_parse_from(tcp_tunnel_args(PUBLIC_GATEWAY, "ccarenv_test-1:2222"))?.args; + let named = Cli::try_parse_from([ + "tcp-tunnel", + "--environment-id", + "ccarenv_test-1", + "--port", + "2222", + ])? + .args; + let (target, metadata) = named.ssh_destination()?; + let (explicit_target, explicit_headers) = args.ssh_destination()?; + assert_eq!(target, explicit_target); + assert_eq!(metadata, explicit_headers); + let request = target.request(metadata, oauth_bearer("local-fixture==")?)?; + assert!(request.headers()[http::header::AUTHORIZATION].is_sensitive()); + assert_eq!(request.method(), Method::CONNECT); + assert_eq!( + request.uri().to_string(), + "codex-portals.api.openai.com:443" + ); + let expected = HeaderMap::from_iter([ + ( + http::header::AUTHORIZATION, + "Bearer local-fixture==".parse()?, + ), + ("x-portals-auth-mode".parse()?, "oauth".parse()?), + ( + "x-portals-environment-id".parse()?, + "ccarenv_test-1".parse()?, + ), + ("x-portals-port".parse()?, "2222".parse()?), + ]); + assert_eq!(request.headers(), &expected); + Ok(()) +} + +#[tokio::test] +async fn invalid_routes_fail_before_reading_oauth_or_writing_stdout() -> anyhow::Result<()> { + for (url, target) in [ + (PUBLIC_GATEWAY, ":22"), + (PUBLIC_GATEWAY, "env"), + (PUBLIC_GATEWAY, "env:0"), + (PUBLIC_GATEWAY, "env:65536"), + (PUBLIC_GATEWAY, "id\r\nother: value:22"), + ("https://codex-portals.api.openai.com.bad", "valid:22"), + ("https://codex-portals.api.openai.com:4443", "valid:22"), + ] { + let args = Cli::try_parse_from(tcp_tunnel_args(url, target))?.args; + let mut out = Vec::new(); + let result = run_ssh_proxy( + args, + async { panic!("unexpected credential lookup") }, + &mut out, + ) + .await; + assert!(result.is_err()); + assert!(out.is_empty()); + } + Ok(()) +} + +#[test] +fn environment_id_defaults_to_ssh_and_rejects_ambiguous_routes() -> anyhow::Result<()> { + let named = ["tcp-tunnel", "--environment-id", "ccarenv_test"]; + assert_eq!( + Cli::try_parse_from(named)?.args.ssh_destination()?.1["x-portals-port"], + "22" + ); + for extra in [ + &["--port", "0"][..], + &["--port", "65536"], + &["--target", "other:22"], + &["--auth-token-stdin", "--proxy-origins-file", "origins"], + ] { + assert!(Cli::try_parse_from(named.into_iter().chain(extra.iter().copied())).is_err()); + } + assert!(Cli::try_parse_from(["tcp-tunnel", "--target", "env:22"]).is_err()); + Ok(()) +} + +#[test] +fn rejects_invalid_oauth_credentials_without_echoing_them() { + for token in [ + "", + "=", + "private=secret", + "private\r\nvalue", + "private secret", + ] { + assert_eq!( + oauth_bearer(token).unwrap_err().to_string(), + "invalid OAuth access token" + ); + } +}