Require OAuth CONNECT metadata for Portals TCP tunnels (#61428)

## What changed

Require `x-portals-auth-mode: oauth` in CONNECT metadata when the TCP tunnel listener connects to a known Portals gateway. Reject missing or non-OAuth metadata before reading credentials.

Remove local tunnel JWT type and version checks and treat tokens as opaque bearer credentials. Drop the unused `base64` dependency from `codex-tcp-tunnel`.

## Testing

Replace JWT validation tests with coverage that rejects missing OAuth metadata and accepts OAuth metadata with an opaque access token.

GitOrigin-RevId: 746816f0227280524c019dd90df203b182934751

codex-rs/Cargo.lock

@@ -5126,7 +5126,6 @@ name = "codex-tcp-tunnel"
 version = "0.0.0"
 dependencies = [
  "anyhow",
- "base64 0.22.1",
  "bytes",
  "clap",
  "h3",

codex-rs/tcp-tunnel/Cargo.toml

@@ -9,7 +9,6 @@ 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/src/control.rs

@@ -6,8 +6,6 @@ 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;
@@ -59,52 +57,25 @@ pub(super) fn read_connect_headers(reader: &mut impl BufRead) -> Result<HeaderMa
     Ok(headers)
 }
 
-#[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| matches!(claims["ver"].as_u64(), Some(1 | 2)))
-                    && parts.next().is_some_and(|signature| !signature.is_empty())
-                    && parts.next().is_none(),
-                "Portals requires an ERS-issued v1 or v2 tunnel JWT"
-            );
-        }
-        let mut auth = HeaderValue::from_str(&format!("Bearer {token}"))
-            .context("invalid MASQUE token header")?;
-        auth.set_sensitive(true);
-        Ok(auth)
+pub(super) fn read_bearer(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(bearer_header(token)?))
 }
 
