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 <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-07 22:45:42 +08:00
co-authored by Codex
parent fb15f3e36c
commit 6c583e341a
3 changed files with 195 additions and 37 deletions
@@ -176,6 +176,54 @@ struct PreparedGenerateRequest {
cyber_policy_session_key: Option<ProviderSessionAffinityKey>,
}
impl PreparedGenerateRequest {
fn checked_native_continuation_scope(
&self,
context: &AttemptContext,
) -> Result<PreviousResponseScope, ProviderError> {
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 => {
@@ -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());
+3 -1
View File
@@ -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