From 5c2a1ddf6ac0a4476e852b72e73a9741dc097f21 Mon Sep 17 00:00:00 2001 From: Nick Dujay Date: Tue, 20 Oct 2026 20:39:43 -0400 Subject: [PATCH 1/2] Prepare template lookup scope for rolling upgrades --- Cargo.lock | 1 + crates/libs/common/svc/src/lib.rs | 1 + crates/libs/common/svc/src/template_lookup.rs | 125 ++++++++++++++++++ .../libs/internal/api/server/src/adapter.rs | 25 +++- crates/libs/metrics/src/auth.rs | 9 ++ crates/libs/node/api/server/src/builder.rs | 12 +- .../server/src/clients/policy_auth_tests.rs | 53 +++++++- .../clients/template_transfer_auth_tests.rs | 7 + crates/libs/node/api/server/src/lib.rs | 1 + .../node/api/server/src/template_lookup.rs | 40 ++++++ crates/libs/node/api/server/src/transfer.rs | 68 +++++----- crates/libs/node/api/server/src/worker.rs | 15 ++- crates/libs/sbx/mgr/server/Cargo.toml | 1 + .../sbx/mgr/server/src/service/auth_lookup.rs | 1 + .../mgr/server/src/service/create_support.rs | 65 +++++++-- .../src/service/create_support/tests.rs | 65 +++++++++ .../sbx/mgr/server/src/service/grpc_core.rs | 17 +++ .../grpc_core/create_after_lookup_tests.rs | 5 +- .../grpc_core/create_template_lookup_tests.rs | 24 +++- .../src/service/grpc_pool_wait_clone.rs | 9 +- crates/libs/sbx/mgr/server/src/service/mod.rs | 1 + .../mgr/server/src/service/network_policy.rs | 34 ++--- .../server/src/service/recovery_cross_node.rs | 5 + .../libs/sbx/mgr/server/src/service/start.rs | 5 + .../sbx/mgr/server/src/service/start/tests.rs | 1 + .../sbx/mgr/server/src/snapshot_restore.rs | 46 ++++--- .../mgr/server/src/snapshot_restore/tests.rs | 52 ++++++++ .../libs/template/mgr/src/service/lookup.rs | 32 +++-- crates/libs/template/mgr/src/service/tests.rs | 1 + .../mgr/src/service/tests/lookup_rpc.rs | 100 ++++++++++++++ crates/libs/template/mgr/src/volume.rs | 81 ++++++++---- .../template/mgr/src/volume/tests/catalog.rs | 68 ++++++++++ docs/architecture.md | 20 +++ 33 files changed, 856 insertions(+), 134 deletions(-) create mode 100644 crates/libs/common/svc/src/template_lookup.rs create mode 100644 crates/libs/node/api/server/src/template_lookup.rs create mode 100644 crates/libs/template/mgr/src/service/tests/lookup_rpc.rs diff --git a/Cargo.lock b/Cargo.lock index 5342af4e78..1f67c6cfc0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6890,6 +6890,7 @@ dependencies = [ "prost", "prost-types", "rcgen", + "rstest", "rustix", "rustls", "serde", diff --git a/crates/libs/common/svc/src/lib.rs b/crates/libs/common/svc/src/lib.rs index a3333c9754..1baba99802 100644 --- a/crates/libs/common/svc/src/lib.rs +++ b/crates/libs/common/svc/src/lib.rs @@ -38,6 +38,7 @@ pub mod space_api_route; pub mod space_api_route_capability; pub mod space_api_route_ed25519; pub mod sql; +pub mod template_lookup; pub use pplx_space_systemd_watchdog as systemd; pub use pplx_space_tokio_runtime as tokio_runtime; pub mod types; diff --git a/crates/libs/common/svc/src/template_lookup.rs b/crates/libs/common/svc/src/template_lookup.rs new file mode 100644 index 0000000000..86ef6fa683 --- /dev/null +++ b/crates/libs/common/svc/src/template_lookup.rs @@ -0,0 +1,125 @@ +//! Template lookup scope propagation during a mixed-version rollout. +//! Receipts confirm protocol support, not authorization. Existing access checks +//! remain authoritative until a separate enforcement protocol is deployed. + +use std::net::IpAddr; + +use pplx_space_common_core::types::SandboxId; +use tonic::{Request, Response, Status}; + +const SCOPE_HEADER: &str = "x-space-template-lookup-scope"; +const RECEIPT_HEADER: &str = "x-space-template-lookup-scope-received"; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TemplateLookupScope { + /// Online lookup using the service's catalog authority and optional org filter. + Cluster, + /// Original authenticated node before forwarding. + Node(IpAddr), + /// Sandbox identity and, when forwarded, its original node. + Sandbox { + id: SandboxId, + caller_node: Option, + }, + /// Execution requested by an authenticated node-local principal. + LocalExecution, +} + +impl std::fmt::Display for TemplateLookupScope { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Cluster => f.write_str("cluster"), + Self::Node(address) => write!(f, "node:{address}"), + Self::Sandbox { id, caller_node } => { + write!(f, "sandbox:{id}")?; + if let Some(address) = caller_node { + write!(f, "@{address}")?; + } + Ok(()) + } + Self::LocalExecution => f.write_str("local"), + } + } +} + +impl std::str::FromStr for TemplateLookupScope { + type Err = Status; + + fn from_str(value: &str) -> Result { + match value { + "cluster" => Ok(Self::Cluster), + "local" => Ok(Self::LocalExecution), + _ => { + if let Some(address) = value.strip_prefix("node:") { + return address + .parse() + .map(Self::Node) + .map_err(|_| Status::invalid_argument("invalid template caller address")); + } + if let Some(id) = value.strip_prefix("sandbox:") { + let (id, address) = id + .split_once('@') + .map_or((id, None), |(id, address)| (id, Some(address))); + return Ok(Self::Sandbox { + id: id.parse().map_err(|_| { + Status::invalid_argument("invalid template sandbox identity") + })?, + caller_node: address.map(str::parse).transpose().map_err(|_| { + Status::invalid_argument("invalid template caller address") + })?, + }); + } + Err(Status::invalid_argument("invalid template lookup scope")) + } + } + } +} + +impl TemplateLookupScope { + pub fn from_request(request: &Request) -> Result, Status> { + request + .metadata() + .get(SCOPE_HEADER) + .map(|value| { + value + .to_str() + .map_err(|_| Status::invalid_argument("invalid template lookup scope"))? + .parse() + }) + .transpose() + } + + pub fn attach(self, request: &mut Request) { + request.metadata_mut().insert( + SCOPE_HEADER, + self.to_string().parse().expect("scope is ASCII"), + ); + } + + /// A receipt must not satisfy the authorization acknowledgement used by strict peers. + pub fn receipt(self, response: &mut Response) { + response.metadata_mut().insert( + RECEIPT_HEADER, + self.to_string().parse().expect("scope is ASCII"), + ); + } + + pub fn observe_response(self, response: &Response) -> Result<(), Status> { + match response.metadata().get(RECEIPT_HEADER) { + None => { + pplx_space_metrics::auth::template_lookup_compatibility("missing_receipt"); + tracing::warn!( + "template lookup peer has no scope receipt; retaining legacy behavior" + ); + Ok(()) + } + Some(value) if value.to_str().ok() == Some(self.to_string().as_str()) => { + pplx_space_metrics::auth::template_lookup_compatibility("received"); + Ok(()) + } + Some(_) => Err(Status::failed_precondition( + "template lookup scope receipt mismatch", + )), + } + } +} diff --git a/crates/libs/internal/api/server/src/adapter.rs b/crates/libs/internal/api/server/src/adapter.rs index ff1046d69e..71790bc609 100644 --- a/crates/libs/internal/api/server/src/adapter.rs +++ b/crates/libs/internal/api/server/src/adapter.rs @@ -1,4 +1,5 @@ use pplx_space_common_core::types::OrgId; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; use super::metadata::*; use super::status::*; @@ -394,6 +395,10 @@ impl ClusterApi for ClusterApiService { request: Request, ) -> Result, Status> { let scope = self.template_read_scope(&request).await?; + let delegation = TemplateLookupScope::from_request(&request)?; + if delegation.is_none() { + pplx_space_metrics::auth::template_lookup_compatibility("missing_scope"); + } let request = request.into_inner(); let template_name = trimmed_required(&request.template_name, "template_name")?; let runtime = validated_template_runtime(&request.runtime)?; @@ -454,13 +459,17 @@ impl ClusterApi for ClusterApiService { &template_name, ) .await; - Ok(Response::new(FindTemplateReleaseResponse { + let mut response = Response::new(FindTemplateReleaseResponse { generic_authorized, authorized_org_id: caller_org.map(|org| org.to_string()), release: Some(release), storage_locations, template_visibility: visibility.into(), - })) + }); + if let Some(scope) = delegation { + scope.receipt(&mut response); + } + Ok(response) } async fn find_template_release_by_hash( @@ -468,6 +477,10 @@ impl ClusterApi for ClusterApiService { request: Request, ) -> Result, Status> { let scope = self.template_read_scope(&request).await?; + let delegation = TemplateLookupScope::from_request(&request)?; + if delegation.is_none() { + pplx_space_metrics::auth::template_lookup_compatibility("missing_scope"); + } let request = request.into_inner(); let template_name = trimmed_required(&request.template_name, "template_name")?; // Canonical `sha256:`-prefixed form. @@ -538,13 +551,17 @@ impl ClusterApi for ClusterApiService { &template_name, ) .await; - Ok(Response::new(FindTemplateReleaseResponse { + let mut response = Response::new(FindTemplateReleaseResponse { generic_authorized, authorized_org_id: caller_org.map(|org| org.to_string()), release: Some(release), storage_locations, template_visibility: visibility.into(), - })) + }); + if let Some(scope) = delegation { + scope.receipt(&mut response); + } + Ok(response) } async fn list_latest_template_releases( diff --git a/crates/libs/metrics/src/auth.rs b/crates/libs/metrics/src/auth.rs index 80aae1d57c..33d2faf667 100644 --- a/crates/libs/metrics/src/auth.rs +++ b/crates/libs/metrics/src/auth.rs @@ -100,3 +100,12 @@ pub fn uds_peer_exe_deleted(service: &str, principal: &str) { &[tag("service", service), tag("principal", principal)], ); } + +/// Per-hop scope protocol support; a receipt does not certify authorization. +pub fn template_lookup_compatibility(result: &str) { + counter( + "auth.template_lookup.compatibility", + 1, + &[tag("result", result)], + ); +} diff --git a/crates/libs/node/api/server/src/builder.rs b/crates/libs/node/api/server/src/builder.rs index 2660e325d7..bcf4306a7a 100644 --- a/crates/libs/node/api/server/src/builder.rs +++ b/crates/libs/node/api/server/src/builder.rs @@ -152,9 +152,12 @@ impl BuilderApi for BuilderApiService { async fn get_template( &self, - r: Request, + mut r: Request, ) -> Result, Status> { - fwd_retry!("template-manager", self.clients.tmpl(), get_template, r) + let scope = crate::template_lookup::bind_caller_scope(&mut r)?; + let response = fwd_retry!("template-manager", self.clients.tmpl(), get_template, r)?; + scope.observe_response(&response)?; + Ok(response) } async fn update_template_settings( @@ -244,11 +247,12 @@ impl BuilderApi for BuilderApiService { async fn pull_template( &self, - r: Request, + mut r: Request, ) -> Result, Status> { pplx_space_common_svc::auth::template_transfer::require_template_transfer_authority(&r)?; + let scope = crate::template_lookup::bind_caller_scope(&mut r)?; let req = r.into_inner(); - let resp = crate::transfer::pull_template(&mut self.clients.tmpl(), &req).await?; + let resp = crate::transfer::pull_template(&mut self.clients.tmpl(), &req, scope).await?; Ok(Response::new(resp)) } diff --git a/crates/libs/node/api/server/src/clients/policy_auth_tests.rs b/crates/libs/node/api/server/src/clients/policy_auth_tests.rs index d893f006ac..c0555ac8af 100644 --- a/crates/libs/node/api/server/src/clients/policy_auth_tests.rs +++ b/crates/libs/node/api/server/src/clients/policy_auth_tests.rs @@ -118,6 +118,12 @@ pub(super) fn auth_layer( pub(super) fn bearer(body: T, token: &str) -> Request { let mut request = Request::new(body); + request + .extensions_mut() + .insert(tonic::transport::server::TcpConnectInfo { + local_addr: None, + remote_addr: Some(([127, 0, 0, 1], 1234).into()), + }); request.metadata_mut().insert( "authorization", format!("Bearer {token}") @@ -139,7 +145,9 @@ enum Rpc { #[tokio::test] async fn policy_rpc_authentication_reaches_handlers() { - let (clients, calls, shutdown, mut servers) = backend().await; + let forwarded_scope = Arc::new(std::sync::Mutex::new(None)); + let (clients, calls, shutdown, mut servers) = + backend_with_scope(Some(forwarded_scope.clone())).await; for (k8s, username, token, trusted) in [ (true, "space-api", PROJECTED_TOKEN, true), (true, "workers", PROJECTED_TOKEN, false), @@ -261,6 +269,15 @@ async fn policy_rpc_authentication_reaches_handlers() { .await .map(|_| ()), }; + if matches!(rpc, Rpc::Create) && result.is_ok() { + use pplx_space_common_svc::template_lookup::TemplateLookupScope; + let expected = if trusted { + TemplateLookupScope::Cluster + } else { + TemplateLookupScope::Node("127.0.0.1".parse().unwrap()) + }; + assert_eq!(*forwarded_scope.lock().unwrap(), Some(expected.to_string())); + } let requires_authority = match rpc { Rpc::Network | Rpc::Lifecycle => true, Rpc::Create | Rpc::Adopt => network || lifecycle || generation != 0, @@ -299,6 +316,17 @@ pub(super) async fn backend() -> ( Arc, tokio_util::sync::CancellationToken, tokio::task::JoinSet<()>, +) { + backend_with_scope(None).await +} + +async fn backend_with_scope( + forwarded_scope: Option>>>, +) -> ( + BackendClients, + Arc, + tokio_util::sync::CancellationToken, + tokio::task::JoinSet<()>, ) { let calls = Arc::new(AtomicUsize::new(0)); let recorded_calls = calls.clone(); @@ -306,6 +334,7 @@ pub(super) async fn backend() -> ( // the request; denial must leave it untouched. No manager/database is needed. let backend = Router::new().fallback(any(move |request: http::Request| { let calls = recorded_calls.clone(); + let scope = forwarded_scope.clone(); async move { if request.uri().path().ends_with("/ExportTemplate") { assert_eq!( @@ -316,8 +345,22 @@ pub(super) async fn backend() -> ( "granted" ); } + if let Some(scope) = scope { + *scope.lock().expect("scope recorder lock") = request + .headers() + .get("x-space-template-lookup-scope") + .map(|value| value.to_str().expect("ASCII scope header").to_owned()); + } calls.fetch_add(1, Ordering::SeqCst); + let lookup_scope = request + .headers() + .get("x-space-template-lookup-scope") + .cloned(); let payload = if request.uri().path().ends_with("/GetTemplate") { + assert!( + lookup_scope.is_some(), + "cache probes must carry authenticated scope" + ); use prost::Message; tmpl::GetTemplateResponse { template: Some(tmpl::TemplateRef { @@ -339,9 +382,13 @@ pub(super) async fn backend() -> ( .to_be_bytes(), ); frame.extend(payload); - http::Response::builder() + let mut response = http::Response::builder() .header("content-type", "application/grpc") - .header("grpc-status", "0") + .header("grpc-status", "0"); + if let Some(scope) = lookup_scope { + response = response.header("x-space-template-lookup-scope-received", scope); + } + response .body(Body::from(frame)) .expect("valid gRPC acknowledgement") } diff --git a/crates/libs/node/api/server/src/clients/template_transfer_auth_tests.rs b/crates/libs/node/api/server/src/clients/template_transfer_auth_tests.rs index f47ac5f0c5..acf3707d59 100644 --- a/crates/libs/node/api/server/src/clients/template_transfer_auth_tests.rs +++ b/crates/libs/node/api/server/src/clients/template_transfer_auth_tests.rs @@ -98,6 +98,13 @@ async fn template_transfer_authority_survives_the_wire( template_id: "example-template".into(), content_hash: Some(format!("sha256:{}", "a".repeat(64))), skip_if_exists: true, + sources: vec![crate::proto::TransferSource { + source: Some(crate::proto::transfer_source::Source::Node( + crate::proto::NodeSource { + address: "http://127.0.0.1:1".into(), + } + )), + }], ..Default::default() } ), diff --git a/crates/libs/node/api/server/src/lib.rs b/crates/libs/node/api/server/src/lib.rs index f72563f1d3..cce724cbce 100644 --- a/crates/libs/node/api/server/src/lib.rs +++ b/crates/libs/node/api/server/src/lib.rs @@ -15,6 +15,7 @@ pub mod node_ctl; pub mod node_info; mod restart_handoff; mod run; +mod template_lookup; pub mod transfer; pub mod uds_per_sandbox_limiter; pub mod worker; diff --git a/crates/libs/node/api/server/src/template_lookup.rs b/crates/libs/node/api/server/src/template_lookup.rs new file mode 100644 index 0000000000..9f9170ed07 --- /dev/null +++ b/crates/libs/node/api/server/src/template_lookup.rs @@ -0,0 +1,40 @@ +use pplx_space_common_svc::auth::{NodeApiAuthenticatedCaller, TemplateReleaseReadAuthority}; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; +use tonic::{Request, Status}; + +pub(crate) fn bind_caller_scope( + request: &mut Request, +) -> Result { + let caller = request + .extensions() + .get::() + .ok_or_else(|| Status::unauthenticated("template lookup requires authenticated caller"))?; + let scope = match caller.template_release_read_authority() { + TemplateReleaseReadAuthority::Unscoped => TemplateLookupScope::Cluster, + TemplateReleaseReadAuthority::NodeScoped => TemplateLookupScope::Node( + request + .remote_addr() + .ok_or_else(|| Status::permission_denied("template caller address is missing"))? + .ip(), + ), + }; + scope.attach(request); + Ok(scope) +} + +/// Record control-plane creation separately from a forwarded node request. +pub(crate) fn bind_create_caller_scope(request: &mut Request) -> Result<(), Status> { + use pplx_space_common_svc::auth::NetworkPolicyMutationAuthority; + if request + .extensions() + .get::() + .is_some_and(|caller| { + caller.network_policy_authority() == NetworkPolicyMutationAuthority::Granted + }) + { + TemplateLookupScope::Cluster.attach(request); + } else { + bind_caller_scope(request)?; + } + Ok(()) +} diff --git a/crates/libs/node/api/server/src/transfer.rs b/crates/libs/node/api/server/src/transfer.rs index da2d254c67..7cee05f8fc 100644 --- a/crates/libs/node/api/server/src/transfer.rs +++ b/crates/libs/node/api/server/src/transfer.rs @@ -1,6 +1,7 @@ use std::time::Duration; use pplx_space_common_svc::bootstrap::wrap_grpc_client_channel; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; use tokio_stream::StreamExt; use tonic::Status; use tonic::codegen::InterceptedService; @@ -154,6 +155,36 @@ fn optional_snapshot_dimensions( } } +async fn probe_cached_template( + client: &mut pplx_space_template_mgr_client::TemplateManagerGrpcClient, + pull: &PullTemplateRequest, + hash: &str, + scope: TemplateLookupScope, +) -> Result, Status> { + let mut request = + tonic::Request::new(pplx_space_template_mgr_client::proto::GetTemplateRequest { + caller_org_id: None, + template: Some(pplx_space_template_mgr_client::proto::TemplateRef { + name: pull.template_id.clone(), + content_hash: Some(hash.to_owned()), + tag: None, + }), + vcpu: pull.snapshot_vcpu, + memory_mib: pull.snapshot_memory_mib, + }); + scope.attach(&mut request); + let Ok(response) = pplx_space_common_grpc::fwd_retry!( + "template-manager", + client.clone(), + get_template, + request + ) else { + return Ok(None); + }; + scope.observe_response(&response)?; + Ok(Some(response.into_inner())) +} + /// Shared PullTemplate orchestration logic used by both WorkerApi and BuilderApi. /// /// 1. Checks if the template already exists locally (`skip_if_exists`). @@ -174,6 +205,7 @@ fn optional_snapshot_dimensions( pub async fn pull_template( tmpl_client: &mut pplx_space_template_mgr_client::TemplateManagerGrpcClient, req: &PullTemplateRequest, + scope: TemplateLookupScope, ) -> Result { let template_id = &req.template_id; @@ -191,23 +223,9 @@ pub async fn pull_template( // the remote's tag may point to a different build than the local one. if req.skip_if_exists && !req_content_hash.is_empty() - && let Ok(resp) = pplx_space_common_grpc::fwd_retry!( - "template-manager", - tmpl_client.clone(), - get_template, - tonic::Request::new(pplx_space_template_mgr_client::proto::GetTemplateRequest { - caller_org_id: None, - template: Some(pplx_space_template_mgr_client::proto::TemplateRef { - name: template_id.clone(), - content_hash: req.content_hash.clone(), - tag: None, - }), - vcpu: req.snapshot_vcpu, - memory_mib: req.snapshot_memory_mib, - }) - ) + && let Some(local) = + probe_cached_template(tmpl_client, req, req_content_hash, scope).await? { - let local = resp.into_inner(); let local_hash = local .template .as_ref() @@ -326,23 +344,9 @@ pub async fn pull_template( // For tag-based pulls: now that we know the resolved hash from the source, // check if this exact build (+ snapshot if requested) already exists locally. if req.skip_if_exists - && let Ok(resp) = pplx_space_common_grpc::fwd_retry!( - "template-manager", - tmpl_client.clone(), - get_template, - tonic::Request::new(pplx_space_template_mgr_client::proto::GetTemplateRequest { - caller_org_id: None, - template: Some(pplx_space_template_mgr_client::proto::TemplateRef { - name: template_id.clone(), - content_hash: Some(export_content_hash.to_string()), - tag: None, - }), - vcpu: req.snapshot_vcpu, - memory_mib: req.snapshot_memory_mib, - }) - ) + && let Some(local) = + probe_cached_template(tmpl_client, req, export_content_hash, scope).await? { - let local = resp.into_inner(); let local_hash = local .template .as_ref() diff --git a/crates/libs/node/api/server/src/worker.rs b/crates/libs/node/api/server/src/worker.rs index 05240c88b3..f57256fe2a 100644 --- a/crates/libs/node/api/server/src/worker.rs +++ b/crates/libs/node/api/server/src/worker.rs @@ -575,12 +575,13 @@ impl WorkerApi for WorkerApiService { async fn create_sandbox( &self, - r: Request, + mut r: Request, ) -> Result, Status> { // Before any check, so a refused create names its sandbox too. if let Some(sandbox_id) = r.get_ref().sandbox_id.as_deref() { record_sandbox_id(&tracing::Span::current(), sandbox_id); } + crate::template_lookup::bind_create_caller_scope(&mut r)?; // Overrides and nonzero generations require policy authority. A generation // persists even without an override and can block later sandbox-specific // policy corrections. No override and generation zero retain inheritance. @@ -1216,9 +1217,12 @@ impl WorkerApi for WorkerApiService { async fn get_template( &self, - r: Request, + mut r: Request, ) -> Result, Status> { - fwd_retry!("template-manager", self.clients.tmpl(), get_template, r) + let scope = crate::template_lookup::bind_caller_scope(&mut r)?; + let response = fwd_retry!("template-manager", self.clients.tmpl(), get_template, r)?; + scope.observe_response(&response)?; + Ok(response) } async fn update_template_settings( @@ -1328,11 +1332,12 @@ impl WorkerApi for WorkerApiService { async fn pull_template( &self, - r: Request, + mut r: Request, ) -> Result, Status> { pplx_space_common_svc::auth::template_transfer::require_template_transfer_authority(&r)?; + let scope = crate::template_lookup::bind_caller_scope(&mut r)?; let req = r.into_inner(); - let resp = crate::transfer::pull_template(&mut self.clients.tmpl(), &req).await?; + let resp = crate::transfer::pull_template(&mut self.clients.tmpl(), &req, scope).await?; Ok(Response::new(resp)) } diff --git a/crates/libs/sbx/mgr/server/Cargo.toml b/crates/libs/sbx/mgr/server/Cargo.toml index 7e9618bc2d..576462135b 100644 --- a/crates/libs/sbx/mgr/server/Cargo.toml +++ b/crates/libs/sbx/mgr/server/Cargo.toml @@ -82,6 +82,7 @@ pplx-space-sbx-mgr-store = { workspace = true } pplx-space-template-mgr-client = { workspace = true } [dev-dependencies] +rstest = { workspace = true } pplx-space-common-core = { workspace = true, features = ["serde", "test-hooks"] } pplx-space-sbx-mgr-store = { workspace = true, features = ["test-support"] } pplx-space-net-gateway = { workspace = true, features = ["test-support"] } diff --git a/crates/libs/sbx/mgr/server/src/service/auth_lookup.rs b/crates/libs/sbx/mgr/server/src/service/auth_lookup.rs index 28918264ad..03e3a03c87 100644 --- a/crates/libs/sbx/mgr/server/src/service/auth_lookup.rs +++ b/crates/libs/sbx/mgr/server/src/service/auth_lookup.rs @@ -992,6 +992,7 @@ impl SandboxManagerService { content_hash: sandbox.template_hash.clone(), tag: None, }, + template_lookup_scope(&sandbox.user, Some(&sandbox.id))?, ) .await?; policy = resolved_template.default_auto_lifecycle; diff --git a/crates/libs/sbx/mgr/server/src/service/create_support.rs b/crates/libs/sbx/mgr/server/src/service/create_support.rs index a85b1f2000..d4021e4af0 100644 --- a/crates/libs/sbx/mgr/server/src/service/create_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/create_support.rs @@ -9,6 +9,7 @@ use super::validation::{ }; use super::*; use pplx_space_common_core::types::{EnvMap, EnvSanitizationIssue}; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; use pplx_space_events::{ SandboxEventDetails, SandboxEventSource, SandboxEventType, SandboxRestoreProvenance, emit_sandbox_event, @@ -652,20 +653,42 @@ pub(super) fn checkpoint_start_response_for_existing_started_sandbox( } } +pub(crate) fn template_lookup_scope( + owner: &SandboxOwner, + sandbox_id: Option<&str>, +) -> Result { + if matches!(owner.kind(), SandboxOwnerKind::PlatformUser(_)) { + let Some(sandbox_id) = sandbox_id else { + return Ok(TemplateLookupScope::Cluster); + }; + let id = sandbox_id + .parse() + .map_err(|_| Status::invalid_argument("invalid template sandbox identity"))?; + Ok(TemplateLookupScope::Sandbox { + id, + caller_node: None, + }) + } else { + Ok(TemplateLookupScope::LocalExecution) + } +} + pub(super) async fn resolve_template_paths( templates: &dyn crate::template_resolver::TemplateResolver, template_ref: pplx_space_template_mgr_client::proto::TemplateRef, + scope: TemplateLookupScope, ) -> Result { use crate::template_resolver::TemplateLookupError; - let inner = templates - .get_template(tonic::Request::new( - pplx_space_template_mgr_client::proto::GetTemplateRequest { - caller_org_id: None, - template: Some(template_ref), - vcpu: None, - memory_mib: None, - }, - )) + let mut request = + tonic::Request::new(pplx_space_template_mgr_client::proto::GetTemplateRequest { + caller_org_id: None, + template: Some(template_ref), + vcpu: None, + memory_mib: None, + }); + scope.attach(&mut request); + let response = templates + .get_template(request) .await .map_err(|error| match error { TemplateLookupError::Connect(e) => { @@ -674,8 +697,9 @@ pub(super) async fn resolve_template_paths( TemplateLookupError::Rpc(e) => { tonic::Status::new(e.code(), format!("GetTemplate failed: {e}")) } - })? - .into_inner(); + })?; + scope.observe_response(&response)?; + let inner = response.into_inner(); let tmpl = inner.template.unwrap_or_default(); let arch = template_metadata_arch(&inner.template_dir).unwrap_or(inner.arch); let spaced_compatibility = template_metadata_spaced_compatibility(&inner.template_dir); @@ -769,6 +793,7 @@ pub(super) fn stage_base_snapshot_runtime_metadata( pub(super) async fn resolve_template_paths_for_start( templates: &dyn crate::template_resolver::TemplateResolver, template_ref: pplx_space_template_mgr_client::proto::TemplateRef, + sandbox: &crate::db::SandboxRow, ) -> Result { // Start restores an existing sandbox's identity. A name-only lookup could // silently pick a new image after an alias moves; only create may do that. @@ -781,9 +806,13 @@ pub(super) async fn resolve_template_paths_for_start( "cannot recover sandbox artifact without its saved template hash", )); } - resolve_template_paths(templates, template_ref) - .await - .map_err(start_template_resolution_failed_status) + resolve_template_paths( + templates, + template_ref, + template_lookup_scope(&sandbox.user, Some(&sandbox.id))?, + ) + .await + .map_err(start_template_resolution_failed_status) } fn start_template_resolution_failed_status(error: Status) -> Status { @@ -814,6 +843,13 @@ pub(super) async fn template_runtime_defaults_for_sandbox_start( content_hash: sandbox.template_hash.clone(), tag: None, }, + match template_lookup_scope(&sandbox.user, Some(&sandbox.id)) { + Ok(scope) => scope, + Err(error) => { + tracing::warn!(%error, "invalid sandbox template scope"); + return TemplateRuntimeDefaults::default(); + } + }, ) .await { @@ -909,6 +945,7 @@ pub(super) async fn template_source_for_sandbox_prepare_in( content_hash: sandbox.template_hash.clone(), tag: None, }, + sandbox, ) .await?; reject_incompatible_template_arch( diff --git a/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs b/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs index 647d1bc13f..4669f8437d 100644 --- a/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs @@ -476,6 +476,14 @@ fn restore_provenance_classifies_legacy_remote_metadata() { #[tokio::test] async fn existing_sandbox_cannot_resolve_a_missing_artifact_by_latest_alias() { + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + crate::service::tests::service_tests::insert_test_sandbox( + &conn, + &id, + crate::state::SandboxState::Stopped, + ); + let sandbox = crate::db::get_sandbox(&conn, &id).unwrap().unwrap(); // This must reject locally rather than asking template-manager to pick // the newest build after the sandbox lost its identity. let templates = crate::template_resolver::fake::FakeTemplateResolver::refusing( @@ -488,6 +496,7 @@ async fn existing_sandbox_cannot_resolve_a_missing_artifact_by_latest_alias() { content_hash: None, tag: None, }, + &sandbox, ) .await .err() @@ -774,3 +783,59 @@ fn template_provenance_does_not_overwrite_a_selected_cold_boot_consumer() { bind_disk_execution(Some(&selected), other_kernel).unwrap_err(); assert_eq!(bind_disk_execution(None, source.clone()).unwrap(), source); } + +#[rstest::rstest] +#[case::missing(None)] +#[case::wrong(Some(TemplateLookupScope::Cluster))] +#[case::matching(Some(TemplateLookupScope::LocalExecution))] +#[tokio::test] +async fn template_resolver_accepts_legacy_or_matching_scope_receipt( + #[case] acknowledgement: Option, +) { + use crate::template_resolver::fake::FakeTemplateResolver; + use pplx_space_template_mgr_client::proto::{GetTemplateResponse, TemplateRef}; + + let directory = tempfile::tempdir().unwrap(); + let template_dir = directory.path().to_str().unwrap().to_owned(); + let scope = TemplateLookupScope::LocalExecution; + let response_dir = template_dir.clone(); + let templates = FakeTemplateResolver::new(move || { + let mut response = tonic::Response::new(GetTemplateResponse { + template_dir: response_dir.clone(), + arch: "x86_64".into(), + network_policy_authority: pplx_space_net_common::proto::PolicyAuthority::Local as i32, + auto_lifecycle_authority: pplx_space_net_common::proto::PolicyAuthority::Local as i32, + ..Default::default() + }); + if let Some(acknowledgement) = acknowledgement { + acknowledgement.receipt(&mut response); + } + Ok(response) + }); + + let result = resolve_template_paths( + &templates, + TemplateRef { + name: "app-runtime".into(), + ..Default::default() + }, + scope, + ) + .await; + + if acknowledgement.is_none() || acknowledgement == Some(scope) { + assert_eq!(result.unwrap().template_dir, template_dir); + } else { + let error = result.err().expect("unacknowledged scope must be rejected"); + assert_eq!(error.code(), tonic::Code::FailedPrecondition); + assert!(error.message().contains("scope receipt mismatch")); + } + let lookups = templates.lookups(); + assert_eq!(lookups.len(), 1); + let mut request = tonic::Request::new(lookups[0].request.clone()); + *request.metadata_mut() = lookups[0].metadata.clone(); + assert_eq!( + TemplateLookupScope::from_request(&request).unwrap(), + Some(scope) + ); +} diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core.rs index 7530059c55..12d1aa05f7 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_core.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_core.rs @@ -223,6 +223,22 @@ impl SandboxManagerService { let template = &req.template; let runtime = &req.runtime; + use pplx_space_common_svc::template_lookup::TemplateLookupScope; + let mut scope = template_lookup_scope(&sandbox_owner, req.sandbox_id.as_deref())?; + if peer.principal() == UdsPeerPrincipal::Service(user_names::SPACE_NODEAPI) { + scope = match (scope, TemplateLookupScope::from_request(&request)?) { + (TemplateLookupScope::Sandbox { id, .. }, Some(TemplateLookupScope::Node(address))) => + TemplateLookupScope::Sandbox { id, caller_node: Some(address) }, + (scope @ TemplateLookupScope::Sandbox { .. }, Some(TemplateLookupScope::Cluster)) => scope, + (_, Some(scope @ (TemplateLookupScope::Node(_) | TemplateLookupScope::Cluster))) => scope, + (scope, None) => { + pplx_space_metrics::auth::template_lookup_compatibility("missing_scope"); + scope + } + _ => return Err(Status::permission_denied("invalid remote template caller scope")), + }; + } + // Resolve template to content-addressed paths via template-manager. // If caller pinned a content_hash, use it; otherwise resolve newest. let mut resolved_template = resolve_template_paths( @@ -232,6 +248,7 @@ impl SandboxManagerService { content_hash: req.content_hash.clone(), tag: None, }, + scope, ) .await?; if let Some(selected) = &verified { diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs index 198a999137..05cf79f0ea 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs @@ -123,7 +123,10 @@ impl TemplateBuild { fn resolver(&self, build: &str) -> Arc { let response = self.response(build); Arc::new(FakeTemplateResolver::new(move || { - Ok(tonic::Response::new(response.clone())) + let mut response = tonic::Response::new(response.clone()); + pplx_space_common_svc::template_lookup::TemplateLookupScope::LocalExecution + .receipt(&mut response); + Ok(response) })) } } diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core/create_template_lookup_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_template_lookup_tests.rs index 9536b47e6e..8b4d2a4a70 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_core/create_template_lookup_tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_template_lookup_tests.rs @@ -9,6 +9,7 @@ use crate::template_resolver::TemplateLookupError; use crate::template_resolver::fake::{FakeTemplateResolver, LookupChannel}; use pplx_space_common_core::host::linux::identities::users; use pplx_space_common_svc::PeerIdentity; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; use pplx_space_net_common::proto::PolicyAuthority; use pplx_space_template_mgr_client::proto::{GetTemplateRequest, TemplateRef}; @@ -29,8 +30,11 @@ fn unreachable_template_manager() -> FakeTemplateResolver { }) } +#[rstest::rstest] #[tokio::test] -async fn create_checks_auto_lifecycle_authority_before_template_lookup() { +async fn create_checks_auto_lifecycle_authority_before_template_lookup( + #[values(false, true)] scoped: bool, +) { if !crate::service::tests::enter_route_test_process(&[ ("EPPO_SDK_KEY", None), ("SPACE_FLAGD_ENABLED", None), @@ -153,6 +157,14 @@ async fn create_checks_auto_lifecycle_authority_before_template_lookup() { "invalid selection must not dial node-manager" ); } + let scope = if scoped && case.peer.uid() == users::SPACE_NODEAPI.uid { + TemplateLookupScope::Cluster + } else { + TemplateLookupScope::LocalExecution + }; + if scoped { + scope.attach(&mut request); + } request.extensions_mut().insert(case.peer); let error = service.create(request).await.expect_err(case.name); @@ -180,10 +192,16 @@ async fn create_checks_auto_lifecycle_authority_before_template_lookup() { Vec::new() }; assert_eq!(templates.requests(), expected_lookups, "{}", case.name); - // Create dials template-manager for its lookup and sends no metadata. for lookup in &lookups { assert_eq!(lookup.channel, LookupChannel::Dialed, "{}", case.name); - assert!(lookup.metadata.is_empty(), "{}", case.name); + let mut request = Request::new(lookup.request.clone()); + *request.metadata_mut() = lookup.metadata.clone(); + assert_eq!( + TemplateLookupScope::from_request(&request).unwrap(), + Some(scope), + "{}", + case.name + ); } assert!(node_manager.calls().is_empty(), "{}", case.name); } diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone.rs b/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone.rs index b7462e869a..673f1bdbf6 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone.rs @@ -264,6 +264,7 @@ impl SandboxManagerService { content_hash: sandbox.template_hash.clone(), tag: None, }, + template_lookup_scope(&sandbox.user, Some(&sandbox.id))?, ) .await?, ) @@ -694,8 +695,12 @@ impl SandboxManagerService { let uuid8 = SandboxRuntimeLabel::from_sandbox_id(new_sandbox_id); // Resolve template and compute hash - let resolved_template = - resolve_template_paths(self.template_resolver.as_ref(), template_ref).await?; + let resolved_template = resolve_template_paths( + self.template_resolver.as_ref(), + template_ref, + template_lookup_scope(&donor.user, Some(&donor.id))?, + ) + .await?; reject_incompatible_template_arch( template, resolved_template diff --git a/crates/libs/sbx/mgr/server/src/service/mod.rs b/crates/libs/sbx/mgr/server/src/service/mod.rs index d19adebaff..ac349dbf67 100644 --- a/crates/libs/sbx/mgr/server/src/service/mod.rs +++ b/crates/libs/sbx/mgr/server/src/service/mod.rs @@ -2042,6 +2042,7 @@ mod cleanup_support; mod cleanup_targets; mod create; mod create_support; +pub(crate) use create_support::template_lookup_scope; pub(crate) mod drain; mod eviction; pub(crate) mod snapshot_capture; diff --git a/crates/libs/sbx/mgr/server/src/service/network_policy.rs b/crates/libs/sbx/mgr/server/src/service/network_policy.rs index 33129c42f2..637d5e7ee3 100644 --- a/crates/libs/sbx/mgr/server/src/service/network_policy.rs +++ b/crates/libs/sbx/mgr/server/src/service/network_policy.rs @@ -133,22 +133,24 @@ pub(crate) async fn fetch_template_default_network_policy( templates: &dyn crate::template_resolver::TemplateResolver, template: &str, ) -> Result { - let response = match templates - .get_template_on_shared_channel(tonic::Request::new( - pplx_space_template_mgr_client::proto::GetTemplateRequest { - caller_org_id: None, - template: Some(pplx_space_template_mgr_client::proto::TemplateRef { - name: template.to_string(), - content_hash: None, - tag: None, - }), - vcpu: None, - memory_mib: None, - }, - )) - .await - { - Ok(response) => response.into_inner(), + let mut request = + tonic::Request::new(pplx_space_template_mgr_client::proto::GetTemplateRequest { + caller_org_id: None, + template: Some(pplx_space_template_mgr_client::proto::TemplateRef { + name: template.to_string(), + content_hash: None, + tag: None, + }), + vcpu: None, + memory_mib: None, + }); + let scope = pplx_space_common_svc::template_lookup::TemplateLookupScope::LocalExecution; + scope.attach(&mut request); + let response = match templates.get_template_on_shared_channel(request).await { + Ok(response) => { + scope.observe_response(&response)?; + response.into_inner() + } Err(error) => { return Err(Status::internal(format!( "template-manager GetTemplate({template}) failed: {error}" diff --git a/crates/libs/sbx/mgr/server/src/service/recovery_cross_node.rs b/crates/libs/sbx/mgr/server/src/service/recovery_cross_node.rs index 701b47635b..9e22ee8235 100644 --- a/crates/libs/sbx/mgr/server/src/service/recovery_cross_node.rs +++ b/crates/libs/sbx/mgr/server/src/service/recovery_cross_node.rs @@ -1918,8 +1918,13 @@ impl SandboxManagerService { .ok_or_else(|| { RestoreError::Other("restore execution identity is missing".into()) })?; + let owner = self + .lookup_sandbox(&canonical_id) + .map_err(|error| RestoreError::Other(error.to_string()))? + .user; let phase2_result = restore_pull_rootfs( &staged, + &owner, execution_fence, &self.node_manager, self.template_resolver.as_ref(), diff --git a/crates/libs/sbx/mgr/server/src/service/start.rs b/crates/libs/sbx/mgr/server/src/service/start.rs index f391f13b0a..a92b0688ab 100644 --- a/crates/libs/sbx/mgr/server/src/service/start.rs +++ b/crates/libs/sbx/mgr/server/src/service/start.rs @@ -546,6 +546,7 @@ pub(super) struct ResolvedSandboxMemoryPolicy { async fn prepare_suspend_restore_dir( service: &SandboxManagerService, + owner: &pplx_space_common_core::types::SandboxOwner, sandbox_id: String, storage_uri: Option, execution_generation: Option, @@ -628,6 +629,7 @@ async fn prepare_suspend_restore_dir( Some(storage_uri) => { crate::snapshot_restore::prepare_restore_dir_from_uri( &sandbox_id, + owner, execution_fence, storage_uri, transfer_scope_id, @@ -639,6 +641,7 @@ async fn prepare_suspend_restore_dir( None => { crate::snapshot_restore::prepare_restore_dir( &sandbox_id, + owner, execution_fence, transfer_scope_id, &service.node_manager, @@ -1497,6 +1500,7 @@ impl SandboxManagerService { |sandbox_id, storage_uri| { prepare_suspend_restore_dir( self, + &sandbox.user, sandbox_id, storage_uri, ctx.execution_generation, @@ -1625,6 +1629,7 @@ impl SandboxManagerService { content_hash: sandbox.template_hash.clone(), tag: None, }, + sandbox, ) .await?; reject_incompatible_template_spaced_compatibility( diff --git a/crates/libs/sbx/mgr/server/src/service/start/tests.rs b/crates/libs/sbx/mgr/server/src/service/start/tests.rs index 9a7671b965..5ec2172543 100644 --- a/crates/libs/sbx/mgr/server/src/service/start/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/start/tests.rs @@ -982,6 +982,7 @@ async fn suspend_restore_dir_keeps_a_previous_scope_it_cannot_cancel() { let error = prepare_suspend_restore_dir( &service, + &service.lookup_sandbox(&id).unwrap().user, id.clone(), Some("s3://snapshots/suspend/synthetic?region=us-east-1".to_owned()), Some(ExecutionGeneration::new(3).unwrap()), diff --git a/crates/libs/sbx/mgr/server/src/snapshot_restore.rs b/crates/libs/sbx/mgr/server/src/snapshot_restore.rs index e688d035e2..82117f1484 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_restore.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_restore.rs @@ -1346,6 +1346,7 @@ fn replacement_for_pull(local_path: &Path, s3_uri: &str, adopted: bool) -> PullR /// `Restoring` row and publishes that directory before the slow rootfs phase. pub async fn prepare_restore_dir( sandbox_id: &str, + owner: &pplx_space_common_core::types::SandboxOwner, execution_fence: pplx_space_node_mgr_client::execution_fence::ExecutionFence, transfer_scope_id: pplx_space_common_core::types::TransferScopeId, node_manager: &crate::node_manager_transport::NodeManagerConnector, @@ -1358,7 +1359,7 @@ pub async fn prepare_restore_dir( RestoreMetadataDestination::InPlace(transfer_scope_id), ) .await?; - restore_pull_rootfs(&staged, execution_fence, node_manager, templates).await?; + restore_pull_rootfs(&staged, owner, execution_fence, node_manager, templates).await?; Ok(RestorePrepared { snapshot_dir: staged.snapshot_dir, storage_uri: staged.storage_uri, @@ -1369,6 +1370,7 @@ pub async fn prepare_restore_dir( pub async fn prepare_restore_dir_from_uri( sandbox_id: &str, + owner: &pplx_space_common_core::types::SandboxOwner, execution_fence: pplx_space_node_mgr_client::execution_fence::ExecutionFence, storage_uri: &str, transfer_scope_id: pplx_space_common_core::types::TransferScopeId, @@ -1386,7 +1388,7 @@ pub async fn prepare_restore_dir_from_uri( RestoreMetadataDestination::InPlace(transfer_scope_id), ) .await?; - restore_pull_rootfs(&staged, execution_fence, node_manager, templates).await?; + restore_pull_rootfs(&staged, owner, execution_fence, node_manager, templates).await?; Ok(RestorePrepared { snapshot_dir: staged.snapshot_dir, storage_uri: staged.storage_uri, @@ -1944,6 +1946,7 @@ fn preserve_attributed_metadata_retry( /// completed, verified files for the next retry. pub async fn restore_pull_rootfs( stage: &RestoreMetadataStaged, + owner: &pplx_space_common_core::types::SandboxOwner, execution_fence: pplx_space_node_mgr_client::execution_fence::ExecutionFence, node_manager: &crate::node_manager_transport::NodeManagerConnector, templates: &dyn crate::template_resolver::TemplateResolver, @@ -1986,6 +1989,7 @@ pub async fn restore_pull_rootfs( RestoreCodec::BtrfsSendV1 => { stage_btrfs_send_restore( transfer_identity, + owner, &stage.config, &stage.snapshot_dir.join("snapshot.json"), &stage.snapshot_dir, @@ -2606,6 +2610,7 @@ async fn stage_encrypted_restore( )] async fn stage_btrfs_send_restore( identity: RestoreTransferIdentity<'_>, + owner: &pplx_space_common_core::types::SandboxOwner, config: &SnapshotUploadConfig, sidecar_path: &Path, snapshot_dir: &Path, @@ -2665,7 +2670,7 @@ async fn stage_btrfs_send_restore( } run_btrfs_staging( Box::pin(async { - ensure_local_template(templates, sandbox_id, template_name, content_hash) + ensure_local_template(templates, sandbox_id, owner, template_name, content_hash) .await .map(drop) }), @@ -2753,24 +2758,28 @@ async fn run_btrfs_staging<'a>( async fn ensure_local_template( templates: &dyn crate::template_resolver::TemplateResolver, sandbox_id: &str, + owner: &pplx_space_common_core::types::SandboxOwner, template_name: &str, content_hash: &str, ) -> Result { use crate::template_resolver::TemplateLookupError; async { + let mut request = + tonic::Request::new(pplx_space_template_mgr_client::proto::GetTemplateRequest { + caller_org_id: None, + template: Some(pplx_space_template_mgr_client::proto::TemplateRef { + name: template_name.to_string(), + content_hash: Some(content_hash.to_string()), + tag: None, + }), + vcpu: None, + memory_mib: None, + }); + let scope = crate::service::template_lookup_scope(owner, Some(sandbox_id)) + .map_err(|error| RestoreError::Other(error.to_string()))?; + scope.attach(&mut request); let resp = templates - .get_template(tonic::Request::new( - pplx_space_template_mgr_client::proto::GetTemplateRequest { - caller_org_id: None, - template: Some(pplx_space_template_mgr_client::proto::TemplateRef { - name: template_name.to_string(), - content_hash: Some(content_hash.to_string()), - tag: None, - }), - vcpu: None, - memory_mib: None, - }, - )) + .get_template(request) .await .map_err(|error| match error { TemplateLookupError::Connect(e) => RestoreError::Other(format!( @@ -2779,8 +2788,11 @@ async fn ensure_local_template( TemplateLookupError::Rpc(error) => { template_pull_error(template_name, content_hash, error) } - })? - .into_inner(); + })?; + scope + .observe_response(&resp) + .map_err(|error| template_pull_error(template_name, content_hash, error))?; + let resp = resp.into_inner(); if resp.template_dir.is_empty() { return Err(RestoreError::Other(format!( "tmplmgr returned no local template dir for {template_name}@{content_hash}" diff --git a/crates/libs/sbx/mgr/server/src/snapshot_restore/tests.rs b/crates/libs/sbx/mgr/server/src/snapshot_restore/tests.rs index b78a39e366..83752e8768 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_restore/tests.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_restore/tests.rs @@ -1754,6 +1754,58 @@ fn restored_sidecar_is_replaced_without_writing_the_pulled_file() { assert!(!tmp.path().join(".snapshot.json.restore").exists()); } +#[rstest::rstest] +#[case::root("root@local", true)] +#[case::operator("deploy@local", true)] +#[case::tenant("user:00000000-0000-0000-0000-000000000001", false)] +#[tokio::test] +async fn restore_template_lookup_uses_validated_owner( + #[case] owner: &str, + #[case] local: bool, + #[values(false, true)] receipt: bool, +) { + use pplx_space_common_svc::template_lookup::TemplateLookupScope; + use pplx_space_template_mgr_client::proto::GetTemplateResponse; + let id = pplx_space_common_core::types::SandboxId::new(); + let expected = if local { + TemplateLookupScope::LocalExecution + } else { + TemplateLookupScope::Sandbox { + id, + caller_node: None, + } + }; + let templates = crate::template_resolver::fake::FakeTemplateResolver::new(move || { + let mut response = tonic::Response::new(GetTemplateResponse { + template_dir: "/synthetic/cached-base".into(), + ..Default::default() + }); + if receipt { + expected.receipt(&mut response); + } + Ok(response) + }); + + let path = ensure_local_template( + &templates, + &id.to_string(), + &owner.parse().unwrap(), + "base", + "sha256:cached", + ) + .await + .unwrap(); + + assert_eq!(path, "/synthetic/cached-base"); + let lookup = templates.lookups().pop().unwrap(); + let mut request = tonic::Request::new(lookup.request); + *request.metadata_mut() = lookup.metadata; + assert_eq!( + TemplateLookupScope::from_request(&request).unwrap(), + Some(expected) + ); +} + // A stopped materialization refused because the staged artifact isn't the // evacuated disk the caller named is evidence disagreeing with itself, the // same class as the other identity mismatches, not a filesystem or database diff --git a/crates/libs/template/mgr/src/service/lookup.rs b/crates/libs/template/mgr/src/service/lookup.rs index 07ee225411..d9e451cb3b 100644 --- a/crates/libs/template/mgr/src/service/lookup.rs +++ b/crates/libs/template/mgr/src/service/lookup.rs @@ -1,4 +1,5 @@ use pplx_space_common_core::types::OrgId; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; use super::*; @@ -17,16 +18,27 @@ impl TemplateManagerService { &self, request: Request, ) -> Result, Status> { - self.get_template_resolving( - request, - crate::volume::find_release_inner, - async |spec, resolve_elapsed, release| { - self.volume_puller - .materialize_release(Arc::clone(&self.db), spec, resolve_elapsed, release) - .await - }, - ) - .await + let scope = TemplateLookupScope::from_request(&request)?; + if scope.is_none() { + pplx_space_metrics::auth::template_lookup_compatibility("missing_scope"); + } + let mut response = self + .get_template_resolving( + request, + async |name, hash, tag, org| { + crate::volume::find_release_in_scope(name, hash, tag, org, scope).await + }, + async |spec, resolve_elapsed, release| { + self.volume_puller + .materialize_release(Arc::clone(&self.db), spec, resolve_elapsed, release) + .await + }, + ) + .await?; + if let Some(scope) = scope { + scope.receipt(&mut response); + } + Ok(response) } pub(super) async fn get_template_resolving( diff --git a/crates/libs/template/mgr/src/service/tests.rs b/crates/libs/template/mgr/src/service/tests.rs index 893769b3fb..891f2033cd 100644 --- a/crates/libs/template/mgr/src/service/tests.rs +++ b/crates/libs/template/mgr/src/service/tests.rs @@ -8,6 +8,7 @@ use std::{path::Path, sync::Arc}; use tracing_subscriber::layer::SubscriberExt as _; mod export; +mod lookup_rpc; #[tokio::test(flavor = "current_thread")] async fn background_loop_observer_contains_inner_panic() { diff --git a/crates/libs/template/mgr/src/service/tests/lookup_rpc.rs b/crates/libs/template/mgr/src/service/tests/lookup_rpc.rs new file mode 100644 index 0000000000..ba14fc931d --- /dev/null +++ b/crates/libs/template/mgr/src/service/tests/lookup_rpc.rs @@ -0,0 +1,100 @@ +use super::*; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; +use pplx_space_template_mgr_client::proto::{ + GetTemplateRequest, TemplateRef, template_manager_client::TemplateManagerClient, + template_manager_server::TemplateManagerServer, +}; +use pplx_space_test_env::TestResult; +use tokio_stream::wrappers::UnixListenerStream; + +struct LookupServer(tokio::task::JoinHandle>); + +impl Drop for LookupServer { + fn drop(&mut self) { + self.0.abort(); + } +} + +#[rstest::rstest] +#[case::legacy(None)] +#[case::local(Some(TemplateLookupScope::LocalExecution))] +#[case::forwarded(Some(TemplateLookupScope::Node("127.0.0.2".parse().unwrap())))] +#[case::sandbox(Some(TemplateLookupScope::Sandbox { + id: pplx_space_common_core::types::SandboxId::new(), caller_node: None +}))] +#[tokio::test] +async fn cached_lookup_preserves_offline_behavior_during_rollout( + #[case] scope: Option, +) -> TestResult { + if !pplx_space_test_env::enter_test_process(&[ + ("SPACE_INTERNAL_API_ENDPOINT", None), + ("SPACE_CLUSTER_NODE_API_ENDPOINT", None), + ]) { + return Ok(()); + } + let CachedBaseFixture { + root, + mut service, + hash_a, + .. + } = cached_base_fixture(); + service.peer_authorizer = UdsPeerAuthorizer::new( + "lookup-rpc-test", + [UdsPeerAccount::with_fallback_uid( + SPACE_SBXMGR_ACCOUNT, + nix::unistd::geteuid().as_raw(), + )], + ); + let socket_dir = tempfile::tempdir()?; + let socket = socket_dir.path().join("manager.sock"); + let listener = tokio::net::UnixListener::bind(&socket)?; + let _server = LookupServer(tokio::spawn( + tonic::transport::Server::builder() + .add_service(TemplateManagerServer::new(service)) + .serve_with_incoming(UnixListenerStream::new(listener)), + )); + let channel = pplx_space_common_grpc::connect_uds_channel("lookup-test", &socket).await?; + let mut client = TemplateManagerClient::new(channel); + let mut request = Request::new(GetTemplateRequest { + template: Some(TemplateRef { + name: "base".into(), + content_hash: Some(hash_a.clone()), + tag: None, + }), + ..Default::default() + }); + if let Some(scope) = scope { + scope.attach(&mut request); + } + + let response = client.get_template(request).await?; + + assert_eq!( + response.get_ref().template.as_ref().unwrap().content_hash, + Some(hash_a.clone()) + ); + assert_eq!( + response.get_ref().template_dir, + root.path().join("base").join(&hash_a).to_string_lossy() + ); + if let Some(scope) = scope { + scope.observe_response(&response)?; + assert!( + response + .metadata() + .contains_key("x-space-template-lookup-scope-received") + ); + } else { + assert!( + !response + .metadata() + .contains_key("x-space-template-lookup-scope-received") + ); + } + assert!( + !response + .metadata() + .contains_key("x-space-template-lookup-scope") + ); + Ok(()) +} diff --git a/crates/libs/template/mgr/src/volume.rs b/crates/libs/template/mgr/src/volume.rs index 8cbd253855..f7f04de09c 100644 --- a/crates/libs/template/mgr/src/volume.rs +++ b/crates/libs/template/mgr/src/volume.rs @@ -59,6 +59,7 @@ //! addressing. use crate::rootfs_backing::RootfsBackingPolicy; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, MutexGuard}; @@ -951,7 +952,15 @@ async fn find_release( ) -> Result { let started = Instant::now(); let generic_only = trigger.is_prefetch(); - let result = find_release_with_requirement(name, req_hash, req_tag, None, generic_only).await; + let result = find_release_with_requirement( + name, + req_hash, + req_tag, + None, + generic_only, + Some(TemplateLookupScope::Cluster), + ) + .await; let result_label = match &result { Ok(_) => "ok", Err(FindReleaseError::NotFound) => "not_found", @@ -1003,7 +1012,24 @@ pub(crate) async fn find_release_inner( req_tag: &str, caller_org: Option, ) -> Result { - find_release_with_requirement(name, req_hash, req_tag, caller_org, false).await + find_release_in_scope( + name, + req_hash, + req_tag, + caller_org, + Some(TemplateLookupScope::Cluster), + ) + .await +} + +pub(crate) async fn find_release_in_scope( + name: &str, + req_hash: &str, + req_tag: &str, + caller_org: Option, + scope: Option, +) -> Result { + find_release_with_requirement(name, req_hash, req_tag, caller_org, false, scope).await } async fn find_release_with_requirement( @@ -1012,6 +1038,7 @@ async fn find_release_with_requirement( req_tag: &str, caller_org: Option, generic_only: bool, + scope: Option, ) -> Result { let mut client = connect_cluster_api_from_env().await.map_err(|error| { warn!(error = %error, "cluster api connect failed"); @@ -1027,16 +1054,17 @@ async fn find_release_with_requirement( // but not for the hash path (digests are runtime-scoped). let resp = if !req_hash.is_empty() { let hash = req_hash.to_string(); - match client - .find_template_release_by_hash(FindTemplateReleaseByHashRequest { - generic_only, - caller_org_id: caller_org.map(|org| org.to_string()), - template_name: name.to_string(), - hash: hash.clone(), - }) - .await - { - Ok(r) => r.into_inner(), + let mut request = tonic::Request::new(FindTemplateReleaseByHashRequest { + generic_only, + caller_org_id: caller_org.map(|org| org.to_string()), + template_name: name.to_string(), + hash: hash.clone(), + }); + if let Some(scope) = scope { + scope.attach(&mut request); + } + match client.find_template_release_by_hash(request).await { + Ok(r) => r, Err(status) => { if status.code() == tonic::Code::NotFound { debug!( @@ -1057,17 +1085,18 @@ async fn find_release_with_requirement( } } else { let tag = lookup_tag(req_tag); - match client - .find_template_release(FindTemplateReleaseRequest { - generic_only, - caller_org_id: caller_org.map(|org| org.to_string()), - template_name: name.to_string(), - runtime: DEFAULT_RUNTIME.to_string(), - tag: tag.clone(), - }) - .await - { - Ok(r) => r.into_inner(), + let mut request = tonic::Request::new(FindTemplateReleaseRequest { + generic_only, + caller_org_id: caller_org.map(|org| org.to_string()), + template_name: name.to_string(), + runtime: DEFAULT_RUNTIME.to_string(), + tag: tag.clone(), + }); + if let Some(scope) = scope { + scope.attach(&mut request); + } + match client.find_template_release(request).await { + Ok(r) => r, Err(status) => { if status.code() == tonic::Code::NotFound { debug!(template = name, %tag, "FindTemplateRelease returned NotFound"); @@ -1079,6 +1108,12 @@ async fn find_release_with_requirement( } }; + if let Some(scope) = scope { + scope + .observe_response(&resp) + .map_err(FindReleaseError::Status)?; + } + let resp = resp.into_inner(); parse_release_response(name, req_hash, caller_org, generic_only, resp) } diff --git a/crates/libs/template/mgr/src/volume/tests/catalog.rs b/crates/libs/template/mgr/src/volume/tests/catalog.rs index b69f032374..e3142eac41 100644 --- a/crates/libs/template/mgr/src/volume/tests/catalog.rs +++ b/crates/libs/template/mgr/src/volume/tests/catalog.rs @@ -165,3 +165,71 @@ async fn assert_pull_outcome(visibility: TemplateVisibility, trigger: PullTrigge ); Ok(()) } + +#[tokio::test] +async fn catalog_receipts_preserve_legacy_org_checks() -> TestResult { + use pplx_space_common_svc::template_lookup::TemplateLookupScope; + use pplx_space_node_api_client::proto::{ + FindTemplateReleaseByHashRequest, FindTemplateReleaseRequest, + cluster_api_client::ClusterApiClient, + }; + let catalog = Catalog::start(TemplateVisibility::OrgSpecific).await?; + let mut client = ClusterApiClient::connect(catalog.endpoint.clone()).await?; + let digest = format!("sha256:{}", "a".repeat(64)); + let owner = "00000000-0000-0000-0000-000000000003"; + let foreign = "00000000-0000-0000-0000-000000000004"; + for scope in [None, Some(TemplateLookupScope::Cluster)] { + for org in [owner, foreign] { + for by_hash in [false, true] { + let metadata = |request: &mut tonic::metadata::MetadataMap| { + request.insert("authorization", format!("Bearer {TOKEN}").parse().unwrap()); + }; + let result = if by_hash { + let mut request = tonic::Request::new(FindTemplateReleaseByHashRequest { + template_name: TEMPLATE.into(), + hash: digest.clone(), + caller_org_id: Some(org.into()), + generic_only: false, + }); + metadata(request.metadata_mut()); + if let Some(scope) = scope { + scope.attach(&mut request); + } + client.find_template_release_by_hash(request).await + } else { + let mut request = tonic::Request::new(FindTemplateReleaseRequest { + template_name: TEMPLATE.into(), + runtime: "fc".into(), + tag: "latest".into(), + caller_org_id: Some(org.into()), + generic_only: false, + }); + metadata(request.metadata_mut()); + if let Some(scope) = scope { + scope.attach(&mut request); + } + client.find_template_release(request).await + }; + if org == foreign { + assert_eq!(result.unwrap_err().code(), tonic::Code::NotFound); + } else { + let response = result?; + assert_eq!(response.get_ref().release.as_ref().unwrap().hash, digest); + assert_eq!(response.get_ref().authorized_org_id.as_deref(), Some(owner)); + assert_eq!( + response + .metadata() + .contains_key("x-space-template-lookup-scope-received"), + scope.is_some() + ); + assert!( + !response + .metadata() + .contains_key("x-space-template-lookup-scope") + ); + } + } + } + } + Ok(()) +} diff --git a/docs/architecture.md b/docs/architecture.md index 6bf2d7a2d6..ac6b450ae9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -24,6 +24,26 @@ empty policy preserves local S3-compatible storage. This policy does not audit existing releases or replace publisher authorization; provision the allowlist before promoting an enforcement image. +## Template lookup rollout + +Template lookup clients attach scope metadata and compatible servers return a +receipt. A missing receipt preserves the existing lookup behavior during a +rolling upgrade. A mismatched receipt fails the request. Authentication, +organization checks, and cache decisions continue to use the existing rules. + +The `auth.template_lookup.compatibility` counter records `missing_scope`, +`missing_receipt`, and `received`, under the emitting service's identity. A +receipt confirms only that one peer received the metadata. It does not certify +that a downstream peer received it or that the scope was authorized. Check +all participating services and exercise cache misses as well as hits; an idle +service or a cached lookup cannot prove the catalog supports the protocol. + +Deploy this compatibility phase before introducing stricter lookup rules. +Strict enforcement requires a separate authorization acknowledgement; it must +not accept the compatibility receipt as proof. Existing clients that omit scope +must be migrated before enforcing it. Rollback to the previous version remains +compatible with these requests and replies. + ## Request and execution flow The API records desired state in PostgreSQL. Controllers and worker services reconcile the running system with that state, so an accepted request is not a readiness signal. From 029b65846209fb74d682cd6150276e5619b3c537 Mon Sep 17 00:00:00 2001 From: Nick Dujay Date: Wed, 21 Oct 2026 16:59:09 -0400 Subject: [PATCH 2/2] Carry lookup metadata through cold catalog resolution Complete the compatibility protocol from public PR #736 for cold requests. Keep on-demand lookup metadata intact and retain cluster scope for background pulls. Add request-level regression coverage for digest and name-only lookups. --- .../libs/template/mgr/src/service/lookup.rs | 4 +- .../mgr/src/service/tests/lookup_rpc.rs | 87 +++++++++++++++++++ crates/libs/template/mgr/src/volume.rs | 81 ++++++++++------- .../template/mgr/src/volume/tests/catalog.rs | 20 ++++- 4 files changed, 158 insertions(+), 34 deletions(-) diff --git a/crates/libs/template/mgr/src/service/lookup.rs b/crates/libs/template/mgr/src/service/lookup.rs index d9e451cb3b..600f0603f6 100644 --- a/crates/libs/template/mgr/src/service/lookup.rs +++ b/crates/libs/template/mgr/src/service/lookup.rs @@ -60,6 +60,7 @@ impl TemplateManagerService { ) -> crate::volume::PullOutcome, ) -> Result, Status> { self.authorize_peer(&request, "GetTemplate", ALLOW_TEMPLATE_OBSERVER)?; + let scope = TemplateLookupScope::from_request(&request)?; let req = request.get_ref(); let tmpl_ref = req .template @@ -235,12 +236,13 @@ impl TemplateManagerService { } None => { self.volume_puller - .pull( + .pull_in_scope( Arc::clone(&self.db), template_id, &req_hash, &req_tag, crate::volume::PullTrigger::OnDemand, + scope, ) .await } diff --git a/crates/libs/template/mgr/src/service/tests/lookup_rpc.rs b/crates/libs/template/mgr/src/service/tests/lookup_rpc.rs index ba14fc931d..4b853ca98a 100644 --- a/crates/libs/template/mgr/src/service/tests/lookup_rpc.rs +++ b/crates/libs/template/mgr/src/service/tests/lookup_rpc.rs @@ -98,3 +98,90 @@ async fn cached_lookup_preserves_offline_behavior_during_rollout( ); Ok(()) } + +#[rstest::rstest] +#[case::legacy(None)] +#[case::local(Some("local"))] +#[case::cluster(Some("cluster"))] +#[case::node(Some("node:127.0.0.2"))] +#[case::sandbox(Some("sandbox:00000000-0000-0000-0000-000000000005"))] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cold_lookup_preserves_scope_at_catalog( + #[case] scope: Option<&str>, + #[values(false, true)] by_hash: bool, +) -> TestResult { + use crate::volume::tests::catalog::{Catalog, TEMPLATE}; + use pplx_space_node_api_client::proto::TemplateVisibility; + use pplx_space_test_env::enter_test_process; + use std::ffi::OsStr; + + const CHILD: &str = "SPACE_TEST_COLD_LOOKUP_SCOPE"; + if std::env::var_os(CHILD).is_none() { + let catalog = Catalog::start(TemplateVisibility::Generic).await?; + assert!(!enter_test_process(&[ + (CHILD, Some(OsStr::new("1"))), + ( + "SPACE_INTERNAL_API_ENDPOINT", + Some(OsStr::new(&catalog.endpoint)) + ), + ( + "SPACE_CLUSTER_API_TOKEN_PATH", + Some(catalog.token_file.path().as_os_str()) + ), + ( + "SPACE_CLUSTER_API_ENDPOINT_DISCOVERY", + Some(OsStr::new("off")) + ), + ("SPACE_INTERNAL_RPC_INSECURE_DEV", Some(OsStr::new("true"))), + ])); + assert_eq!( + *catalog.lookup_scopes.lock().unwrap(), + vec![scope.map(str::parse::).transpose()?], + ); + return Ok(()); + } + assert!(enter_test_process(&[])); + let root = tempfile::tempdir()?; + let mut service = + TemplateManagerService::new_with_template_root(crate::db::test_conn(), root.path())?; + service.peer_authorizer = UdsPeerAuthorizer::new( + "cold-lookup-test", + [UdsPeerAccount::with_fallback_uid( + SPACE_SBXMGR_ACCOUNT, + nix::unistd::geteuid().as_raw(), + )], + ); + let socket = root.path().join("manager.sock"); + let listener = tokio::net::UnixListener::bind(&socket)?; + let _server = LookupServer(tokio::spawn( + tonic::transport::Server::builder() + .add_service(TemplateManagerServer::new(service)) + .serve_with_incoming(UnixListenerStream::new(listener)), + )); + let channel = pplx_space_common_grpc::connect_uds_channel("cold-lookup-test", &socket).await?; + let mut client = TemplateManagerClient::new(channel); + let mut request = Request::new(GetTemplateRequest { + template: Some(TemplateRef { + name: TEMPLATE.into(), + content_hash: by_hash.then(|| format!("sha256:{}", "a".repeat(64))), + tag: None, + }), + vcpu: Some(1), + memory_mib: Some(512), + ..Default::default() + }); + if let Some(scope) = scope { + scope.parse::()?.attach(&mut request); + } + + let status = client.get_template(request).await.unwrap_err(); + + // The catalog has a release but no locations, so a cold pull reaches + // materialization without a volume-manager or writes to the host cache. + assert_eq!(status.code(), tonic::Code::FailedPrecondition); + assert!( + status.message().contains("no remote storage locations"), + "{status}" + ); + Ok(()) +} diff --git a/crates/libs/template/mgr/src/volume.rs b/crates/libs/template/mgr/src/volume.rs index f7f04de09c..850a2ba23d 100644 --- a/crates/libs/template/mgr/src/volume.rs +++ b/crates/libs/template/mgr/src/volume.rs @@ -407,6 +407,25 @@ impl VolumePuller { /// This keeps caller cancellation from multiplying rootfs /// decompressions while preserving the recover-from-disk behavior for /// process death. + pub async fn pull( + &self, + db: Arc>, + name: &str, + req_hash: &str, + req_tag: &str, + trigger: PullTrigger, + ) -> PullOutcome { + self.pull_in_scope( + db, + name, + req_hash, + req_tag, + trigger, + Some(TemplateLookupScope::Cluster), + ) + .await + } + #[tracing::instrument( skip_all, fields( @@ -417,13 +436,14 @@ impl VolumePuller { content_hash = tracing::field::Empty, ) )] - pub async fn pull( + pub(crate) async fn pull_in_scope( &self, db: Arc>, name: &str, req_hash: &str, req_tag: &str, trigger: PullTrigger, + scope: Option, ) -> PullOutcome { let started = Instant::now(); let base_tags = pull_metric_tags(name, req_hash, req_tag, trigger); @@ -435,7 +455,7 @@ impl VolumePuller { req_tag, trigger, }; - let outcome = pull_inner(self, db, spec, &base_tags, &mut path_taken).await; + let outcome = pull_inner(self, db, spec, &base_tags, &mut path_taken, scope).await; pplx_space_metrics::template::pull_duration( &base_tags, @@ -775,28 +795,28 @@ async fn pull_inner( spec: PullSpec<'_>, base_tags: &[String], path_taken: &mut PullPath, + scope: Option, ) -> PullOutcome { let name = spec.name; // 1. Resolve (release.hash, storage_uris) via ClusterApi. - let release = - match find_release(name, spec.req_hash, spec.req_tag, base_tags, spec.trigger).await { - Ok(found) => found, - Err(FindReleaseError::NotFound) => { - return PullOutcome::ClassifiedFailure(PullFailure::ClusterMetadataMissing); - } - Err(FindReleaseError::Unavailable) => { - return PullOutcome::ClassifiedFailure(PullFailure::ClusterMetadataUnavailable); - } - Err(FindReleaseError::InvalidResponse) => { - return PullOutcome::ClassifiedFailure(PullFailure::ClusterMetadataInvalid); - } - Err(FindReleaseError::Status(status)) => { - return PullOutcome::DownstreamFailure { - failure: PullFailure::ClusterMetadataUnavailable, - status, - }; - } - }; + let release = match find_release(spec, base_tags, scope).await { + Ok(found) => found, + Err(FindReleaseError::NotFound) => { + return PullOutcome::ClassifiedFailure(PullFailure::ClusterMetadataMissing); + } + Err(FindReleaseError::Unavailable) => { + return PullOutcome::ClassifiedFailure(PullFailure::ClusterMetadataUnavailable); + } + Err(FindReleaseError::InvalidResponse) => { + return PullOutcome::ClassifiedFailure(PullFailure::ClusterMetadataInvalid); + } + Err(FindReleaseError::Status(status)) => { + return PullOutcome::DownstreamFailure { + failure: PullFailure::ClusterMetadataUnavailable, + status, + }; + } + }; // The catalog has enforced generic-only selection in the same SQL snapshot // as the release and locations. Cached visibility is only an additional @@ -944,23 +964,20 @@ async fn pull_resolved_owner( /// Connects to the cluster-role node-api and asks /// `FindTemplateRelease(name, runtime, tag)`. async fn find_release( - name: &str, - req_hash: &str, - req_tag: &str, + spec: PullSpec<'_>, base_tags: &[String], - trigger: PullTrigger, + scope: Option, ) -> Result { let started = Instant::now(); - let generic_only = trigger.is_prefetch(); - let result = find_release_with_requirement( + let PullSpec { name, req_hash, req_tag, - None, - generic_only, - Some(TemplateLookupScope::Cluster), - ) - .await; + trigger, + } = spec; + let generic_only = trigger.is_prefetch(); + let result = + find_release_with_requirement(name, req_hash, req_tag, None, generic_only, scope).await; let result_label = match &result { Ok(_) => "ok", Err(FindReleaseError::NotFound) => "not_found", diff --git a/crates/libs/template/mgr/src/volume/tests/catalog.rs b/crates/libs/template/mgr/src/volume/tests/catalog.rs index e3142eac41..b3f1fb2974 100644 --- a/crates/libs/template/mgr/src/volume/tests/catalog.rs +++ b/crates/libs/template/mgr/src/volume/tests/catalog.rs @@ -1,6 +1,7 @@ use crate::volume::{PullFailure, PullOutcome, PullTrigger, VolumePuller}; use pplx_space_common_core::types::OrgId; use pplx_space_common_svc::auth::{NodeApiAuthConfig, NodeApiAuthLayer, NodeRole}; +use pplx_space_common_svc::template_lookup::TemplateLookupScope; use pplx_space_internal_api_server::ClusterApiService; use pplx_space_node_api_client::proto::{ RegisterTemplateReleaseRequest, RegisterTemplateRequest, TemplateVisibility, @@ -18,6 +19,7 @@ const CHILD_ENDPOINT: &str = "SPACE_TEST_PREFETCH_CATALOG_ENDPOINT"; pub(crate) struct Catalog { pub(crate) endpoint: String, pub(crate) token_file: tempfile::NamedTempFile, + pub(crate) lookup_scopes: Arc>>>, server: tokio::task::JoinHandle<()>, _database: pplx_space_db_test_fixtures::TestDb, } @@ -77,10 +79,21 @@ impl Catalog { None, )?; let auth = NodeApiAuthLayer::from_config(config).await?; + let lookup_scopes = Arc::new(Mutex::new(Vec::new())); + let observed_scopes = Arc::clone(&lookup_scopes); let server = tokio::spawn(async move { tonic::transport::Server::builder() .layer(auth) - .add_service(ClusterApiServer::new(service)) + .add_service(ClusterApiServer::with_interceptor( + service, + move |request: tonic::Request<()>| { + observed_scopes + .lock() + .unwrap() + .push(TemplateLookupScope::from_request(&request)?); + Ok(request) + }, + )) .serve_with_incoming(TcpListenerStream::new(listener)) .await .unwrap(); @@ -88,6 +101,7 @@ impl Catalog { Ok(Self { endpoint, token_file, + lookup_scopes, server, _database: database, }) @@ -138,6 +152,10 @@ async fn final_catalog_resolution_controls_prefetch( retired_setting.map(OsStr::new) ), ])); + assert_eq!( + *catalog.lookup_scopes.lock().unwrap(), + vec![Some(TemplateLookupScope::Cluster); 2] + ); Ok(()) }