-fn json_segment(segment: &str) -> Option<serde_json::Value> {
-    serde_json::from_slice(&URL_SAFE_NO_PAD.decode(segment).ok()?).ok()
+pub(super) fn bearer_header(token: &str) -> Result<HeaderValue> {
+    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(auth)
 }

codex-rs/tcp-tunnel/src/lib.rs

@@ -47,7 +47,7 @@ use diagnostics::Code;
 use diagnostics::Diagnostics;
 use diagnostics::Phase;
 
-use control::TunnelCredential;
+use control::read_bearer;
 use control::read_connect_headers;
 
 const PROXY_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
@@ -222,7 +222,7 @@ pub async fn run(args: Args) -> Result<()> {
         let (metadata, mut tokens) = control_input(
             std::io::BufReader::new(std::io::stdin()),
             args.connect_headers_stdin,
-            &args.proxy_url.origin().ascii_serialization(),
+            ssh::GATEWAY_ORIGINS.contains(&args.proxy_url.origin().ascii_serialization().as_str()),
         )?;
         let connect = metadata.await.context("CONNECT metadata input closed")??;
         let initial_auth = tokens
@@ -289,9 +289,8 @@ async fn update_auth_tokens(
 fn control_input(
     mut reader: impl BufRead + Send + 'static,
     connect_headers_stdin: bool,
-    proxy_origin: &str,
+    require_portals_oauth: bool,
 ) -> 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.
@@ -310,18 +309,17 @@ fn control_input(
                     return;
                 }
             };
-            let credential = if portals
+            if require_portals_oauth
                 && headers
                     .get("x-portals-auth-mode")
                     .is_none_or(|mode| mode != "oauth")
             {
-                TunnelCredential::ErsTunnelJwt
-            } else {
-                TunnelCredential::Bearer
-            };
+                let _ = metadata.send(Err(anyhow!("Portals requires OAuth CONNECT metadata")));
+                return;
+            }
             let _ = metadata.send(Ok(headers));
             loop {
-                let token = credential.read(&mut reader);
+                let token = read_bearer(&mut reader);
                 let finished = !matches!(&token, Ok(Some(_)));
                 if send.blocking_send(token).is_err() || finished {
                     break;

codex-rs/tcp-tunnel/src/lib_tests.rs

@@ -113,7 +113,7 @@ async fn control_input_acknowledges_bearer_renewal_with_or_without_metadata() ->
         let (metadata, mut tokens) = control_input(
             Cursor::new(input),
             includes_metadata,
-            "https://proxy.example",
+            /*require_portals_oauth*/ false,
         )?;
         let metadata = metadata.await??;
         if includes_metadata {
@@ -138,125 +138,28 @@ async fn control_input_acknowledges_bearer_renewal_with_or_without_metadata() ->
 }
 
 #[tokio::test]
-async fn portals_jwt_listener_rejects_invalid_credentials() -> 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 v3 = format!("{HEADER}.eyJ2ZXIiOjN9.fixture");
-    let origin = super::ssh::PUBLIC_GATEWAY;
-    for invalid in [
-        &v3[..],
-        "opaque",
-        "malformed.parts",
-        &format!("{HEADER}.e30.fixture"),
-        "eyJ0eXAiOiJKV1QifQ.eyJ2ZXIiOjF9.fixture",
-        // String versions and malformed claims.
-        &format!("{HEADER}.eyJ2ZXIiOiIxIn0.fixture"),
-        &format!("{HEADER}.eyJ2ZXIiOiIyIn0.fixture"),
-        &format!("{HEADER}.invalid-json.fixture"),
-        // Missing or empty signatures and extra segments, for both versions.
-        &format!("{HEADER}.eyJ2ZXIiOjF9"),
-        &format!("{HEADER}.eyJ2ZXIiOjJ9"),
-        &format!("{HEADER}.eyJ2ZXIiOjF9."),
-        &format!("{HEADER}.eyJ2ZXIiOjJ9."),
-        &format!("{v1}.extra"),
-        &format!("{v2}.extra"),
-    ] {
-        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 v1 or v2 tunnel JWT"
-        );
-        assert!(tokens.recv().await.is_none());
-    }
-
-    let (_, mut tokens) = control_input(
-        Cursor::new(format!("{v2}\n{v2}\n{v3}\n")),
+async fn portals_listener_requires_oauth_metadata() -> Result<()> {
+    let (headers, mut tokens) = control_input(
+        Cursor::new(b"opaque-access-token\n"),
         /*connect_headers_stdin*/ false,
-        origin,
+        /*require_portals_oauth*/ true,
     )?;
-    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 v1 or v2 tunnel JWT"
+        headers.await?.unwrap_err().to_string(),
+        "Portals requires OAuth CONNECT metadata"
     );
-    assert_eq!(output, b"AUTH_UPDATED\n");
-    assert_eq!(*latest.borrow(), initial);
-    Ok(())
-}
+    assert!(tokens.recv().await.is_none());
 
-#[tokio::test]
-async fn other_bearer_modes_accept_ers_v1_v2_and_unrelated_jwts() -> Result<()> {
-    let unrelated = "eyJ0eXAiOiJKV1QifQ.eyJ2ZXIiOjF9.fixture";
-    let ers_v1 = "eyJ0eXAiOiJjY2EtdHVubmVsK2p3dCJ9.eyJ2ZXIiOjF9.fixture";
-    let ers_v2 = "eyJ0eXAiOiJjY2EtdHVubmVsK2p3dCJ9.eyJ2ZXIiOjJ9.fixture";
-    let ers_v3 = "eyJ0eXAiOiJjY2EtdHVubmVsK2p3dCJ9.eyJ2ZXIiOjN9.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{ers_v2}\n{ers_v3}\n"
-            )),
-            /*connect_headers_stdin*/ true,
-            origin,
-        )?;
-        headers.await??;
-        for token in [unrelated, ers_v1, ers_v2] {
-            let actual = tokens.recv().await.unwrap()?.unwrap();
-            assert_eq!(actual.to_str()?, format!("Bearer {token}"));
-            assert!(actual.is_sensitive());
-        }
-        let error = tokens.recv().await.unwrap().unwrap_err();
-        assert_eq!(
-            error.to_string(),
-            "Portals requires an ERS-issued v1 or v2 tunnel JWT"
-        );
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn portals_jwt_listener_accepts_v1_and_v2_at_startup_and_on_renewal() -> Result<()> {
-    // Inert parser fixtures with a placeholder signature. No signing key or live credential.
-    const V1: &str = "eyJ0eXAiOiJjY2EtdHVubmVsK2p3dCJ9.eyJ2ZXIiOjF9.fixture";
-    const V2: &str = "eyJ0eXAiOiJjY2EtdHVubmVsK2p3dCJ9.eyJ2ZXIiOjJ9.fixture";
-    for (first, renewed) in [(V1, V1), (V1, V2), (V2, V1), (V2, V2)] {
-        let (_, mut tokens) = control_input(
-            Cursor::new(format!("{first}\n{renewed}\n")),
-            /*connect_headers_stdin*/ false,
-            super::ssh::PUBLIC_GATEWAY,
-        )?;
-        let initial = tokens.recv().await.unwrap()?.unwrap();
-        assert_eq!(initial.to_str()?, format!("Bearer {first}"));
-        assert!(initial.is_sensitive());
-        let (updates, latest) = watch::channel(initial);
-        let (ready, readiness) = oneshot::channel();
-        ready.send(()).unwrap();
-        let mut output = Vec::new();
-        update_auth_tokens(&mut tokens, updates, readiness, &mut output).await?;
-        assert_eq!(latest.borrow().to_str()?, format!("Bearer {renewed}"));
-        assert!(latest.borrow().is_sensitive());
-        assert_eq!(output, b"AUTH_UPDATED\n");
-    }
+    let (headers, mut tokens) = control_input(
+        Cursor::new(b"[[\"x-portals-auth-mode\",\"oauth\"]]\nopaque-access-token\n"),
+        /*connect_headers_stdin*/ true,
+        /*require_portals_oauth*/ true,
+    )?;
+    assert_eq!(headers.await??["x-portals-auth-mode"], "oauth");
+    assert_eq!(
+        tokens.recv().await.unwrap()?.unwrap(),
+        "Bearer opaque-access-token"
+    );
     Ok(())
 }
 
