diff --git a/backend/apps/gateway/src/bootstrap/command.rs b/backend/apps/gateway/src/bootstrap/command.rs index 27a3807df..f2ddb76ef 100644 --- a/backend/apps/gateway/src/bootstrap/command.rs +++ b/backend/apps/gateway/src/bootstrap/command.rs @@ -103,6 +103,10 @@ async fn execute( let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control()); or_shutdown!(plugin_runtime, management.bind(&plugin_runtime)); let core = or_shutdown!(plugin_runtime, core.activate().await); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_client_identity_port(&core.client_identity_port()) + ); let nested_models = core.nested_model_execution_port(); let affinity_lookup = core.affinity_lookup_port(); or_shutdown!( diff --git a/backend/apps/gateway/src/bootstrap/server.rs b/backend/apps/gateway/src/bootstrap/server.rs index 97d353075..2e52d185a 100644 --- a/backend/apps/gateway/src/bootstrap/server.rs +++ b/backend/apps/gateway/src/bootstrap/server.rs @@ -38,6 +38,10 @@ pub async fn run() -> Result<(), BootstrapError> { let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control()); or_shutdown!(plugin_runtime, management.bind(&plugin_runtime)); let mut core = or_shutdown!(plugin_runtime, core.activate().await); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_client_identity_port(&core.client_identity_port()) + ); let nested_models = core.nested_model_execution_port(); let affinity_lookup = core.affinity_lookup_port(); or_shutdown!( diff --git a/backend/crates/gateway-core/src/engine/authentication.rs b/backend/crates/gateway-core/src/engine/authentication.rs index ec2024d33..2589f071d 100644 --- a/backend/crates/gateway-core/src/engine/authentication.rs +++ b/backend/crates/gateway-core/src/engine/authentication.rs @@ -9,10 +9,32 @@ use std::{ use futures::future::BoxFuture; use crate::{ + error::GatewayError, policy::{ClientApiKeyId, PlaintextClientApiKey}, - routing::extensions::{ExtensionSetId, ExtensionSetReference}, + routing::{ + extensions::{ExtensionSetId, ExtensionSetReference}, + request_settings::RequestSettings, + }, }; +use super::execution::AuthenticatedClient; + +/// 身份解析与认证只返回冻结权限，不提供模型执行入口 +pub trait ClientIdentityPort: Send + Sync { + /// 只解析身份与冻结权限，不建立模型执行图或记录使用事实 + fn resolve_client( + &self, + client_key_id: ClientApiKeyId, + settings: Option, + ) -> BoxFuture<'_, Result>; + + /// 复用入口认证计划及 principal 映射，不进行模型准入 + fn verify_client( + &self, + request: ClientAuthenticationRequest, + ) -> BoxFuture<'_, Result>; +} + pub const MAXIMUM_AUTHORIZATION_BYTES: usize = 8 * 1024; /// API 从数据面 Authorization 头构造的有界认证信封 diff --git a/backend/crates/gateway-core/src/engine/execution/service.rs b/backend/crates/gateway-core/src/engine/execution/service.rs index 8148aeb43..1e85a906f 100644 --- a/backend/crates/gateway-core/src/engine/execution/service.rs +++ b/backend/crates/gateway-core/src/engine/execution/service.rs @@ -27,7 +27,7 @@ use crate::{ ClientAdmissionRequest, }, authentication::{ - ClientAuthenticationRequest, FrontendAuthenticationDecision, + ClientAuthenticationRequest, ClientIdentityPort, FrontendAuthenticationDecision, FrontendAuthenticationExtensionIndex, }, budget::{ClientBudgetCharge, ClientBudgetPort}, @@ -1705,7 +1705,7 @@ impl ExecutionService for DefaultExecutionService { } } -impl NestedModelExecutionPort for DefaultExecutionService { +impl ClientIdentityPort for DefaultExecutionService { fn resolve_client( &self, client_key_id: ClientApiKeyId, @@ -1737,7 +1737,9 @@ impl NestedModelExecutionPort for DefaultExecutionService { } }) } +} +impl NestedModelExecutionPort for DefaultExecutionService { fn models( &self, client: AuthenticatedClient, diff --git a/backend/crates/gateway-core/src/engine/nested.rs b/backend/crates/gateway-core/src/engine/nested.rs index fb94cd4c7..4519b0641 100644 --- a/backend/crates/gateway-core/src/engine/nested.rs +++ b/backend/crates/gateway-core/src/engine/nested.rs @@ -13,8 +13,8 @@ use futures::future::BoxFuture; use crate::{ account::ProviderAccountId, error::GatewayError, identity::ProviderKind, - lifecycle::CancellationToken, operation::Operation, policy::ClientApiKeyId, - provider_ports::ProviderSessionAffinityKey, routing::PublicModelId, + lifecycle::CancellationToken, operation::Operation, provider_ports::ProviderSessionAffinityKey, + routing::PublicModelId, }; use super::{ @@ -111,19 +111,6 @@ impl fmt::Debug for NestedModelExecutionRequest { /// 嵌套模型执行只进入 Core 的既有执行会话；Runtime 不实现重试、commit 或计费 pub trait NestedModelExecutionPort: Send + Sync { - /// 只解析身份与冻结权限，不建立模型执行图或记录使用事实 - fn resolve_client( - &self, - client_key_id: ClientApiKeyId, - settings: Option, - ) -> BoxFuture<'_, Result>; - - /// 复用入口认证计划及 principal 映射，不进行模型准入 - fn verify_client( - &self, - request: super::authentication::ClientAuthenticationRequest, - ) -> BoxFuture<'_, Result>; - fn models( &self, client: AuthenticatedClient, diff --git a/backend/crates/gateway-core/src/lib.rs b/backend/crates/gateway-core/src/lib.rs index 75607fc63..04e17b73b 100644 --- a/backend/crates/gateway-core/src/lib.rs +++ b/backend/crates/gateway-core/src/lib.rs @@ -34,6 +34,7 @@ use engine::ExecutionStore; use engine::admission::{ ClientAdmissionPort, ClientAdmissionRecoveryPort, restore_client_admission_startup, }; +use engine::authentication::ClientIdentityPort; use engine::continuation::NativeContinuationPort; use engine::execution::{ ClientApiKeyUsageSink, ClientKeyVerifier, DefaultExecutionService, ExecutionService, @@ -127,6 +128,7 @@ pub struct CoreBundle { client_key_verifier: Arc, snapshot_control: Arc, account_probe: Arc, + client_identity: Arc, nested_model_execution: Arc, affinity_lookup: Arc, health_probes: Vec>, @@ -181,6 +183,7 @@ impl CoreStartup { let execution: Arc = service.clone(); let client_key_verifier: Arc = service.clone(); let account_probe: Arc = service.clone(); + let client_identity: Arc = service.clone(); let nested_model_execution: Arc = service.clone(); let affinity_lookup: Arc = service; let snapshot_control: Arc = publisher; @@ -191,6 +194,7 @@ impl CoreStartup { client_key_verifier, snapshot_control, account_probe, + client_identity, nested_model_execution, affinity_lookup, health_probes, @@ -266,6 +270,11 @@ impl CoreBundle { Arc::clone(&self.account_probe) } + #[must_use] + pub fn client_identity_port(&self) -> Arc { + Arc::clone(&self.client_identity) + } + #[must_use] pub fn nested_model_execution_port(&self) -> Arc { Arc::clone(&self.nested_model_execution) @@ -395,6 +404,7 @@ pub struct CoreCommandPlaneStartup { pub struct CoreCommandPlaneBundle { control: CoreControlPlaneBundle, account_probe: Arc, + client_identity: Arc, nested_model_execution: Arc, affinity_lookup: Arc, } @@ -425,11 +435,13 @@ impl CoreCommandPlaneStartup { .with_snapshot_refresh(Arc::clone(&control.publisher)), ); let account_probe: Arc = service.clone(); + let client_identity: Arc = service.clone(); let nested_model_execution: Arc = service.clone(); let affinity_lookup: Arc = service; Ok(CoreCommandPlaneBundle { control, account_probe, + client_identity, nested_model_execution, affinity_lookup, }) @@ -452,6 +464,11 @@ impl CoreCommandPlaneBundle { self.control.snapshot_control() } + #[must_use] + pub fn client_identity_port(&self) -> Arc { + Arc::clone(&self.client_identity) + } + #[must_use] pub fn nested_model_execution_port(&self) -> Arc { Arc::clone(&self.nested_model_execution) diff --git a/backend/crates/gateway-core/tests/engine/execution/identity.rs b/backend/crates/gateway-core/tests/engine/execution/identity.rs index 557630fe2..b157c2600 100644 --- a/backend/crates/gateway-core/tests/engine/execution/identity.rs +++ b/backend/crates/gateway-core/tests/engine/execution/identity.rs @@ -28,7 +28,7 @@ fn identity_queries_keep_frozen_permissions_without_using_execution_resources() let settings = service.request_settings().unwrap(); let key = ClientApiKeyId::new("key_start_test").unwrap(); let client = - NestedModelExecutionPort::resolve_client(&service, key.clone(), Some(settings.clone())) + ClientIdentityPort::resolve_client(&service, key.clone(), Some(settings.clone())) .await .unwrap(); snapshots.publish(start_snapshot_with_policy( @@ -59,12 +59,11 @@ fn identity_queries_keep_frozen_permissions_without_using_execution_resources() .await .unwrap(); assert_eq!(models, vec![PublicModelId::new("gpt-start").unwrap()]); - let frozen = - NestedModelExecutionPort::resolve_client(&service, key.clone(), Some(settings)) - .await - .unwrap(); + let frozen = ClientIdentityPort::resolve_client(&service, key.clone(), Some(settings)) + .await + .unwrap(); assert_eq!(frozen.snapshot().revision().get(), 1); - let current = NestedModelExecutionPort::resolve_client(&service, key, None).await; + let current = ClientIdentityPort::resolve_client(&service, key, None).await; assert!(matches!(current, Err(error) if error.kind() == GatewayErrorKind::Unauthorized)); let cancellation = CancellationToken::default(); @@ -108,7 +107,7 @@ fn execution_scope_limits_apply_only_when_starting_a_model() { scope.extending(format!("ancestor-{index}")).unwrap() }); for extension_scope in [repeated, exhausted] { - let client = NestedModelExecutionPort::resolve_client( + let client = ClientIdentityPort::resolve_client( &service, ClientApiKeyId::new("key_start_test").unwrap(), None, @@ -234,7 +233,7 @@ fn verification_keeps_parent_origin_and_requires_the_frontend_principal_mapping( let mut cancelled_origin = origin.clone(); cancelled_origin.cancellation = CancellationToken::default(); cancelled_origin.cancellation.cancel(); - let cancelled = NestedModelExecutionPort::verify_client( + let cancelled = ClientIdentityPort::verify_client( &service, ClientAuthenticationRequest::new("External fixture") .unwrap() @@ -244,7 +243,7 @@ fn verification_keeps_parent_origin_and_requires_the_frontend_principal_mapping( .await; assert!(matches!(cancelled, Err(error) if error.kind() == GatewayErrorKind::Cancelled)); assert_eq!(plan.calls.load(Ordering::SeqCst), 0); - let result = NestedModelExecutionPort::verify_client( + let result = ClientIdentityPort::verify_client( &service, ClientAuthenticationRequest::new("External fixture") .unwrap() diff --git a/backend/crates/gateway-core/tests/engine/execution/mod.rs b/backend/crates/gateway-core/tests/engine/execution/mod.rs index 55f6f353c..7c9dbbc94 100644 --- a/backend/crates/gateway-core/tests/engine/execution/mod.rs +++ b/backend/crates/gateway-core/tests/engine/execution/mod.rs @@ -13,6 +13,7 @@ use std::{ time::{Duration, Instant, SystemTime}, }; +use gateway_core::engine::authentication::ClientIdentityPort; use gateway_core::engine::budget::{ClientBudgetCharge, ClientBudgetError, ClientBudgetPort}; use gateway_core::error::GatewayError; @@ -1233,7 +1234,7 @@ fn explicit_key_model_execution_preserves_http_origin_and_leaves_background_orig Arc::new(crate::RecordingDiagnostics::default()), ); let input = request(&service, ClientTransport::InternalPlugin); - let client = NestedModelExecutionPort::resolve_client( + let client = ClientIdentityPort::resolve_client( &service, ClientApiKeyId::new("key_start_test").unwrap(), None, diff --git a/backend/crates/gateway-plugin/runtime/src/callback/client.rs b/backend/crates/gateway-plugin/runtime/src/callback/client.rs new file mode 100644 index 000000000..9e4958d55 --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/src/callback/client.rs @@ -0,0 +1,154 @@ +//! 客户端认证、冻结权限查询与调用内身份复用 + +use std::sync::{Arc, OnceLock, Weak}; + +use gateway_admin::model::AdminError; +use gateway_core::{ + account::ProviderAccountId, + engine::{ + authentication::{ClientAuthenticationRequest, ClientIdentityPort}, + execution::AuthenticatedClient, + }, + policy::ClientApiKeyId, + routing::UpstreamModelId, +}; +use gateway_plugin_sdk::{CallContext, ErrorCode, PluginFault, call::data}; + +use super::{CallResources, denied, invalid}; +use crate::RpcReply; + +const MAX_CLIENT_IDENTITIES_PER_CALL: usize = 16; + +pub(crate) struct PluginClientIdentityPortSlot { + port: OnceLock>, +} + +impl PluginClientIdentityPortSlot { + pub(crate) const fn new() -> Self { + Self { + port: OnceLock::new(), + } + } + + pub(crate) fn bind(&self, port: &Arc) -> Result<(), AdminError> { + self.port + .set(Arc::downgrade(port)) + .map_err(|_| AdminError::conflict("插件身份端口已经绑定")) + } + + fn upgrade(&self) -> Result, PluginFault> { + self.port.get().and_then(Weak::upgrade).ok_or_else(denied) + } +} + +pub(super) struct PluginClients { + slot: Arc, +} + +impl PluginClients { + pub(super) fn new(slot: Arc) -> Self { + Self { slot } + } + + pub(super) async fn call( + &self, + context: &CallContext, + call: &Arc, + method: &str, + params: serde_json::Value, + payload: Vec, + ) -> Result { + match method { + data::CLIENT_VERIFY => { + if params != serde_json::json!({}) { + return Err(invalid()); + } + let request: data::VerifyClientRequest = + serde_json::from_slice(&payload).map_err(|_| invalid())?; + let mut authentication = ClientAuthenticationRequest::new(request.authorization) + .map_err(|_| invalid())? + .with_invocation(call.child_origin(context)?); + if let Some(settings) = call.request_settings() { + authentication = authentication.with_settings(settings); + } + let port = self.slot.upgrade()?; + call.observe_dispatch()?; + let client = port + .verify_client(authentication) + .await + .map_err(super::error::gateway)?; + super::admin::encode(&data::VerifiedClient { + client_key_id: client.policy().key_id().as_str().to_owned(), + }) + } + data::ACCESS_CHECK => { + if params != serde_json::json!({}) { + return Err(invalid()); + } + let request: data::AccessQuery = + serde_json::from_slice(&payload).map_err(|_| invalid())?; + let client = self.identity(call, request.client_key_id).await?; + let scope = client.policy().account_scope(); + let model = request + .upstream_model + .map(UpstreamModelId::new) + .transpose() + .map_err(|_| invalid())?; + let accounts = request + .account_ids + .into_iter() + .map(|account_id| { + let account = + ProviderAccountId::new(account_id.clone()).map_err(|_| invalid())?; + let allowed = model.as_ref().map_or_else( + || scope.allows(&account), + |model| scope.allows_model(&account, model.as_str()), + ); + Ok(data::AccountAccessFacts { + account_id, + allowed, + }) + }) + .collect::, PluginFault>>()?; + super::admin::encode(&data::AccessFacts { + schema_version: 1, + client_key_id: client.policy().key_id().as_str().to_owned(), + accounts, + }) + } + _ => Err(denied()), + } + } + + pub(super) async fn identity( + &self, + call: &CallResources, + client_key_id: String, + ) -> Result { + let port = self.slot.upgrade()?; + let key = ClientApiKeyId::new(client_key_id.clone()).map_err(|_| invalid())?; + let settings = call.request_settings(); + let mut identities = call.client_identities.lock().await; + if let Some(client) = identities + .get(&client_key_id) + .filter(|client| client.request_settings() == settings.as_ref()) + { + return Ok(client.clone()); + } + if !identities.contains_key(&client_key_id) + && identities.len() >= MAX_CLIENT_IDENTITIES_PER_CALL + { + return Err(PluginFault::new( + ErrorCode::Capacity, + "client identity capacity is exhausted", + )); + } + let client = port + .resolve_client(key, settings) + .await + .map_err(super::error::gateway)?; + // 查询与执行共用调用内快照，执行链和子执行取消只在真正启动时建立 + identities.insert(client_key_id, client.clone()); + Ok(client) + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs index 2e69f4305..b8feed5d8 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs @@ -4,6 +4,7 @@ mod accounts; mod admin; mod affinity; mod call_resources; +mod client; mod data; pub(crate) mod error; mod facts; @@ -38,6 +39,7 @@ use crate::{CallbackHandler, RpcReply}; pub(crate) use accounts::PluginAccountPortSlot; pub(crate) use affinity::PluginAffinityPortSlot; use call_resources::{CallResources, HttpStream}; +pub(crate) use client::PluginClientIdentityPortSlot; pub(crate) use keys::PluginClientKeyPortSlot; pub(crate) use middleware::{ MiddlewareBinding, MiddlewareBodyAuthority, MiddlewareCallback, MiddlewareCompletionBody, @@ -48,37 +50,14 @@ pub(crate) use resources::PluginResourcePorts; pub(crate) use scope::{CallbackScope, InvocationContext}; pub(crate) struct PluginCallbackPorts { - services: Arc, - http: Arc, - network: NetworkPolicy, - accounts: Arc, - keys: Arc, - models: Arc, - affinity: Arc, - resources: Arc, -} - -impl PluginCallbackPorts { - pub(crate) fn new( - services: Arc, - http: Arc, - accounts: Arc, - keys: Arc, - models: Arc, - affinity: Arc, - resources: Arc, - ) -> Self { - Self { - services, - http, - network: NetworkPolicy::unrestricted(), - accounts, - keys, - models, - affinity, - resources, - } - } + pub(crate) services: Arc, + pub(crate) http: Arc, + pub(crate) accounts: Arc, + pub(crate) keys: Arc, + pub(crate) clients: Arc, + pub(crate) models: Arc, + pub(crate) affinity: Arc, + pub(crate) resources: Arc, } pub(crate) struct PluginCallbacks { @@ -91,6 +70,7 @@ pub(crate) struct PluginCallbacks { keys: Arc, affinity: Arc, log: Arc, + clients: Arc, models: Arc, private_state: Arc, http: Arc, @@ -112,6 +92,7 @@ impl PluginCallbacks { ports: PluginCallbackPorts, request_limits: Arc, ) -> Result { + let clients = Arc::new(client::PluginClients::new(ports.clients)); Ok(Self { registration_body_limit: None, services: ports.services, @@ -132,14 +113,16 @@ impl PluginCallbacks { log::PluginLog::new(manifest, log_slots) .map_err(|_| AdminError::invalid("插件清单身份不合法"))?, ), + clients: clients.clone(), models: Arc::new(model::PluginModels::new( ports.models, + clients, request_limits.clone(), maximum_payload, )), private_state, http: ports.http, - network: Arc::new(ports.network), + network: Arc::new(NetworkPolicy::unrestricted()), scopes: Mutex::new(BTreeMap::new()), pending_middleware: Mutex::new(BTreeMap::new()), calls: Mutex::new(BTreeMap::new()), @@ -437,6 +420,7 @@ impl CallbackHandler for PluginCallbacks { let resources = self.resources.clone(); let data = self.data.clone(); let keys = self.keys.clone(); + let clients = self.clients.clone(); let models = self.models.clone(); let request_limits = self.request_limits.clone(); let affinity = self.affinity.clone(); @@ -513,10 +497,14 @@ impl CallbackHandler for PluginCallbacks { } if matches!( method.as_str(), - "host.models.list" - | gateway_plugin_sdk::call::data::CLIENT_VERIFY + gateway_plugin_sdk::call::data::CLIENT_VERIFY | gateway_plugin_sdk::call::data::ACCESS_CHECK ) { + return clients + .call(&context, &call, &method, params, payload) + .await; + } + if method == "host.models.list" { return models.call(&context, &call, &method, params, payload).await; } if matches!( diff --git a/backend/crates/gateway-plugin/runtime/src/callback/model.rs b/backend/crates/gateway-plugin/runtime/src/callback/model.rs index 5f8dae51a..624782ded 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/model.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/model.rs @@ -1,4 +1,4 @@ -//! 复用 Core 的身份查询与模型执行入口；不拥有权限、路由、重试或计费规则 +//! 模型目录与执行适配；身份解析交给客户端端口，路由与计费沿用 Core use std::sync::{Arc, OnceLock, Weak}; @@ -8,7 +8,6 @@ use gateway_core::{ account::ProviderAccountId, engine::{ CommitRequirement, ModelRequestId, - authentication::ClientAuthenticationRequest, execution::{ AuthenticatedClient, ClientTransport, ExecutionRequestMetadata, ExecutionSession, }, @@ -22,12 +21,10 @@ use gateway_core::{ GenerateRequest, ImageRequest, ImageRequestKind, Operation, ProtocolPayload, RawJsonPayload, StandaloneSearchRequest, }, - policy::ClientApiKeyId, - routing::{PublicModelId, UpstreamModelId}, + routing::PublicModelId, }; use gateway_plugin_sdk::{ CallContext, ErrorCode, PluginFault, - call::data, call::host::{ ModelEventBatch, ModelExecuteRequest, ModelExecuteResult, ModelListRequest, ModelListResult, ModelOperation, ModelStreamCloseRequest, ModelStreamReadRequest, @@ -39,7 +36,6 @@ use super::{CallResources, denied, invalid}; use crate::RpcReply; const MAX_MODEL_STREAMS_PER_CALL: usize = 16; -const MAX_CLIENT_IDENTITIES_PER_CALL: usize = 16; pub(crate) struct PluginModelPortSlot { port: OnceLock>, @@ -65,6 +61,7 @@ impl PluginModelPortSlot { pub(super) struct PluginModels { slot: Arc, + clients: Arc, request_limits: Arc, maximum_payload: usize, } @@ -72,11 +69,13 @@ pub(super) struct PluginModels { impl PluginModels { pub(super) fn new( slot: Arc, + clients: Arc, request_limits: Arc, maximum_payload: usize, ) -> Self { Self { slot, + clients, request_limits, maximum_payload, } @@ -91,64 +90,6 @@ impl PluginModels { payload: Vec, ) -> Result { match method { - data::CLIENT_VERIFY => { - if params != serde_json::json!({}) { - return Err(invalid()); - } - let request: data::VerifyClientRequest = - serde_json::from_slice(&payload).map_err(|_| invalid())?; - let mut authentication = ClientAuthenticationRequest::new(request.authorization) - .map_err(|_| invalid())? - .with_invocation(call.child_origin(context)?); - if let Some(settings) = call.request_settings() { - authentication = authentication.with_settings(settings); - } - let port = self.slot.upgrade()?; - call.observe_dispatch()?; - let client = port - .verify_client(authentication) - .await - .map_err(super::error::gateway)?; - super::admin::encode(&data::VerifiedClient { - client_key_id: client.policy().key_id().as_str().to_owned(), - }) - } - data::ACCESS_CHECK => { - if params != serde_json::json!({}) { - return Err(invalid()); - } - let request: data::AccessQuery = - serde_json::from_slice(&payload).map_err(|_| invalid())?; - let port = self.slot.upgrade()?; - let client = self.identity(&port, call, request.client_key_id).await?; - let scope = client.policy().account_scope(); - let model = request - .upstream_model - .map(UpstreamModelId::new) - .transpose() - .map_err(|_| invalid())?; - let accounts = request - .account_ids - .into_iter() - .map(|account_id| { - let account = - ProviderAccountId::new(account_id.clone()).map_err(|_| invalid())?; - let allowed = model.as_ref().map_or_else( - || scope.allows(&account), - |model| scope.allows_model(&account, model.as_str()), - ); - Ok(data::AccountAccessFacts { - account_id, - allowed, - }) - }) - .collect::, PluginFault>>()?; - super::admin::encode(&data::AccessFacts { - schema_version: 1, - client_key_id: client.policy().key_id().as_str().to_owned(), - accounts, - }) - } "host.models.list" => { if !payload.is_empty() { return Err(invalid()); @@ -162,7 +103,7 @@ impl PluginModels { return Err(invalid()); } let port = self.slot.upgrade()?; - let client = self.identity(&port, call, request.client_key_id).await?; + let client = self.clients.identity(call, request.client_key_id).await?; let models = port .models( client, @@ -196,7 +137,7 @@ impl PluginModels { let port = self.slot.upgrade()?; let stream = method == "host.model.execute_stream"; let started = if let Some(client_key_id) = request.client_key_id.take() { - let client = self.identity(&port, call, client_key_id).await?; + let client = self.clients.identity(call, client_key_id).await?; check_size(client.snapshot().responses_max_decompressed_body_bytes())?; let request = self.request_parts(request, payload, stream)?; let request = Self::bound_request(context, call, request, client); @@ -295,38 +236,6 @@ impl PluginModels { } } - async fn identity( - &self, - port: &Arc, - call: &CallResources, - client_key_id: String, - ) -> Result { - let key = ClientApiKeyId::new(client_key_id.clone()).map_err(|_| invalid())?; - let settings = call.request_settings(); - let mut identities = call.client_identities.lock().await; - if let Some(client) = identities - .get(&client_key_id) - .filter(|client| client.request_settings() == settings.as_ref()) - { - return Ok(client.clone()); - } - if !identities.contains_key(&client_key_id) - && identities.len() >= MAX_CLIENT_IDENTITIES_PER_CALL - { - return Err(PluginFault::new( - ErrorCode::Capacity, - "client identity capacity is exhausted", - )); - } - let client = port - .resolve_client(key, settings) - .await - .map_err(super::error::gateway)?; - // 查询与执行共用调用内快照，执行链和子执行取消只在真正启动时建立 - identities.insert(client_key_id, client.clone()); - Ok(client) - } - fn request( &self, context: &CallContext, diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs index aaa0d0653..8667ceac9 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs @@ -161,15 +161,16 @@ impl PluginRuntime { &manifest, self.log_slots.clone(), private_state.clone(), - PluginCallbackPorts::new( - self.service_ports.clone(), - self.http.clone(), - self.account_ports.clone(), - self.client_key_ports.clone(), - self.model_ports.clone(), - self.affinity_ports.clone(), - self.resource_ports.clone(), - ), + PluginCallbackPorts { + services: self.service_ports.clone(), + http: self.http.clone(), + accounts: self.account_ports.clone(), + keys: self.client_key_ports.clone(), + clients: self.client_identity_port.clone(), + models: self.model_ports.clone(), + affinity: self.affinity_ports.clone(), + resources: self.resource_ports.clone(), + }, self.request_limits.clone(), )?; callbacks.registration_body_limit = request_body_limit; diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs index b4e52b9bb..d708eb950 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs @@ -8,7 +8,8 @@ use super::{ use crate::{ PackageLimits, RpcLimits, RpcSession, callback::{ - PluginAccountPortSlot, PluginAffinityPortSlot, PluginClientKeyPortSlot, PluginModelPortSlot, + PluginAccountPortSlot, PluginAffinityPortSlot, PluginClientIdentityPortSlot, + PluginClientKeyPortSlot, PluginModelPortSlot, }, }; use gateway_admin::{ @@ -21,7 +22,7 @@ use gateway_admin::{ }; use gateway_core::{ engine::{ - authentication::FrontendAuthenticationExtensionIndex, + authentication::{ClientIdentityPort, FrontendAuthenticationExtensionIndex}, extensions::ExecutionExtensionIndex, nested::{AffinityLookupPort, NestedModelExecutionPort}, observation::RequestObserverExtensionIndex, @@ -69,6 +70,7 @@ pub struct PluginRuntime { pub(super) account_ports: Arc, pub(super) client_key_ports: Arc, pub(super) resource_ports: Arc, + pub(super) client_identity_port: Arc, pub(super) model_ports: Arc, pub(super) affinity_ports: Arc, pub(super) observers: RequestObserverExtensionIndex, @@ -113,6 +115,7 @@ impl PluginRuntime { account_ports: Arc::new(PluginAccountPortSlot::new()), client_key_ports: Arc::new(PluginClientKeyPortSlot::new()), resource_ports: Arc::new(crate::callback::PluginResourcePorts::new()), + client_identity_port: Arc::new(PluginClientIdentityPortSlot::new()), model_ports: Arc::new(PluginModelPortSlot::new()), affinity_ports: Arc::new(PluginAffinityPortSlot::new()), observers: RequestObserverExtensionIndex::default(), @@ -168,6 +171,14 @@ impl PluginRuntime { self.client_key_ports.bind(access) } + /// 身份查询独立装配，复用 Core 的认证与冻结权限，不要求绑定模型执行端口 + pub fn bind_client_identity_port( + &self, + clients: &Arc, + ) -> Result<(), AdminError> { + self.client_identity_port.bind(clients) + } + /// Core 激活后一次性绑定嵌套执行与亲和查询端口；Runtime 仅保存 Weak，避免组合根强环 pub fn bind_model_ports( &self, diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/authentication.rs b/backend/crates/gateway-plugin/runtime/tests/generation/authentication.rs index f9660085a..0814e2893 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/authentication.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/authentication.rs @@ -276,6 +276,15 @@ async fn startup_failure_of_an_unrelated_plugin_preserves_native_authentication_ #[tokio::test] async fn callback_verification_reenters_the_authentication_instance_without_model_binding() { + assert_callback_identity_queries(true).await; +} + +#[tokio::test] +async fn client_queries_do_not_require_model_execution_ports() { + assert_callback_identity_queries(false).await; +} + +async fn assert_callback_identity_queries(include_models: bool) { use gateway_admin::{ model::plugins::management::PluginManagementRequest, ports::plugin_management::PluginManagement, @@ -292,10 +301,14 @@ async fn callback_verification_reenters_the_authentication_instance_without_mode .directory .path() .join("authenticated-facts.json"); - let facts = json!([ - {"method":"host.data.access.check", "query":{"client_key_id":"key_frontend", "account_ids":[account.as_str(), "acct_missing"]}}, - {"method":"host.models.list", "params":{"client_key_id":"key_frontend", "protocol":"codex", "client_version":"0.154.0"}} - ]); + let mut facts = vec![json!( + {"method":"host.data.access.check", "query":{"client_key_id":"key_frontend", "account_ids":[account.as_str(), "acct_missing"]}} + )]; + if include_models { + facts.push(json!( + {"method":"host.models.list", "params":{"client_key_id":"key_frontend", "protocol":"codex", "client_version":"0.154.0"}} + )); + } let mut config = configuration( json!({"outcome":"authenticated","principal":"external-user"}), true, @@ -303,20 +316,32 @@ async fn callback_verification_reenters_the_authentication_instance_without_mode ); config["sdk_session"] = json!(true); config["frontend_authentication_marker"] = json!(marker); - config["auth_queries"] = facts.clone(); + config["auth_queries"] = json!(facts); config["management_registration"] = json!({"routes":[{"method":"GET","path":"facts","request_content_types":[],"response_content_types":["application/json"]}]}); - config["data_queries"] = json!([ - {"method":"host.client.verify", "query":{"authorization":"External controlled-fixture"}}, - facts[0].clone(), facts[1].clone() - ]); + let mut queries = vec![json!( + {"method":"host.client.verify", "query":{"authorization":"External controlled-fixture"}} + )]; + queries.extend(facts); + config["data_queries"] = json!(queries); environment.install_plugin(config).await; let (runtime, core) = environment - .runtime_with_limits(RpcLimits { - maximum_executing_calls: 1, - maximum_queue_timeout: std::time::Duration::from_millis(500), - ..RpcLimits::default() - }) + .runtime_with_identity_port( + RpcLimits { + maximum_executing_calls: 1, + maximum_queue_timeout: std::time::Duration::from_millis(500), + ..RpcLimits::default() + }, + crate::support::native::provider_registry(), + ) .await; + if include_models { + runtime + .bind_model_ports( + &core.nested_model_execution_port(), + &core.affinity_lookup_port(), + ) + .unwrap(); + } let snapshot = core.snapshots().acquire().unwrap(); let generation = snapshot.extensions().unwrap(); let view = runtime.views(generation).await.unwrap().remove(0); @@ -348,9 +373,11 @@ async fn callback_verification_reenters_the_authentication_instance_without_mode {"account_id":account.as_str(),"allowed":true}, {"account_id":"acct_missing","allowed":false} ]) ); - assert_eq!(authenticated_facts[1], results[2]); - // 此替身只登记执行能力，没有原生目录；查询应成功返回空目录而不是执行绑定错误 - assert_eq!(results[2], json!({"models":[]})); + if include_models { + assert_eq!(authenticated_facts[1], results[2]); + // 此替身只登记执行能力，没有原生目录；查询应成功返回空目录而不是执行绑定错误 + assert_eq!(results[2], json!({"models":[]})); + } assert!( environment .bound_model_requests("key_frontend") diff --git a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs index 1a7c30463..11ffbd3a7 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs @@ -463,6 +463,21 @@ impl Environment { &self, limits: RpcLimits, providers: gateway_core::engine::provider::ProviderRegistry, + ) -> (Arc, gateway_core::CoreBundle) { + let (runtime, core) = self.runtime_with_identity_port(limits, providers).await; + runtime + .bind_model_ports( + &core.nested_model_execution_port(), + &core.affinity_lookup_port(), + ) + .unwrap(); + (runtime, core) + } + + pub async fn runtime_with_identity_port( + &self, + limits: RpcLimits, + providers: gateway_core::engine::provider::ProviderRegistry, ) -> (Arc, gateway_core::CoreBundle) { let runtime = self.plugin_runtime_with_limits(limits); let core = gateway_core::prepare( @@ -487,10 +502,7 @@ impl Environment { .insert(Arc::as_ptr(&runtime) as usize, access); let core = core.activate().await.unwrap(); runtime - .bind_model_ports( - &core.nested_model_execution_port(), - &core.affinity_lookup_port(), - ) + .bind_client_identity_port(&core.client_identity_port()) .unwrap(); (runtime, core) } @@ -511,6 +523,9 @@ impl Environment { .await .unwrap(); runtime.bind_snapshots(&core.snapshots()).unwrap(); + runtime + .bind_client_identity_port(&core.client_identity_port()) + .unwrap(); runtime .bind_model_ports( &core.nested_model_execution_port(), diff --git a/docs/architecture.md b/docs/architecture.md index 2db106f05..926fafe63 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -112,6 +112,7 @@ flowchart LR Runtime <-->|SDK 协议| Plugin["插件进程"] Runtime -->|受管资源| Host["Host
进程、网络、Worker"] Runtime -->|账号事务端口| Admin + Runtime -->|身份查询端口| Core Runtime -->|模型执行端口| Core ``` @@ -183,8 +184,9 @@ flowchart LR Provider 来源与执行事件快照可由插件读取，包含费用明细、上游观测、会话状态和原始 wire；读取快照不转移计量和资源所有权。 数据面插件通过受管 HTTP 已发送、无法证明未发送，或进入主动 HTTP／公开服务调用链时，Runtime 把该事实并入 Core 的请求副作用水位，后续不能按 Provider 的 `not_sent` 结果透明重放；RPC 返回或取消前同步封存未完成的发送，并禁止已关闭调用再分派 HTTP／公开服务 -- **身份查询与执行准入分离**：Core 的 `AuthenticatedClient` 持有身份和冻结权限，模型目录与账号授权直接读取这份事实。 - Runtime 只按父调用缓存身份快照，查询不建立模型执行图；真正发起模型执行时才检查深度、建立执行归属并申请原生资源。 +- **身份查询与执行准入分离**：Core 通过 `ClientIdentityPort` 独立提供身份解析和认证，`AuthenticatedClient` 持有身份和冻结权限，模型目录与账号授权直接读取这份事实。 + Runtime 的客户端适配器承接认证与授权查询，模型适配器复用同一身份解析。 + 身份快照只按父调用缓存，查询不建立模型执行图；真正发起模型执行时才检查深度、建立执行归属并申请原生资源。 认证复用原有认证计划与 principal 映射，通用 `InvocationOrigin` 传递父调用来源、取消、冻结计划和副作用归属，认证不依赖模型执行上下文 - **挂起与恢复由共同边界管理**：SDK 在父处理 future 返回 `Pending` 后报告挂起，再次 poll 前等待 Runtime 的执行授权。 Runtime 的调用表持有唯一准入状态，挂起保留总在途名额；结果接收独立于任务恢复，回调额度持续保留到消费、丢弃或父调用终结。