From 6c583e341ae5d21cb85107a67bb9fc2f9879b5cd Mon Sep 17 00:00:00 2001 From: Zyy Date: Wed, 7 Oct 2026 22:45:42 +0800 Subject: [PATCH] fix(openai): reject invalid continuation replay before account selection Check the recorded native continuation scope before selecting an account or committing a session binding. Reuse the same scope validation after middleware while preserving native connection checks and migration policy. Cover ReplayOwner and ReplayAny with persisted provider metadata or a connection-local pin, with and without an existing session binding. Rejected replays must not acquire leases, write bindings or aliases, send upstream requests, or redirect subsequent child requests. Refs #364 Co-authored-by: Codex --- .../providers/openai/src/provider/mod.rs | 88 ++++++----- .../provider/contract/session_binding.rs | 140 ++++++++++++++++++ docs/architecture.md | 4 +- 3 files changed, 195 insertions(+), 37 deletions(-) diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 7121a80ad..197d4eee2 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -176,6 +176,54 @@ struct PreparedGenerateRequest { cyber_policy_session_key: Option, } +impl PreparedGenerateRequest { + fn checked_native_continuation_scope( + &self, + context: &AttemptContext, + ) -> Result { + let pin = context.continuation().and_then(ContinuationBinding::pinned); + let scope = match self + .previous_session + .as_ref() + .map(|state| state.continuation_scope) + { + Some(OpenAiContinuationScope::Persisted) => PreviousResponseScope::Persisted, + Some(OpenAiContinuationScope::ConnectionLocal) => { + PreviousResponseScope::ConnectionLocal + } + Some(OpenAiContinuationScope::ReplayRequired) => PreviousResponseScope::ExternalUnknown, + None => match pin.map(|pin| pin.scope()) { + Some(NativeContinuationScope::Persisted) => PreviousResponseScope::Persisted, + Some(NativeContinuationScope::ConnectionLocal) => { + PreviousResponseScope::ConnectionLocal + } + None => PreviousResponseScope::ExternalUnknown, + }, + }; + // 已知不能脱离原连接的重放必须在选号前拒绝，避免失败请求先改写共享绑定 + // Native 仍交给连接池核对原连接；此处不推测连接是否存活或上游状态是否可用 + if pin.is_some() + && matches!( + context.continuation_attempt(), + ContinuationAttempt::ReplayOwner | ContinuationAttempt::ReplayAny + ) + && scope == PreviousResponseScope::ConnectionLocal + { + tracing::warn!( + request_id = context.request_id().as_str(), + attempt_index = context.attempt_index().get(), + continuation_scope = "connection_local", + continuation_attempt = context.continuation_attempt().as_str(), + continuation_recovery_disposition = "client_replay_required", + continuation_recovery_action = "stop_proxy_recovery", + "OpenAI connection-local continuation replay was rejected before send" + ); + return Err(continuation_replay_required_error("scope_unavailable")); + } + Ok(scope) + } +} + struct SelectedGenerate { lease: CodexCredentialLease, account_selection_wait_ms: u64, @@ -578,6 +626,9 @@ impl Provider for CodexProvider { let preselection = upstream .filter(|_| adapter.is_none()) .map(|upstream| self.prepare_generate_request(generate, upstream, &context)); + if let Some(prepared) = preselection.as_ref() { + prepared.checked_native_continuation_scope(&context)?; + } let (selection_session_affinity, selection_cyber_policy_key, requires_websocket) = preselection.map_or((None, None, false), |prepared| { let requires_websocket = @@ -715,6 +766,7 @@ impl CodexProvider { &middleware_headers, ); let processed = self.prepare_generate_request(&generate, upstream, &context); + let native_scope = processed.checked_native_continuation_scope(&context)?; let mut upstream_request = processed.upstream; let session_transport_key = derive_codex_transport_key(&upstream_request, context.client_api_key_ref()); @@ -788,42 +840,6 @@ impl CodexProvider { { match continuation { ContinuationBinding::Pinned(continuation) => { - let native_scope = match previous_session - .as_ref() - .map(|state| state.continuation_scope) - { - Some(OpenAiContinuationScope::Persisted) => { - PreviousResponseScope::Persisted - } - Some(OpenAiContinuationScope::ConnectionLocal) => { - PreviousResponseScope::ConnectionLocal - } - Some(OpenAiContinuationScope::ReplayRequired) => { - PreviousResponseScope::ExternalUnknown - } - None => match continuation.scope() { - NativeContinuationScope::Persisted => PreviousResponseScope::Persisted, - NativeContinuationScope::ConnectionLocal => { - PreviousResponseScope::ConnectionLocal - } - }, - }; - if matches!( - context.continuation_attempt(), - ContinuationAttempt::ReplayOwner | ContinuationAttempt::ReplayAny - ) && native_scope == PreviousResponseScope::ConnectionLocal - { - tracing::warn!( - request_id = context.request_id().as_str(), - attempt_index = context.attempt_index().get(), - continuation_scope = "connection_local", - continuation_attempt = context.continuation_attempt().as_str(), - continuation_recovery_disposition = "client_replay_required", - continuation_recovery_action = "stop_proxy_recovery", - "OpenAI connection-local continuation replay was rejected before send" - ); - return Err(continuation_replay_required_error("scope_unavailable")); - } let previous_response_scope = match context.continuation_attempt() { ContinuationAttempt::Native => native_scope, ContinuationAttempt::ReplayOwner | ContinuationAttempt::ReplayAny => { diff --git a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs index b3bc1a642..d53eeeca0 100644 --- a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs +++ b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs @@ -783,6 +783,146 @@ async fn child_binding_only_constrains_builtin_scheduling_and_preserves_plugin_c } } +#[tokio::test] +async fn connection_local_replay_does_not_claim_or_migrate_a_session() { + use gateway_core::engine::continuation::NativeContinuationScope; + + for existing_binding in [true, false] { + for scope_from_state in [false, true] { + for recovery in [ + ContinuationAttempt::ReplayOwner, + ContinuationAttempt::ReplayAny, + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let leases = Arc::new(TestLeaseCoordinator::default()); + let server = MockServer::start().await; + let provider = provider_with_affinity_and_base_url_and_leases( + &store, + affinity.clone(), + server.uri(), + leases.clone(), + ); + if existing_binding { + drop( + provider + .clone() + .execute( + planned_request("openai", turn_request("root", "seed-turn")), + context("req_seed_owner", CancellationToken::new()), + ) + .await + .unwrap(), + ); + } + create_account(&store, "acct_subagent_b").await; + let renewals = affinity.renewal_ttls(); + let aliases = affinity.alias_ttls(); + leases.requests.lock().unwrap().clear(); + + let provider_kind = ProviderKind::new("openai").unwrap(); + let account = ProviderAccountId::new("acct_subagent_a").unwrap(); + let client = ClientApiKeyId::new("key_openai_contract").unwrap(); + let owner = ProviderAccountStateOwner::new(provider_kind.clone(), account.clone()); + let pin = NativeContinuationPin::new( + PreviousResponseId::new("client-previous"), + PreviousResponseId::new("upstream-previous"), + client.clone(), + provider_kind, + account.clone(), + ) + .with_scope(if scope_from_state { + NativeContinuationScope::Persisted + } else { + NativeContinuationScope::ConnectionLocal + }); + // 覆盖 Coordinator 排除旧 owner 后改走 ReplayAny 的入口 + let excluded = if recovery == ContinuationAttempt::ReplayAny { + BTreeSet::from([account]) + } else { + BTreeSet::new() + }; + let attempt = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_invalid_replay").unwrap(), + client, + ), + NonZeroU32::new(2).unwrap(), + SystemTime::now() + Duration::from_secs(5), + account_policy(), + AccountAttemptContext::new(excluded, None, Some(owner)) + .with_account_scope(contract_account_scope()), + Some(ContinuationBinding::Pinned(pin)), + CancellationToken::new(), + ) + .with_continuation_attempt(recovery); + let mut generate = generate_with_session_context("root", Some("root"), None); + if scope_from_state { + // Provider 保存的状态比通用 pin 更精确，不能只检查 pin 的作用域 + generate = generate.with_provider_session_state(ProviderSessionState::new( + "openai", json!({ + "account_id":"acct_subagent_a", "conversation_id":"conversation-root", + "continuation_scope":"connection_local" + }).as_object().unwrap().clone(), + ).unwrap()); + } + let error = provider + .clone() + .execute( + planned_request("openai", Operation::Generate(generate)), + attempt, + ) + .await + .err() + .expect("connection-local replay requires client recovery"); + assert_eq!( + error.kind(), + ProviderErrorKind::ContinuationRecoveryRequired + ); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert_eq!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + assert_eq!( + affinity.renewal_ttls(), + renewals, + "rejected replay must not write a binding" + ); + assert_eq!(affinity.alias_ttls(), aliases); + assert_eq!(affinity.binding_count(), usize::from(existing_binding)); + assert!( + leases.requests.lock().unwrap().is_empty(), + "rejected replay must not acquire a lease" + ); + if existing_binding { + let child = provider + .clone() + .execute( + planned_request( + "openai", + Operation::Generate(generate_with_session_context( + "root", + Some("child"), + None, + )), + ), + context("req_child_after_rejected_replay", CancellationToken::new()), + ) + .await + .unwrap(); + assert_eq!( + child.metadata().provider_account_id().as_str(), + "acct_subagent_a" + ); + } + assert!(server.received_requests().await.unwrap().is_empty()); + } + } + } +} + #[tokio::test] async fn old_native_continuation_cannot_restore_the_pre_migration_account() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/docs/architecture.md b/docs/architecture.md index caf23f38f..895edf0c7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -614,7 +614,9 @@ Continuation 仍受原请求的 Client Key、账号范围、Provider 和发送/ - OpenAI 在交付前收到可安全重放的明确额度拒绝时，先隔离账号，再投影 `ClientReplayRequired`； 丢弃未交付的原错误帧，由客户端提交完整历史开启新链，不把原增量输入交给其他账号。 真实错误分类、状态码、发送状态和上游诊断保持不变，客户端合同见 [Responses API](api.md#3-openai-数据面与模型目录) -- OpenAI 按 native → replay owner → replay any 推进，并保留官方 `previous_response_id` 语义 +- OpenAI 按 native → replay owner → replay any 推进，并保留官方 `previous_response_id` 语义； + 已知为 connection-local 的原生续接进入重放阶段时，在选号前返回 `ClientReplayRequired`，不认领或迁移会话绑定。 + 正常 native 续接仍由连接池核对原连接及 response ID 是否可用 - xAI 使用客户端提交的完整历史作为重放输入 - scope 外账号、跨 Key 复用或不明确发送结果均 fail closed