Add an OpenSSH ProxyCommand mode to codex tcp-tunnel (#60568)

## Why

The existing loopback listener mode does not provide the stdin/stdout transport needed by OpenSSH's `ProxyCommand`.

## What changed

- Relay one SSH session over HTTP/3 CONNECT with `--environment-id ENV --port PORT`, defaulting to the public Portals gateway and port 22. Also accept an explicit `--proxy-url` with `--target ENV:PORT`.
- Require ChatGPT OAuth login, enforce cloud configuration and application network policy, and restrict OAuth credentials to the built-in gateway origin. Leave SSH authentication and host-key verification to OpenSSH.
- Preserve binary stdio, propagate input half-close, exit on gateway EOF, and report rejected CONNECT requests by HTTP status without exposing arbitrary peer error text.
- Require ERS-issued v2 tunnel JWTs for Portals listener mode unless OAuth mode is selected, and reject older ERS tunnel JWTs during credential updates.
- Disable UDP segmentation offload for SSH tunnels on Linux to avoid stalled bulk transfers on affected paths. Document an OpenSSH configuration example.

## Testing

Add coverage for argument validation, gateway routing, credential boundaries, JWT startup and renewal checks, binary relay and half-close behavior, gateway EOF and rejection, and cloud policy enforcement before gateway connection.

GitOrigin-RevId: 0d9afbddef01748efcc1c4b88f7e19923deff7e1

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",

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(),

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<Value> {
     let output = command.output().await?;
     ensure!(output.status.success(), "{output:?}");

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" }

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
+```

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<HeaderMa
     Ok(headers)
 }
 
-pub(super) fn read_auth_token(reader: &mut impl BufRead) -> Result<Option<HeaderValue>> {
-    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<Option<HeaderValue>> {
+        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<HeaderValue> {
+        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::Value> {
+    serde_json::from_slice(&URL_SAFE_NO_PAD.decode(segment).ok()?).ok()
 }

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<PathBuf>,
+    /// 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<String>,
+    /// Open an SSH stdio tunnel to this canonical Portals environment.
+    #[arg(long, conflicts_with = "auth_token_stdin")]
+    environment_id: Option<String>,
+    /// Published SSH port for --environment-id (default 22).
+    #[arg(long, requires = "environment_id")]
+    port: Option<NonZeroU16>,
+    /// Loopback listener address (default 127.0.0.1:0).
+    #[arg(long, requires = "auth_token_stdin")]
+    listen_addr: Option<SocketAddr>,
+    /// 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<Self> {
+    fn request(&self, mut headers: HeaderMap, bearer: HeaderValue) -> Result<Request<()>> {
+        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<str>], target: &str) -> Result<Self> {
         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<Result<HeaderMap>>, 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<quinn::ClientConfig> {
+    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<Self> {
         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<h3_quinn::OpenStreams, Bytes>,
     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(()),
+            }
+        }
+    }
 }

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();

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::<NonZeroU16>()
+                        .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<HeaderValue> {
+    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<Output = Result<String>>,
+    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::<Code>() {
+        Some(Code::Rejected(status)) => Code::Rejected(*status).into(),
+        _ => anyhow::anyhow!("SSH tunnel stream failed"),
+    })
+}
+
+#[cfg(test)]
+#[path = "ssh_tests.rs"]
+mod tests;

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"
+        );
+    }
+}