@@ -265,7 +168,7 @@ async fn closed_parent_pipe_finishes_even_before_proxy_readiness() -> Result<()>
     let (_, mut tokens) = control_input(
         Cursor::new([]),
         /*connect_headers_stdin*/ false,
-        "https://proxy.example",
+        /*require_portals_oauth*/ false,
     )?;
     let (updates, _) = watch::channel("Bearer initial".parse()?);
     let (_ready, readiness) = oneshot::channel();
@@ -319,7 +222,7 @@ async fn invalid_control_input_stops_credentials_without_exposing_secrets() -> R
         let (metadata, mut tokens) = control_input(
             Cursor::new(input),
             /*connect_headers_stdin*/ true,
-            "https://proxy.example",
+            /*require_portals_oauth*/ false,
         )?;
         let error = metadata.await?.expect_err(case);
         assert!(!format!("{error:#}").contains("private-"), "{case}");
@@ -333,7 +236,7 @@ async fn invalid_control_input_stops_credentials_without_exposing_secrets() -> R
         let (metadata, mut tokens) = control_input(
             Cursor::new(input),
             /*connect_headers_stdin*/ false,
-            "https://proxy.example",
+            /*require_portals_oauth*/ false,
         )?;
         assert!(metadata.await??.is_empty(), "{case}");
         let error = tokens.recv().await.expect(case).expect_err(case);

codex-rs/tcp-tunnel/src/ssh.rs

@@ -19,7 +19,7 @@ use super::ProxyConnection;
 use super::ProxyTarget;
 use super::bridge;
 use super::client_config;
-use super::control::TunnelCredential;
+use super::control::bearer_header;
 
 // 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";
@@ -74,7 +74,7 @@ fn oauth_bearer(token: &str) -> Result<HeaderValue> {
                 .all(|b| b.is_ascii_alphanumeric() || b"-._~+/".contains(&b)),
         "invalid OAuth access token"
     );
-    TunnelCredential::Bearer.header(token)
+    bearer_header(token)
 }
 
 /// Relay one SSH session. Poll the active account's token only after verifying the gateway.