diff --git a/backend/crates/providers/openai/src/config.rs b/backend/crates/providers/openai/src/config.rs index 2b95cafda..e2732d06d 100644 --- a/backend/crates/providers/openai/src/config.rs +++ b/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(()) diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index c54bf8793..2d847c58c 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/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, - ) -> Result { + ) -> Option { + 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( diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index 89df8b527..e8d7214c3 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/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::().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 { + let value = value.trim(); + if let Ok(seconds) = value.parse::() { + 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> { + 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() } diff --git a/backend/crates/providers/openai/src/transport/websocket/handshake.rs b/backend/crates/providers/openai/src/transport/websocket/handshake.rs index e5661dd2c..b8b97c1e7 100644 --- a/backend/crates/providers/openai/src/transport/websocket/handshake.rs +++ b/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>>) -> CodexWebSo .headers() .get("retry-after") .and_then(|value| value.to_str().ok()) - .and_then(|value| value.parse::().ok()) - .filter(|seconds| *seconds > 0) + .and_then(parse_retry_after) .or_else(|| events::retry_after_seconds_from_body(&body)); CodexWebSocketExchangeError::upstream( status_code, diff --git a/backend/crates/providers/openai/tests/config.rs b/backend/crates/providers/openai/tests/config.rs index 6488b96a7..7f712d721 100644 --- a/backend/crates/providers/openai/tests/config.rs +++ b/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(); diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index 6dcdc991a..b507ce49a 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/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!( diff --git a/backend/crates/providers/openai/tests/transport/usage.rs b/backend/crates/providers/openai/tests/transport/usage.rs index 95a89abb8..c2c4e40c9 100644 --- a/backend/crates/providers/openai/tests/transport/usage.rs +++ b/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)); diff --git a/backend/crates/providers/openai/tests/transport/websocket.rs b/backend/crates/providers/openai/tests/transport/websocket.rs index fa2734fb5..8bae133cd 100644 --- a/backend/crates/providers/openai/tests/transport/websocket.rs +++ b/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();