fix(openai): parse Retry-After dates, stop dropping raw frames, restrict base_url

Retry-After now accepts the HTTP-date form in addition to integer seconds.
When a raw SSE passthrough frame fails to construct, fall back to the
serialized form instead of silently dropping the event. Reject upstream
base_url values that are http or resolve to loopback/private ranges, so an
operator-supplied URL cannot point the data plane at an internal address.

Co-authored-by: Claude <noreply@anthropic.com>

backend/crates/providers/openai/src/config.rs

@@ -5,7 +5,7 @@ use std::time::Duration;
 
 use chrono::{DateTime, Utc};
 use serde::Deserialize;
-use url::Url;
+use url::{Host, Url};
 
 use crate::credential::CodexQuotaRefreshPolicy;
 use crate::transport::profile::{CodexWireProfile, CodexWireProfileState};
@@ -132,7 +132,22 @@ impl CodexApiConfig {
     fn validate(&self) -> Result<(), OpenAiConfigError> {
         let url = Url::parse(&self.base_url)
             .map_err(|_| OpenAiConfigError::InvalidField("openai.api.base_url"))?;
-        if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() {
+        // 上游地址只接受 https；明文 http 仅放行本机回环（本地联调），
+        // 避免把上游指向内网明文服务。凭据/查询串会污染端点拼接，一并拒绝。
+        let is_loopback_host = match url.host() {
+            Some(Host::Domain("localhost")) => true,
+            Some(Host::Ipv4(address)) => address.is_loopback(),
+            Some(Host::Ipv6(address)) => address.is_loopback(),
+            Some(Host::Domain(_)) | None => false,
+        };
+        let is_loopback_http = url.scheme() == "http" && is_loopback_host;
+        let is_secure_public = url.scheme() == "https" && url.host_str().is_some();
+        if !(is_loopback_http || is_secure_public)
+            || !url.username().is_empty()
+            || url.password().is_some()
+            || url.query().is_some()
+            || url.fragment().is_some()
+        {
             return Err(OpenAiConfigError::InvalidField("openai.api.base_url"));
         }
         Ok(())

backend/crates/providers/openai/src/transport/canonical.rs

@@ -273,7 +273,7 @@ impl CodexCanonicalDecoder {
         self.merge_timing_signals(signals);
         if matches!(event_type, Some("response.failed" | "error")) {
             let failure = ResponsesSseFailure::from_event(event_type.unwrap_or_default(), &value);
-            if let Ok(wire) = Self::wire_for_event(event, value, raw_sse_frame) {
+            if let Some(wire) = Self::wire_for_event(event, value, raw_sse_frame) {
                 output.push(ProviderEvent::wire(wire));
             }
             return Err(CodexCanonicalError::Upstream(failure));
@@ -282,7 +282,7 @@ impl CodexCanonicalDecoder {
         if let Some(event_type) = event_type {
             let _ = self.decode_event(event_type, &value, &mut canonical);
         }
-        let Ok(wire) = Self::wire_for_event(event, value, raw_sse_frame) else {
+        let Some(wire) = Self::wire_for_event(event, value, raw_sse_frame) else {
             return Ok(());
         };
         output.push(if canonical.is_empty() {
@@ -301,28 +301,47 @@ impl CodexCanonicalDecoder {
         self.timing_signals.text_output |= signals.text_output;
     }
 
+    /// 构造下发的 wire event。
+    ///
+    /// 上游事件名或 SSE id 不满足 wire 安全约束时剥离该元数据继续下发，
+    /// 而不是把整帧静默丢弃；raw 帧存在时原始字节仍原样透传，JSON 正文
+    /// 继续供旁路观测使用。
     fn wire_for_event(
         event: SseEvent,
         value: Value,
         raw_sse_frame: Option<Bytes>,
-    ) -> Result<ProtocolWireEvent, gateway_core::error::IdentifierError> {
+    ) -> Option<ProtocolWireEvent> {
+        let event_type = event.event.filter(|name| {
+            ProtocolWireEvent::json("openai", Some(name.clone()), Value::Null).is_ok()
+        });
+        let sse_id = event.id.filter(|id| {
+            ProtocolWireEvent::json_with_sse_metadata(
+                "openai",
+                None,
+                Value::Null,
+                Some(id.clone()),
+                None,
+            )
+            .is_ok()
+        });
         match raw_sse_frame {
             Some(raw_sse_frame) => ProtocolWireEvent::json_with_raw_sse_metadata(
                 "openai",
-                event.event,
+                event_type,
                 value,
                 raw_sse_frame,
-                event.id,
+                sse_id,
                 event.retry,
             ),
             None => ProtocolWireEvent::json_with_sse_metadata(
                 "openai",
-                event.event,
+                event_type,
                 value,
-                event.id,
+                sse_id,
                 event.retry,
             ),
         }
+        .ok()
     }
 
     fn failure(

backend/crates/providers/openai/src/transport/client.rs

@@ -10,6 +10,7 @@ use std::{
 
 use crate::transport::profile::CodexWireProfileState;
 use bytes::Bytes;
+use chrono::{DateTime, NaiveDateTime, Utc};
 use futures::{Stream, StreamExt};
 use gateway_protocol::openai::{
     WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY, events::retry_after_seconds_from_body,
@@ -491,11 +492,33 @@ pub(super) fn retry_after_seconds(headers: &HeaderMap, body: Option<&str>) -> Op
     headers
         .get(RETRY_AFTER)
         .and_then(|value| value.to_str().ok())
-        .and_then(|value| value.trim().parse::<u64>().ok())
-        .filter(|seconds| *seconds > 0)
+        .and_then(parse_retry_after)
         .or_else(|| body.and_then(retry_after_seconds_from_body))
 }
 
+/// 解析 RFC 7231 `Retry-After` 值：delay-seconds 或 HTTP-date，统一换算为剩余秒数。
+pub(super) fn parse_retry_after(value: &str) -> Option<u64> {
+    let value = value.trim();
+    if let Ok(seconds) = value.parse::<u64>() {
+        return (seconds > 0).then_some(seconds);
+    }
+    let remaining = parse_http_date(value)?.signed_duration_since(Utc::now());
+    u64::try_from(remaining.num_seconds())
+        .ok()
+        .filter(|seconds| *seconds > 0)
+}
+
+/// IMF-fixdate 优先；RFC 7231 要求接收方兼容过时的 RFC 850 与 asctime 形式。
+fn parse_http_date(value: &str) -> Option<DateTime<Utc>> {
+    if let Ok(date) = DateTime::parse_from_rfc2822(value) {
+        return Some(date.with_timezone(&Utc));
+    }
+    ["%A, %d-%b-%y %H:%M:%S GMT", "%a %b %e %H:%M:%S %Y"]
+        .iter()
+        .find_map(|format| NaiveDateTime::parse_from_str(value, format).ok())
+        .map(|date| date.and_utc())
+}
+
 pub(super) fn truncate_for_error(body: &str) -> String {
     body.chars().take(200).collect()
 }

backend/crates/providers/openai/src/transport/websocket/handshake.rs

@@ -18,7 +18,8 @@ use crate::{
         responses::CodexResponsesRequest, websocket::websocket_response_create_payload_text,
     },
     transport::{
-        diagnostics::CodexUpstreamSendPhase, endpoints::CODEX_RESPONSES_PATH, response_meta, tls,
+        client::parse_retry_after, diagnostics::CodexUpstreamSendPhase,
+        endpoints::CODEX_RESPONSES_PATH, response_meta, tls,
     },
 };
 
@@ -185,8 +186,7 @@ fn websocket_opening_error(response: &WsResponse<Option<Vec<u8>>>) -> CodexWebSo
         .headers()
         .get("retry-after")
         .and_then(|value| value.to_str().ok())
-        .and_then(|value| value.parse::<u64>().ok())
-        .filter(|seconds| *seconds > 0)
+        .and_then(parse_retry_after)
         .or_else(|| events::retry_after_seconds_from_body(&body));
     CodexWebSocketExchangeError::upstream(
         status_code,

backend/crates/providers/openai/tests/config.rs

@@ -45,6 +45,43 @@ fn openai_config_rejects_noncanonical_versions_and_empty_fields() {
     );
 }
 
+#[test]
+fn openai_config_restricts_upstream_base_url_to_https_or_loopback_http() {
+    for base_url in [
+        "http://internal.example.com/backend-api",
+        "http://10.0.0.7/backend-api",
+        "https://chatgpt.com/backend-api?debug=1",
+        "https://user:pass@chatgpt.com/backend-api",
+        "https://chatgpt.com/backend-api#fragment",
+        "ftp://chatgpt.com/backend-api",
+    ] {
+        let mut config = valid_config();
+        config.api.base_url = base_url.to_owned();
+        assert!(
+            config
+                .resolve_and_validate(Path::new("/srv/gateway"))
+                .is_err(),
+            "expected {base_url} to be rejected"
+        );
+    }
+
+    for base_url in [
+        "https://chatgpt.com/backend-api",
+        "http://127.0.0.1:8080/backend-api",
+        "http://localhost:8080/backend-api",
+        "http://[::1]:8080/backend-api",
+    ] {
+        let mut config = valid_config();
+        config.api.base_url = base_url.to_owned();
+        assert!(
+            config
+                .resolve_and_validate(Path::new("/srv/gateway"))
+                .is_ok(),
+            "expected {base_url} to be accepted"
+        );
+    }
+}
+
 #[test]
 fn openai_config_keeps_the_quota_exhaustion_scheduling_switch() {
     let mut config = valid_config();

backend/crates/providers/openai/tests/transport/canonical.rs

@@ -128,6 +128,44 @@ fn raw_sse_passthrough_should_forward_unparseable_frames_without_failure() {
     );
 }
 
+#[test]
+fn raw_sse_passthrough_should_forward_frames_with_unsafe_event_metadata() {
+    // 上游事件名超出 wire 安全上限（256 字节）时剥离该元数据继续下发，
+    // 原始帧字节仍原样透传，而不是整帧静默丢弃。
+    let event_name = "x".repeat(300);
+    let frame = format!("event: {event_name}\ndata: {{\"type\":\"noise\",\"marker\":1}}\n\n");
+
+    let events = CodexCanonicalDecoder::new("fallback")
+        .with_raw_sse_passthrough()
+        .push(frame.as_bytes())
+        .expect("unsafe event metadata must not abort transparent transport");
+    let wire = events[0].wire_event().expect("wire event");
+
+    assert_eq!(
+        wire.raw_sse_frame().map(AsRef::as_ref),
+        Some(frame.as_bytes())
+    );
+    assert!(wire.has_json_data());
+    assert_eq!(wire.event_type(), None);
+}
+
+#[test]
+fn decoder_should_strip_unsafe_event_metadata_instead_of_dropping_the_event() {
+    // 无 raw 帧的路径（WebSocket JSON 投影）同样不允许因元数据违规丢事件：
+    // 剥离事件名后 JSON 正文仍完整下发。
+    let event_name = "y".repeat(300);
+    let frame = format!("event: {event_name}\ndata: {{\"type\":\"noise\",\"marker\":1}}\n\n");
+
+    let events = CodexCanonicalDecoder::new("fallback")
+        .push(frame.as_bytes())
+        .expect("unsafe event metadata must not drop the event");
+    let wire = events[0].wire_event().expect("wire event");
+
+    assert_eq!(wire.event_type(), None);
+    assert!(wire.has_json_data());
+    assert_eq!(wire.data()["marker"], serde_json::json!(1));
+}
+
 #[test]
 fn decoder_should_emit_calculated_cost_for_complete_known_model_usage() {
     let body = concat!(

backend/crates/providers/openai/tests/transport/usage.rs

@@ -220,6 +220,79 @@ async fn content_length_over_limit_error_body_should_keep_only_safe_metadata() {
     assert_oversized_error(error, StatusCode::TOO_MANY_REQUESTS, Some(7));
 }
 
+#[tokio::test]
+async fn retry_after_http_date_should_be_converted_to_remaining_seconds() {
+    let server = MockServer::start().await;
+    let retry_at = (Utc::now() + chrono::TimeDelta::seconds(90))
+        .format("%a, %d %b %Y %H:%M:%S GMT")
+        .to_string();
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(
+            ResponseTemplate::new(429)
+                .insert_header("retry-after", retry_at.as_str())
+                .set_body_raw(
+                    r#"{"error":{"message":"rate limited"}}"#,
+                    "application/json",
+                ),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+
+    let error = client(&server.uri())
+        .fetch_usage(context())
+        .await
+        .expect_err("http-date rate limit");
+
+    let CodexClientError::Upstream {
+        status,
+        retry_after_seconds,
+        ..
+    } = error
+    else {
+        panic!("expected an upstream rate-limit error");
+    };
+    assert_eq!(status, StatusCode::TOO_MANY_REQUESTS);
+    let seconds = retry_after_seconds.expect("http-date retry-after should yield seconds");
+    assert!(
+        (1..=90).contains(&seconds),
+        "expected remaining seconds within 90s, got {seconds}"
+    );
+}
+
+#[tokio::test]
+async fn retry_after_http_date_in_the_past_should_be_ignored() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(
+            ResponseTemplate::new(429)
+                .insert_header("retry-after", "Mon, 01 Jan 2001 00:00:00 GMT")
+                .set_body_raw(
+                    r#"{"error":{"message":"rate limited"}}"#,
+                    "application/json",
+                ),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+
+    let error = client(&server.uri())
+        .fetch_usage(context())
+        .await
+        .expect_err("http-date rate limit");
+
+    let CodexClientError::Upstream {
+        retry_after_seconds,
+        ..
+    } = error
+    else {
+        panic!("expected an upstream rate-limit error");
+    };
+    assert_eq!(retry_after_seconds, None);
+}
+
 #[tokio::test]
 async fn chunked_body_over_limit_should_be_rejected_without_content_length() {
     let (base_url, server) = spawn_chunked_server(usage_body(MAX_CODEX_USAGE_BODY_BYTES + 1));

backend/crates/providers/openai/tests/transport/websocket.rs

@@ -423,6 +423,51 @@ async fn websocket_execute_response_create_request_should_preserve_opening_error
     assert!(body.contains("rate limited"));
 }
 
+#[tokio::test]
+async fn websocket_opening_error_should_convert_http_date_retry_after_to_seconds() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let retry_at = (chrono::Utc::now() + chrono::TimeDelta::seconds(120))
+        .format("%a, %d %b %Y %H:%M:%S GMT")
+        .to_string();
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        let _request = read_http_request(&mut stream).await;
+        let body = r#"{"error":{"message":"rate limited"}}"#;
+        stream
+            .write_all(
+                format!(
+                    "HTTP/1.1 429 Too Many Requests\r\nretry-after: {retry_at}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
+                    body.len()
+                )
+                .as_bytes(),
+            )
+            .await
+            .unwrap();
+    });
+    let prepared = prepared_websocket_request(&format!("http://{addr}"));
+
+    let error = execute_response_create_request(&prepared)
+        .await
+        .expect_err("failed opening should surface upstream status");
+    server.await.unwrap();
+
+    let CodexClientError::Upstream {
+        status,
+        retry_after_seconds,
+        ..
+    } = error
+    else {
+        panic!("expected upstream opening error");
+    };
+    assert_eq!(status, reqwest::StatusCode::TOO_MANY_REQUESTS);
+    let seconds = retry_after_seconds.expect("http-date retry-after should yield seconds");
+    assert!(
+        (1..=120).contains(&seconds),
+        "expected remaining seconds within 120s, got {seconds}"
+    );
+}
+
 #[tokio::test]
 async fn websocket_execute_response_create_request_should_reject_binary_event() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();