diff --git a/Cargo.lock b/Cargo.lock index 1f67c6cfc0..80210ab877 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5933,6 +5933,7 @@ dependencies = [ name = "pplx-space-internal-api-server" version = "0.8.0" dependencies = [ + "axum", "futures-util", "hex", "insta", diff --git a/crates/libs/common/svc/src/auth/k8s.rs b/crates/libs/common/svc/src/auth/k8s.rs index f3bad50c45..db2642fef0 100644 --- a/crates/libs/common/svc/src/auth/k8s.rs +++ b/crates/libs/common/svc/src/auth/k8s.rs @@ -260,8 +260,8 @@ impl K8sTokenReviewer { } } }; - let username = match self.review_token(token, policy.audience).await { - Ok(username) => username, + let reviewed = match self.review_token(token, policy.audience).await { + Ok(reviewed) => reviewed, Err(TokenReviewFailure::Rejected { reason }) => { self.cache_result( key.clone(), @@ -299,13 +299,16 @@ impl K8sTokenReviewer { self.cache_result( key.clone(), CachedTokenReviewResult::Accepted { - username: username.clone(), + username: reviewed.username.clone(), + pod: reviewed.pod.clone(), }, expires_at, ); } - self.authorize_username(&policy, surface, method, &username, true)?; - Ok(policy.authenticated_caller(&username)) + self.authorize_username(&policy, surface, method, &reviewed.username, true)?; + let mut caller = policy.authenticated_caller(&reviewed.username); + caller.kubernetes_pod = reviewed.pod; + Ok(caller) } fn cached_result(&self, key: &TokenCacheKey) -> Option { @@ -330,10 +333,14 @@ impl K8sTokenReviewer { cached: CachedTokenReviewResult, ) -> Result { match cached { - CachedTokenReviewResult::Accepted { username } => { + CachedTokenReviewResult::Accepted { username, pod } => { pplx_space_metrics::auth::token_review_cache("hit_authenticated"); self.authorize_username(policy, surface, method, &username, false)?; - Ok(policy.authenticated_caller(&username)) + { + let mut caller = policy.authenticated_caller(&username); + caller.kubernetes_pod = pod; + Ok(caller) + } } CachedTokenReviewResult::Rejected { reason } => { pplx_space_metrics::auth::token_review_cache("hit_deny"); @@ -445,7 +452,7 @@ impl K8sTokenReviewer { &self, token: &str, audience: &str, - ) -> Result { + ) -> Result { let review = TokenReview { metadata: Default::default(), spec: TokenReviewSpec { @@ -860,8 +867,13 @@ struct CachedTokenReview { #[derive(Clone, Debug, Eq, PartialEq)] enum CachedTokenReviewResult { - Accepted { username: String }, - Rejected { reason: TokenReviewRejectReason }, + Accepted { + username: String, + pod: Option, + }, + Rejected { + reason: TokenReviewRejectReason, + }, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -1000,7 +1012,7 @@ fn jwt_exp_cache_deadline(token: &str, now: Instant, now_system: SystemTime) -> fn validate_token_review_status( status: Option, audience: &str, -) -> Result { +) -> Result { let status = status.ok_or(TokenReviewFailure::Rejected { reason: TokenReviewRejectReason::MissingStatus, })?; @@ -1027,12 +1039,49 @@ fn validate_token_review_status( reason: TokenReviewRejectReason::AudienceMismatch, }); } - status - .user - .and_then(|user| user.username) - .ok_or(TokenReviewFailure::Rejected { - reason: TokenReviewRejectReason::MissingUser, - }) + let user = status.user.ok_or(TokenReviewFailure::Rejected { + reason: TokenReviewRejectReason::MissingUser, + })?; + let username = user.username.ok_or(TokenReviewFailure::Rejected { + reason: TokenReviewRejectReason::MissingUser, + })?; + let pod = reviewed_pod_identity(&username, user.extra.as_ref()); + Ok(ReviewedToken { username, pod }) +} + +#[derive(Debug)] +struct ReviewedToken { + username: String, + pod: Option, +} + +fn reviewed_pod_identity( + username: &str, + extra: Option<&std::collections::BTreeMap>>, +) -> Option { + let mut account = username.strip_prefix("system:serviceaccount:")?.split(':'); + let namespace = account.next()?.to_owned(); + let service_account = account.next()?.to_owned(); + if namespace.is_empty() || service_account.is_empty() || account.next().is_some() { + return None; + } + let value = |key: &str| super::node_identity::single_extra_value(extra, key); + Some(super::KubernetesPodIdentity { + namespace, + service_account, + pod_name: value("authentication.kubernetes.io/pod-name")? + .parse() + .ok()?, + pod_uid: value("authentication.kubernetes.io/pod-uid")? + .parse() + .ok()?, + node_name: value("authentication.kubernetes.io/node-name")? + .parse() + .ok()?, + node_uid: value("authentication.kubernetes.io/node-uid")? + .parse() + .ok()?, + }) } fn token_digest(token: &str) -> [u8; 32] { diff --git a/crates/libs/common/svc/src/auth/k8s/tests.rs b/crates/libs/common/svc/src/auth/k8s/tests.rs index 661594fbd5..0e803d2fb0 100644 --- a/crates/libs/common/svc/src/auth/k8s/tests.rs +++ b/crates/libs/common/svc/src/auth/k8s/tests.rs @@ -42,6 +42,53 @@ fn test_k8s_config() -> K8sTokenAuthConfig { } } +#[tokio::test] +async fn token_review_node_binding_survives_cache_and_never_appears_for_incomplete_tokens() { + let fields = [ + ("authentication.kubernetes.io/pod-name", "worker-daemon"), + ("authentication.kubernetes.io/pod-uid", "pod-a"), + ("authentication.kubernetes.io/node-name", "worker-a"), + ("authentication.kubernetes.io/node-uid", "node-a"), + ]; + for missing in 0..=fields.len() { + let mut status = accepted_token_review_status("space-node-api"); + status.user.as_mut().unwrap().extra = Some( + fields + .iter() + .enumerate() + .filter(|(index, _)| *index != missing) + .map(|(_, (key, value))| (key.to_string(), vec![value.to_string()])) + .collect(), + ); + let (reviewer, calls) = reviewer_returning_status(test_k8s_config(), status); + for _ in 0..2 { + let caller = reviewer + .authenticate( + NodeRole::Worker, + AuthSurface::Normal, + "/space.node.api.v1.WorkerApi/GetSandboxState", + &projected_token(), + ) + .await + .unwrap(); + if missing == fields.len() { + let identity = caller + .kubernetes_pod() + .expect("complete TokenReview binding"); + assert_eq!(identity.namespace(), "space"); + assert_eq!(identity.service_account(), "space-api"); + assert_eq!(identity.pod_name().as_str(), "worker-daemon"); + assert_eq!(identity.pod_uid().as_str(), "pod-a"); + assert_eq!(identity.node_name().as_str(), "worker-a"); + assert_eq!(identity.node_uid().as_str(), "node-a"); + } else { + assert!(caller.kubernetes_pod().is_none()); + } + } + assert_eq!(calls.load(Ordering::SeqCst), 1); + } +} + #[tokio::test] async fn data_plane_identity_cannot_reuse_guest_authority_for_control_or_transfer() { for username in [ @@ -174,6 +221,7 @@ fn cache_key(token: &str, audience: &str) -> TokenCacheKey { fn accepted_cache_entry(username: &str, expires_at: Instant) -> CachedTokenReview { CachedTokenReview { result: CachedTokenReviewResult::Accepted { + pod: None, username: username.to_string(), }, expires_at, @@ -674,6 +722,7 @@ fn cache_token_review_result_evicts_earliest_expiring_entry_at_capacity() { &mut cache, inserted.clone(), CachedTokenReviewResult::Accepted { + pod: None, username: "third-sa".to_string(), }, now + Duration::from_secs(20), @@ -1092,6 +1141,7 @@ async fn concurrent_cache_inserts_at_capacity_do_not_panic() { reviewer.cache_result( cache_key(&format!("token-{index}"), "space-node-api"), CachedTokenReviewResult::Accepted { + pod: None, username: format!("system:serviceaccount:space:space-api-{index}"), }, Instant::now() + Duration::from_secs(30), @@ -1114,7 +1164,7 @@ fn validate_token_review_status_accepts_confirmed_audience() { ) .unwrap(); - assert_eq!(username, "system:serviceaccount:space:space-api"); + assert_eq!(username.username, "system:serviceaccount:space:space-api"); } #[test] diff --git a/crates/libs/common/svc/src/auth/mod.rs b/crates/libs/common/svc/src/auth/mod.rs index 4673bb01be..75b8e5e955 100644 --- a/crates/libs/common/svc/src/auth/mod.rs +++ b/crates/libs/common/svc/src/auth/mod.rs @@ -12,8 +12,10 @@ use tower::{Layer, Service}; use crate::space_api_route::SPACE_API_BRIDGE_CALLER_POLICY_METADATA_KEY; pub mod build; +pub mod node_identity; pub mod template; pub mod template_transfer; +pub use node_identity::{NodeIdentityVerification, NodeIdentityVerifier}; #[cfg(all(feature = "k8s", any(test, feature = "test-support")))] pub mod test_support; @@ -288,6 +290,7 @@ struct NodeApiAuthenticator { /// without re-parsing or trusting caller-controlled metadata. #[derive(Clone, Debug, Eq, PartialEq)] pub struct NodeApiAuthenticatedCaller { + kubernetes_pod: Option, wait_ready_lifecycle_authority: WaitReadyLifecycleAuthority, network_policy_authority: NetworkPolicyMutationAuthority, data_plane_session_refresh_authority: DataPlaneSessionRefreshAuthority, @@ -295,6 +298,40 @@ pub struct NodeApiAuthenticatedCaller { template_transfer_authority: template_transfer::TemplateTransferAuthority, } +/// Pod and node binding returned by an audience-checked Kubernetes TokenReview. +/// This records token provenance; consumers still validate current pod, node, +/// peer address, and operation ownership before granting node-scoped writes. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct KubernetesPodIdentity { + namespace: String, + service_account: String, + pod_name: pplx_space_common_core::types::PodName, + pod_uid: pplx_space_common_core::types::PodUid, + node_name: pplx_space_common_core::types::K8sNodeName, + node_uid: pplx_space_common_core::types::K8sNodeUid, +} + +impl KubernetesPodIdentity { + pub fn namespace(&self) -> &str { + &self.namespace + } + pub fn service_account(&self) -> &str { + &self.service_account + } + pub fn pod_name(&self) -> &pplx_space_common_core::types::PodName { + &self.pod_name + } + pub fn pod_uid(&self) -> &pplx_space_common_core::types::PodUid { + &self.pod_uid + } + pub fn node_name(&self) -> &pplx_space_common_core::types::K8sNodeName { + &self.node_name + } + pub fn node_uid(&self) -> &pplx_space_common_core::types::K8sNodeUid { + &self.node_uid + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum WaitReadyLifecycleAuthority { Denied, @@ -360,6 +397,10 @@ enum KubernetesAuthenticationMode { } impl NodeApiAuthenticatedCaller { + pub fn kubernetes_pod(&self) -> Option<&KubernetesPodIdentity> { + self.kubernetes_pod.as_ref() + } + const fn static_token(mode: KubernetesAuthenticationMode) -> Self { // In a static-only deployment (no Kubernetes auth) the shared internal // token is the trust boundary, so it retains full authority. Once @@ -389,6 +430,7 @@ impl NodeApiAuthenticatedCaller { ), }; Self { + kubernetes_pod: None, wait_ready_lifecycle_authority, network_policy_authority, data_plane_session_refresh_authority, @@ -412,6 +454,7 @@ impl NodeApiAuthenticatedCaller { data_plane_session_refresh_authority: DataPlaneSessionRefreshAuthority, ) -> Self { Self { + kubernetes_pod: None, wait_ready_lifecycle_authority, network_policy_authority, data_plane_session_refresh_authority, @@ -425,6 +468,7 @@ impl NodeApiAuthenticatedCaller { #[cfg(any(test, feature = "test-support"))] pub const fn data_plane_edge_for_test() -> Self { Self { + kubernetes_pod: None, wait_ready_lifecycle_authority: WaitReadyLifecycleAuthority::Denied, network_policy_authority: NetworkPolicyMutationAuthority::Denied, data_plane_session_refresh_authority: DataPlaneSessionRefreshAuthority::Granted, @@ -437,6 +481,7 @@ impl NodeApiAuthenticatedCaller { #[cfg(any(test, feature = "test-support"))] pub const fn unprivileged_workload_for_test() -> Self { Self { + kubernetes_pod: None, wait_ready_lifecycle_authority: WaitReadyLifecycleAuthority::Denied, network_policy_authority: NetworkPolicyMutationAuthority::Denied, data_plane_session_refresh_authority: DataPlaneSessionRefreshAuthority::Denied, diff --git a/crates/libs/common/svc/src/auth/node_identity.rs b/crates/libs/common/svc/src/auth/node_identity.rs new file mode 100644 index 0000000000..ad2c42e06b --- /dev/null +++ b/crates/libs/common/svc/src/auth/node_identity.rs @@ -0,0 +1,240 @@ +//! Live verification for RPCs whose authority belongs to one worker incarnation. + +use super::KubernetesPodIdentity; +use tonic::{Request, Status}; + +/// Required verification waits for capacity; advisory verification never queues +/// and has a short independent budget. Neither infers identity on failure. +#[derive(Clone, Copy)] +pub enum NodeIdentityVerification { + Required, + Advisory, +} + +/// Request-scoped live verification. The service injects its Kubernetes client; +/// checks never initialize clients or consult a process-global cache. Every +/// positive verdict is from this request's bounded live GET. +#[derive(Clone)] +#[cfg_attr(not(feature = "k8s"), derive(Default))] +pub struct NodeIdentityVerifier { + #[cfg(feature = "k8s")] + client: Option, + #[cfg(feature = "k8s")] + checks: std::sync::Arc, +} + +#[cfg(feature = "k8s")] +impl Default for NodeIdentityVerifier { + fn default() -> Self { + Self { + #[cfg(feature = "k8s")] + client: None, + #[cfg(feature = "k8s")] + checks: std::sync::Arc::new(tokio::sync::Semaphore::new(8)), + } + } +} + +impl NodeIdentityVerifier { + #[cfg(feature = "k8s")] + pub fn new(client: kube::Client) -> Self { + Self { + client: Some(client), + checks: std::sync::Arc::new(tokio::sync::Semaphore::new(8)), + } + } + + pub async fn verify( + &self, + request: &Request, + verification: NodeIdentityVerification, + ) -> Result { + let identity = request + .extensions() + .get::() + .and_then(|caller| caller.kubernetes_pod()) + .cloned() + .ok_or_else(|| { + Status::permission_denied("a complete Kubernetes pod-bound identity is required") + })?; + let peer = request + .remote_addr() + .map(|address| address.ip()) + .ok_or_else(|| Status::permission_denied("node-bound caller has no peer address"))?; + self.verify_identity(identity, peer, verification).await + } + + #[cfg(feature = "k8s")] + pub async fn verify_identity( + &self, + identity: KubernetesPodIdentity, + peer: std::net::IpAddr, + verification: NodeIdentityVerification, + ) -> Result { + let client = self + .client + .as_ref() + .ok_or_else(|| Status::unavailable("Kubernetes identity client unavailable"))?; + with_verification_budget(&self.checks, verification, async { + let pods: kube::Api = + kube::Api::namespaced(client.clone(), identity.namespace()); + let pod = + pods.get(identity.pod_name().as_str()) + .await + .map_err(|error| match error { + kube::Error::Api(response) if response.code == 404 => { + Status::permission_denied("caller pod no longer exists") + } + _ => Status::unavailable("caller pod could not be verified"), + })?; + verify_observed_pod(&identity, &pod, peer)?; + Ok(identity) + }) + .await + } + + #[cfg(not(feature = "k8s"))] + async fn verify_identity( + &self, + _identity: KubernetesPodIdentity, + _peer: std::net::IpAddr, + _verification: NodeIdentityVerification, + ) -> Result { + Err(Status::unavailable( + "Kubernetes identity verification is not compiled in", + )) + } +} + +/// TokenReview extras require exactly one nonempty value. Shared with credential +/// authentication; accepting the first of several bindings would be ambiguous. +pub fn single_extra_value<'a>( + extra: Option<&'a std::collections::BTreeMap>>, + key: &str, +) -> Option<&'a str> { + match extra?.get(key)?.as_slice() { + [value] if !value.is_empty() => Some(value), + _ => None, + } +} + +#[cfg(feature = "k8s")] +async fn with_verification_budget( + checks: &tokio::sync::Semaphore, + verification: NodeIdentityVerification, + check: impl std::future::Future>, +) -> Result { + use std::time::Duration; + let budget = match verification { + NodeIdentityVerification::Required => Duration::from_secs(5), + NodeIdentityVerification::Advisory => Duration::from_millis(50), + }; + tokio::time::timeout(budget, async { + let _permit = match verification { + NodeIdentityVerification::Required => checks + .acquire() + .await + .map_err(|_| Status::unavailable("identity verifier closed"))?, + NodeIdentityVerification::Advisory => checks + .try_acquire() + .map_err(|_| Status::unavailable("advisory identity verifier is busy"))?, + }; + check.await + }) + .await + .map_err(|_| Status::unavailable("caller pod verification timed out"))? +} + +#[cfg(feature = "k8s")] +fn verify_observed_pod( + identity: &KubernetesPodIdentity, + pod: &k8s_openapi::api::core::v1::Pod, + peer: std::net::IpAddr, +) -> Result<(), Status> { + let spec = pod + .spec + .as_ref() + .ok_or_else(|| Status::permission_denied("caller pod has no spec"))?; + check_pod_uid(pod, identity.pod_uid())?; + check_pod_running(pod)?; + check_peer_ip(pod, Some(peer))?; + if spec.node_name.as_deref() != Some(identity.node_name().as_str()) + || spec.service_account_name.as_deref() != Some(identity.service_account()) + { + return Err(Status::permission_denied( + "caller pod node or service account changed", + )); + } + if pod.metadata.deletion_timestamp.is_some() { + // A rolling restart is not an ownership violation. The durable worker + // must retry on the successor instead of terminalizing its upload. + return Err(Status::unavailable( + "caller daemon is draining; retry on successor", + )); + } + Ok(()) +} + +#[cfg(all(test, feature = "k8s"))] +mod tests; + +#[cfg(feature = "k8s")] +use k8s_openapi::api::core::v1::Pod; +#[cfg(feature = "k8s")] +use pplx_space_common_core::types::PodUid; +#[cfg(feature = "k8s")] +use std::net::IpAddr; + +#[cfg(feature = "k8s")] +pub fn check_pod_uid(pod: &Pod, expected: &PodUid) -> Result<(), Status> { + if pod.metadata.uid.as_deref() != Some(expected.as_str()) { + return Err(Status::permission_denied("pod_uid_mismatch")); + } + Ok(()) +} + +#[cfg(feature = "k8s")] +pub fn check_pod_running(pod: &Pod) -> Result<(), Status> { + if pod + .status + .as_ref() + .and_then(|status| status.phase.as_deref()) + != Some("Running") + { + return Err(Status::permission_denied("pod_not_running")); + } + Ok(()) +} + +#[cfg(feature = "k8s")] +pub fn check_peer_ip(pod: &Pod, peer: Option) -> Result<(), Status> { + let peer = peer + .map(normalize_ip) + .ok_or_else(|| Status::permission_denied("peer_address_missing"))?; + let matches = pod + .status + .as_ref() + .and_then(|status| status.pod_ips.as_ref()) + .is_some_and(|ips| { + ips.iter().any(|ip| { + ip.ip + .parse::() + .is_ok_and(|ip| normalize_ip(ip) == peer) + }) + }); + if !matches { + return Err(Status::permission_denied("peer_ip_mismatch")); + } + Ok(()) +} + +#[cfg(feature = "k8s")] +fn normalize_ip(ip: IpAddr) -> IpAddr { + match ip { + IpAddr::V6(ip) => ip + .to_ipv4_mapped() + .map(IpAddr::V4) + .unwrap_or(IpAddr::V6(ip)), + ip => ip, + } +} diff --git a/crates/libs/common/svc/src/auth/node_identity/tests.rs b/crates/libs/common/svc/src/auth/node_identity/tests.rs new file mode 100644 index 0000000000..d9454fdc47 --- /dev/null +++ b/crates/libs/common/svc/src/auth/node_identity/tests.rs @@ -0,0 +1,79 @@ +use super::*; + +#[test] +fn live_identity_rejects_replaced_or_stopped_daemon_and_wrong_peer() { + let identity = KubernetesPodIdentity { + namespace: "space".into(), + service_account: "worker".into(), + pod_name: "worker-daemon".parse().unwrap(), + pod_uid: "pod-a".parse().unwrap(), + node_name: "worker-a".parse().unwrap(), + node_uid: "node-a".parse().unwrap(), + }; + let pod: k8s_openapi::api::core::v1::Pod = serde_json::from_value(serde_json::json!({ + "metadata": {"name": "worker-daemon", "uid": "pod-a"}, + "spec": {"nodeName": "worker-a", "serviceAccountName": "worker"}, + "status": {"phase": "Running", "podIPs": [{"ip": "192.0.2.1"}]} + })) + .unwrap(); + let peer = "192.0.2.1".parse().unwrap(); + verify_observed_pod(&identity, &pod, peer).unwrap(); + for change in 0..7 { + let mut changed = pod.clone(); + let mut actual_peer = peer; + match change { + 0 => changed.metadata.uid = Some("pod-b".into()), + 1 => changed.spec.as_mut().unwrap().node_name = Some("worker-b".into()), + 2 => changed.spec.as_mut().unwrap().service_account_name = Some("other".into()), + 3 => changed.status.as_mut().unwrap().phase = Some("Succeeded".into()), + 4 => actual_peer = "192.0.2.2".parse().unwrap(), + 5 | 6 => { + changed.metadata.deletion_timestamp = Some( + serde_json::from_value(serde_json::json!("2026-01-01T00:00:00Z")).unwrap(), + ); + if change == 6 { + changed.metadata.uid = Some("pod-b".into()); + } + } + _ => unreachable!(), + } + assert_eq!( + verify_observed_pod(&identity, &changed, actual_peer) + .unwrap_err() + .code(), + if change == 5 { + tonic::Code::Unavailable + } else { + tonic::Code::PermissionDenied + } + ); + } +} + +#[tokio::test] +async fn advisory_identity_verification_never_queues_or_outlives_its_budget() { + let checks = tokio::sync::Semaphore::new(1); + let permit = checks.acquire().await.unwrap(); + let error = + with_verification_budget::<()>(&checks, NodeIdentityVerification::Advisory, async { + panic!("a busy advisory verifier must not begin I/O"); + }) + .await + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::Unavailable); + drop(permit); + let error = with_verification_budget( + &checks, + NodeIdentityVerification::Advisory, + std::future::pending::>(), + ) + .await + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::Unavailable); + // Cancellation returns the permit; required work remains serviceable. + with_verification_budget(&checks, NodeIdentityVerification::Required, async { + Ok(()) + }) + .await + .unwrap(); +} diff --git a/crates/libs/credential/mgr/server/src/k8s_identity.rs b/crates/libs/credential/mgr/server/src/k8s_identity.rs index 7421cac79f..a8f54e1c9e 100644 --- a/crates/libs/credential/mgr/server/src/k8s_identity.rs +++ b/crates/libs/credential/mgr/server/src/k8s_identity.rs @@ -44,7 +44,10 @@ use sha2::{Digest, Sha256}; use tokio_util::sync::CancellationToken; use tonic::Status; -use pplx_space_common_core::types::{PodName, PodUid}; +use pplx_space_common_core::types::PodName; +use pplx_space_common_svc::auth::node_identity::{ + check_peer_ip, check_pod_running, check_pod_uid, single_extra_value, +}; use pplx_space_common_svc::sql::node_registration::registered_node; use crate::auth::{ClientService, K8sIdentity, K8sNodeIdentity}; @@ -891,67 +894,6 @@ fn validate_token_review_status( }) } -fn single_extra_value<'a>( - extra: Option<&'a std::collections::BTreeMap>>, - key: &str, -) -> Option<&'a str> { - let values = extra?.get(key)?; - match values.as_slice() { - [value] if !value.is_empty() => Some(value), - _ => None, - } -} - -fn check_pod_uid(pod: &Pod, expected: &PodUid) -> Result<(), Status> { - if pod.metadata.uid.as_deref() != Some(expected.as_str()) { - return Err(Status::permission_denied("pod_uid_mismatch")); - } - Ok(()) -} - -fn check_pod_running(pod: &Pod) -> Result<(), Status> { - if pod - .status - .as_ref() - .and_then(|status| status.phase.as_deref()) - != Some("Running") - { - return Err(Status::permission_denied("pod_not_running")); - } - Ok(()) -} - -fn check_peer_ip(pod: &Pod, peer: Option) -> Result<(), Status> { - let peer = peer - .map(normalize_ip) - .ok_or_else(|| Status::permission_denied("peer_address_missing"))?; - let matches = pod - .status - .as_ref() - .and_then(|status| status.pod_ips.as_ref()) - .is_some_and(|ips| { - ips.iter().any(|ip| { - ip.ip - .parse::() - .is_ok_and(|ip| normalize_ip(ip) == peer) - }) - }); - if !matches { - return Err(Status::permission_denied("peer_ip_mismatch")); - } - Ok(()) -} - -fn normalize_ip(ip: IpAddr) -> IpAddr { - match ip { - IpAddr::V6(ip) => ip - .to_ipv4_mapped() - .map(IpAddr::V4) - .unwrap_or(IpAddr::V6(ip)), - ip => ip, - } -} - fn token_digest(token: &str) -> [u8; 32] { let mut hasher = Sha256::new(); hasher.update(token.as_bytes()); diff --git a/crates/libs/internal/api/server/Cargo.toml b/crates/libs/internal/api/server/Cargo.toml index 7eca936f31..251bceb2d9 100644 --- a/crates/libs/internal/api/server/Cargo.toml +++ b/crates/libs/internal/api/server/Cargo.toml @@ -64,6 +64,7 @@ pplx-space-snapshot-store = { workspace = true } pplx-space-template-mgr-client = { workspace = true } [dev-dependencies] +axum = { workspace = true } pplx-space-common-core = { workspace = true, features = ["serde", "sqlx", "test-hooks"] } hex = { workspace = true } insta = { workspace = true, features = ["yaml", "redactions"] } diff --git a/crates/libs/internal/api/server/src/backup_ownership.rs b/crates/libs/internal/api/server/src/backup_ownership.rs new file mode 100644 index 0000000000..28bb628910 --- /dev/null +++ b/crates/libs/internal/api/server/src/backup_ownership.rs @@ -0,0 +1,235 @@ +//! Backup authority is historical evidence plus a freshly authenticated publisher. +use super::ClusterApiService; +use pplx_space_common_core::types::{ExecutionGeneration, K8sNodeUid, SubstrateKind}; +use pplx_space_common_svc::auth::{ + KubernetesPodIdentity, NodeApiAuthenticatedCaller, NodeIdentityVerification, +}; +use pplx_space_snapshot_store::{ + self as store, InsertOutcome, SnapshotRecord, backup_authority as authority, +}; +use tonic::{Request, Status}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum IdentityMode { + Shadow, + Compatibility, + Strict, +} +impl IdentityMode { + pub(super) fn from_env() -> Result { + match std::env::var("SPACE_BACKUP_IDENTITY_MODE").as_deref() { + Err(std::env::VarError::NotPresent) | Ok("shadow") => Ok(Self::Shadow), + Ok("compatibility") => Ok(Self::Compatibility), + Ok("strict") => Ok(Self::Strict), + _ => Err(Status::failed_precondition( + "SPACE_BACKUP_IDENTITY_MODE must be shadow, compatibility, or strict", + )), + } + } +} + +pub(super) struct Caller { + pub remote_addr: Option, + identity: Option>, +} +impl Caller { + fn verified(&self) -> Result<&KubernetesPodIdentity, Status> { + self.identity + .as_ref() + .ok_or_else(|| { + Status::permission_denied("backup requires a pod-bound worker identity") + })? + .as_ref() + .map_err(Clone::clone) + } +} +pub(super) fn database_error(error: sqlx::Error) -> Status { + Status::internal(format!("backup authority: {error}")) +} + +impl ClusterApiService { + pub(super) async fn backup_caller(&self, request: &Request) -> Caller { + // Verification is bounded and completed before any database lock. Static + // authentication carries no historical Kubernetes incarnation evidence. + let bound = request + .extensions() + .get::() + .and_then(|caller| caller.kubernetes_pod()) + .is_some(); + Caller { + remote_addr: request.remote_addr(), + identity: if bound { + Some( + self.backup_identity + .verify(request, NodeIdentityVerification::Required) + .await, + ) + } else { + None + }, + } + } + + async fn check_incarnation( + &self, + conn: &mut sqlx::PgConnection, + attempt: &SnapshotRecord, + uid: Option<&K8sNodeUid>, + caller: &Caller, + ) -> Result, Status> { + // Shadow deploys readers before issuing new authority. Legacy rows keep + // the existing peer/generation policy even when inventory or live lookup + // is unavailable. Persisted proof still takes the required path below. + if uid.is_none() && self.backup_identity_mode == IdentityMode::Shadow { + return Ok(None); + } + let source = attempt + .source_node_id() + .ok_or_else(|| Status::permission_denied("backup has no source node"))?; + let node = authority::node_incarnation(conn, source) + .await + .map_err(database_error)? + .ok_or_else(|| Status::permission_denied("backup source node no longer registered"))?; + match node.managed_by { + SubstrateKind::Space if uid.is_none() => return Ok(None), + SubstrateKind::Space => { + return Err(Status::permission_denied( + "verified source changed substrate", + )); + } + SubstrateKind::Kubernetes => (), + } + if uid.is_none() + && caller.identity.is_none() + && self.backup_identity_mode != IdentityMode::Strict + { + return Ok(None); + } + let identity = caller.verified()?; + if node.node_name.as_ref() != Some(identity.node_name()) + || node.node_uid.as_ref() != Some(identity.node_uid()) + || uid.is_some_and(|uid| uid != identity.node_uid()) + { + return Err(Status::permission_denied( + "backup publisher incarnation does not match its recorded source", + )); + } + Ok(Some(identity.clone())) + } + + pub(super) async fn authorize_backup_incarnation( + &self, + conn: &mut sqlx::PgConnection, + attempt: &SnapshotRecord, + caller: &Caller, + ) -> Result<(), Status> { + self.check_incarnation(conn, attempt, attempt.source_node_uid(), caller) + .await?; + if attempt.source_node_uid().is_none() && self.backup_identity_mode == IdentityMode::Strict + { + let node = authority::node_incarnation( + conn, + attempt.source_node_id().expect("source checked"), + ) + .await + .map_err(database_error)?; + if node.is_some_and(|node| node.managed_by == SubstrateKind::Kubernetes) { + return Err(Status::failed_precondition( + "legacy backup has no recorded incarnation; explicit recovery required", + )); + } + } + Ok(()) + } + + pub(super) async fn authorize_backup_registration( + &self, + conn: &mut sqlx::PgConnection, + attempt: &SnapshotRecord, + outcome: InsertOutcome, + prior: Option<&authority::Registration>, + generation: Option, + caller: &Caller, + ) -> Result<(), Status> { + if outcome != InsertOutcome::Created || attempt.source_node_uid().is_some() { + return self + .authorize_backup_incarnation(conn, attempt, caller) + .await; + } + let expected_uid = prior.map(|prior| &prior.source_node_uid); + let Some(identity) = self + .check_incarnation(conn, attempt, expected_uid, caller) + .await? + else { + return Ok(()); + }; + let proven = if prior.is_some() { + true + } else if let Some(generation) = generation { + // An existing unverified registration cannot acquire historical proof + // from today's registry or from a later execution's supplied generation. + let legacy = store::artifact_registered_by_another_attempt( + &mut *conn, + attempt.sandbox_id(), + attempt + .artifact() + .and_then(|artifact| artifact.local_artifact_id()) + .expect("registration artifact"), + attempt.id(), + ) + .await + .map_err(|error| Status::internal(format!("backup registration lookup: {error}")))?; + let execution = + authority::execution_incarnation(conn, attempt.sandbox_id(), generation) + .await + .map_err(database_error)?; + if execution.as_ref().is_some_and(|execution| { + execution + .node_name + .as_ref() + .is_some_and(|name| name != identity.node_name()) + || execution + .node_uid + .as_ref() + .is_some_and(|uid| uid != identity.node_uid()) + }) { + return Err(Status::permission_denied( + "capture execution belongs to another worker incarnation", + )); + } + !legacy + && execution.is_some_and(|execution| { + execution.node_name.as_ref() == Some(identity.node_name()) + && execution.node_uid.as_ref() == Some(identity.node_uid()) + }) + } else { + false + }; + if !proven { + return if self.backup_identity_mode == IdentityMode::Strict { + Err(Status::failed_precondition( + "backup execution incarnation is unknown; explicit recovery required", + )) + } else { + Ok(()) + }; + } + // A verified retry preserves its existing authority even after rollback + // to shadow; creating an unverified successor would downgrade enforcement. + if !authority::certify( + conn, + attempt.id(), + attempt.source_node_id().expect("source checked"), + identity.node_uid(), + prior.map(|prior| prior.id), + ) + .await + .map_err(database_error)? + { + return Err(Status::aborted( + "backup registration changed during identity verification", + )); + } + Ok(()) + } +} diff --git a/crates/libs/internal/api/server/src/lib.rs b/crates/libs/internal/api/server/src/lib.rs index 8268b9738e..e0f6daaa99 100644 --- a/crates/libs/internal/api/server/src/lib.rs +++ b/crates/libs/internal/api/server/src/lib.rs @@ -276,6 +276,8 @@ fn published_node_status_db(status: PublishedNodeStatus) -> Result>, reserved_template_names: template_publication::ReservedTemplateNames, trusted_template_storage: trusted_storage::TrustedTemplateStorageBuckets, @@ -570,6 +572,8 @@ impl ClusterApiService { let template_releases_store = template_releases_store::TemplateReleasesStore::new(pool.clone()); Self { + backup_identity: Default::default(), + backup_identity_mode: backup_ownership::IdentityMode::Shadow, ingress_activation: None, reserved_template_names: template_publication::ReservedTemplateNames::default(), trusted_template_storage: trusted_storage::TrustedTemplateStorageBuckets::default(), @@ -633,6 +637,27 @@ impl ClusterApiService { service.trusted_template_storage = trusted_storage::TrustedTemplateStorageBuckets::from_env() .map_err(|error| Status::failed_precondition(error.to_string()))?; + service.backup_identity_mode = backup_ownership::IdentityMode::from_env()?; + #[cfg(feature = "kubernetes")] + { + match tokio::time::timeout( + std::time::Duration::from_secs(5), + kube::Client::try_default(), + ) + .await + { + Ok(Ok(client)) => { + service.backup_identity = + pplx_space_common_svc::auth::NodeIdentityVerifier::new(client) + } + _ if service.backup_identity_mode == backup_ownership::IdentityMode::Strict => { + return Err(Status::unavailable("backup identity client unavailable")); + } + _ => tracing::warn!( + "backup identity client unavailable; verified attempts remain fail-closed" + ), + } + } Ok(service) } @@ -660,7 +685,7 @@ impl ClusterApiService { bytes_uploaded: i64, failure_reason: Option<&str>, artifact_digests: &[snapshot_store::SnapshotArtifactDigest], - remote_addr: Option, + caller: &backup_ownership::Caller, ) -> Result<(), Status> { let attempt = self .registered_backup_attempt(snapshot_id) @@ -670,7 +695,7 @@ impl ClusterApiService { .authorize_backup_source_node( &self.pool, attempt.source_node_id(), - remote_addr, + caller.remote_addr, "backup attempt", ) .await?; @@ -692,6 +717,12 @@ impl ClusterApiService { .await .map_err(|e| Status::internal(format!("begin finalize snapshot tx: {e}")))?; + snapshot_store::backup_authority::lock_placement(&mut tx, attempt.sandbox_id()) + .await + .map_err(backup_ownership::database_error)?; + self.authorize_backup_incarnation(&mut tx, &attempt, caller) + .await?; + // `RETURNING` the sandbox so a READY transition can wake the // cohort-restore listener without a second lookup, and the node and // artifact identity so it can clear exactly this artifact's row from @@ -708,12 +739,16 @@ impl ClusterApiService { END WHERE id = $1 AND state IN ('creating', $2) + AND source_node_id = $5 + AND source_node_uid IS NOT DISTINCT FROM $6 RETURNING sandbox_id, source_node_id, local_artifact_id, artifact_kind" )) .bind(snapshot_id) .bind(target_state.as_str()) .bind(size_bytes) .bind(failure_reason) + .bind(source_node) + .bind(attempt.source_node_uid()) .fetch_optional(&mut *tx) .await .map_err(|e| Status::internal(format!("finalize snapshot: {e}")))?; @@ -833,18 +868,43 @@ impl ClusterApiService { /// no-op — the heartbeat is advisory, not a correctness gate, so there's /// no value in distinguishing "missing" from "already terminal" with an /// error the caller would only swallow. - async fn touch_backup_attempt(&self, snapshot_id: SnapshotId) -> Result<(), Status> { - sqlx::query(sql!( - "UPDATE snapshots - SET last_attempt_at = NOW() - WHERE id = $1 - AND state = 'creating'" - )) - .bind(snapshot_id) - .execute(&self.pool) + async fn touch_backup_attempt( + &self, + snapshot_id: SnapshotId, + caller: &backup_ownership::Caller, + ) -> Result<(), Status> { + let Some(attempt) = self.registered_backup_attempt(snapshot_id).await? else { + return Ok(()); + }; + let mut tx = self + .pool + .begin() + .await + .map_err(backup_ownership::database_error)?; + snapshot_store::backup_authority::lock_placement(&mut tx, attempt.sandbox_id()) + .await + .map_err(backup_ownership::database_error)?; + let source = self + .authorize_backup_source_node( + &mut *tx, + attempt.source_node_id(), + caller.remote_addr, + "backup attempt", + ) + .await?; + self.authorize_backup_incarnation(&mut tx, &attempt, caller) + .await?; + snapshot_store::backup_authority::touch( + &mut tx, + snapshot_id, + source, + attempt.source_node_uid(), + ) .await - .map_err(|e| Status::internal(format!("touch snapshot: {e}")))?; - + .map_err(backup_ownership::database_error)?; + tx.commit() + .await + .map_err(backup_ownership::database_error)?; Ok(()) } @@ -852,7 +912,7 @@ impl ClusterApiService { &self, request: Request, ) -> Result, Status> { - let remote_addr = request.remote_addr(); + let caller = self.backup_caller(&request).await; let req = request.into_inner(); let sandbox_id = SandboxId::from_str(req.sandbox_id.trim()) .map_err(|e| Status::invalid_argument(format!("invalid sandbox_id: {e}")))?; @@ -925,7 +985,7 @@ impl ClusterApiService { .transpose() .map_err(|e| Status::invalid_argument(format!("invalid execution_generation: {e}")))?; let sandbox = self.lookup_snapshot_sandbox(sandbox_id).await?; - let snapshot = sandbox + let mut snapshot = sandbox .new_snapshot( snapshot_id, sandbox_id, @@ -933,7 +993,7 @@ impl ClusterApiService { SnapshotState::Creating, storage_uri, )? - .with_artifact(artifact) + .with_artifact(artifact.clone()) .with_execution_metadata(execution_metadata); let mut tx = self .pool @@ -941,6 +1001,32 @@ impl ClusterApiService { .await .map_err(|e| Status::internal(format!("begin tx: {e}")))?; + let placement = snapshot_store::backup_authority::lock_placement(&mut tx, sandbox_id) + .await + .map_err(backup_ownership::database_error)? + .ok_or_else(|| Status::not_found("sandbox no longer exists"))?; + if snapshot_store::backup_authority::provenance_conflicts(&mut tx, sandbox_id, &artifact) + .await + .map_err(backup_ownership::database_error)? + { + return Err(Status::failed_precondition( + "artifact provenance conflicts with its existing identity", + )); + } + let prior = snapshot_store::backup_authority::prior_registration( + &mut tx, + sandbox_id, + local_artifact_id, + ) + .await + .map_err(backup_ownership::database_error)?; + snapshot = snapshot.with_source_node_id( + prior + .as_ref() + .map(|prior| prior.source_node_id) + .or(placement.source_node_id), + ); + let (persisted, outcome) = snapshot_store::insert(&mut tx, &snapshot) .await .map_err(snapshot_insert_status)?; @@ -980,10 +1066,19 @@ impl ClusterApiService { self.authorize_backup_source_node( &mut *tx, persisted.source_node_id(), - remote_addr, + caller.remote_addr, "backup attempt", ) .await?; + self.authorize_backup_registration( + &mut tx, + &persisted, + outcome, + prior.as_ref(), + execution_generation, + &caller, + ) + .await?; // A verdict-only request has had its answer: roll the insert back, so // asking can't leave a row that outranks the sandbox's ready snapshot. if req.fence_verdict_only { @@ -1007,7 +1102,7 @@ impl ClusterApiService { &self, request: Request, ) -> Result, Status> { - let remote_addr = request.remote_addr(); + let caller = self.backup_caller(&request).await; let req = request.into_inner(); let backup_attempt_id = BackupAttemptId::from_str(req.backup_attempt_id.trim()) .map_err(|e| Status::invalid_argument(format!("invalid backup_attempt_id: {e}")))?; @@ -1027,7 +1122,7 @@ impl ClusterApiService { req.bytes_uploaded, req.failure_reason.as_deref(), &artifact_digests, - remote_addr, + &caller, ) .await?; Ok(Response::new(FinalizeArtifactBackupResponse {})) @@ -1037,10 +1132,11 @@ impl ClusterApiService { &self, request: Request, ) -> Result, Status> { + let caller = self.backup_caller(&request).await; let req = request.into_inner(); let backup_attempt_id = BackupAttemptId::from_str(req.backup_attempt_id.trim()) .map_err(|e| Status::invalid_argument(format!("invalid backup_attempt_id: {e}")))?; - self.touch_backup_attempt(SnapshotId::from_uuid(backup_attempt_id.as_uuid())) + self.touch_backup_attempt(SnapshotId::from_uuid(backup_attempt_id.as_uuid()), &caller) .await?; Ok(Response::new(TouchArtifactBackupResponse {})) } @@ -1557,6 +1653,7 @@ fn normalize_ip(value: IpAddr) -> IpAddr { mod adapter; mod auto_lifecycle_delete_completions; mod backup_generation_fence; +mod backup_ownership; mod execution_observation; mod metadata; mod node_local_backup_pending; diff --git a/crates/libs/internal/api/server/src/tests.rs b/crates/libs/internal/api/server/src/tests.rs index 5abef4b4e3..8bba3a08b0 100644 --- a/crates/libs/internal/api/server/src/tests.rs +++ b/crates/libs/internal/api/server/src/tests.rs @@ -1,3 +1,5 @@ +#[cfg(feature = "kubernetes")] +mod backup_identity; mod prefetch_catalog; mod reconcile_claim_origins; mod substrate_scope; @@ -6417,9 +6419,12 @@ async fn finalize_unknown_snapshot_returns_not_found() -> TestResult { /// Helper to heartbeat a backup attempt through `TouchArtifactBackup`. async fn touch(service: &ClusterApiService, snapshot_id: SnapshotId) -> Result<(), tonic::Status> { service - .touch_artifact_backup(Request::new(TouchArtifactBackupRequest { - backup_attempt_id: snapshot_id.to_string(), - })) + .touch_artifact_backup(request_from_node( + TouchArtifactBackupRequest { + backup_attempt_id: snapshot_id.to_string(), + }, + TEST_WORKER_IP, + )) .await .map(|_| ()) } diff --git a/crates/libs/internal/api/server/src/tests/backup_generation_fence.rs b/crates/libs/internal/api/server/src/tests/backup_generation_fence.rs index 5cb73152cc..0d766bee26 100644 --- a/crates/libs/internal/api/server/src/tests/backup_generation_fence.rs +++ b/crates/libs/internal/api/server/src/tests/backup_generation_fence.rs @@ -241,6 +241,9 @@ async fn teardown_backup_racing_a_retire_sees_the_retired_execution() -> TestRes let mut retire = pool.begin().await?; retire_execution(&mut *retire, sandbox_id, 4).await?; + let blocker: i32 = sqlx::query_scalar(sql!("SELECT pg_backend_pid()")) + .fetch_one(&mut *retire) + .await?; let backup_attempt_id = SnapshotId::new(); let begin = begin_teardown_backup( @@ -255,9 +258,9 @@ async fn teardown_backup_racing_a_retire_sees_the_retired_execution() -> TestRes loop { let waiting: i64 = sqlx::query_scalar(sql!( "SELECT COUNT(*) FROM pg_stat_activity - WHERE wait_event_type = 'Lock' - AND query LIKE '%FROM sandbox_k8s_executions%'" + WHERE $1 = ANY(pg_blocking_pids(pid))" )) + .bind(blocker) .fetch_one(pool) .await?; if waiting > 0 { @@ -282,8 +285,8 @@ async fn teardown_backup_racing_a_retire_sees_the_retired_execution() -> TestRes /// A newer execution inserted while Begin is registering can't slip past the /// fence. The inserter holds the sandbox row `FOR UPDATE`, as every execution -/// insert does, so the snapshot insert's foreign-key check waits on it, and -/// the fence then reads the committed newer execution. +/// insert does. Begin waits on that same owner, then its fence reads the +/// committed newer execution. #[tokio::test(flavor = "current_thread")] async fn teardown_backup_racing_a_newer_execution_sees_it() -> TestResult { let Some(fixture) = cluster_fixture_with_max_connections(5).await? else { @@ -310,6 +313,9 @@ async fn teardown_backup_racing_a_newer_execution_sees_it() -> TestResult { .execute(&mut *claim) .await?; insert_execution(&mut *claim, sandbox_id, 5, "creating", true).await?; + let blocker: i32 = sqlx::query_scalar(sql!("SELECT pg_backend_pid()")) + .fetch_one(&mut *claim) + .await?; let backup_attempt_id = SnapshotId::new(); let begin = begin_teardown_backup( @@ -324,9 +330,9 @@ async fn teardown_backup_racing_a_newer_execution_sees_it() -> TestResult { loop { let waiting: i64 = sqlx::query_scalar(sql!( "SELECT COUNT(*) FROM pg_stat_activity - WHERE wait_event_type = 'Lock' - AND query LIKE '%INSERT INTO snapshots%'" + WHERE $1 = ANY(pg_blocking_pids(pid))" )) + .bind(blocker) .fetch_one(pool) .await?; if waiting > 0 { diff --git a/crates/libs/internal/api/server/src/tests/backup_identity.rs b/crates/libs/internal/api/server/src/tests/backup_identity.rs new file mode 100644 index 0000000000..deeeac5c1e --- /dev/null +++ b/crates/libs/internal/api/server/src/tests/backup_identity.rs @@ -0,0 +1,284 @@ +//! Real TokenReview middleware, live pod GET, RPCs, and PostgreSQL ownership. +use super::*; +use crate::proto::{cluster_api_client::ClusterApiClient, cluster_api_server::ClusterApiServer}; +use axum::{Json, routing::get}; +use pplx_space_common_svc::auth::{self, test_support::*}; +use std::sync::{ + Arc, + atomic::{AtomicU16, Ordering}, +}; +use tokio_stream::wrappers::TcpListenerStream; + +fn authenticated(body: T, token: &str) -> Request { + let mut request = Request::new(body); + request + .metadata_mut() + .insert("authorization", format!("Bearer {token}").parse().unwrap()); + request +} + +#[rstest::rstest] +#[case(crate::backup_ownership::IdentityMode::Shadow)] +#[case(crate::backup_ownership::IdentityMode::Compatibility)] +#[case(crate::backup_ownership::IdentityMode::Strict)] +#[tokio::test] +async fn verified_backup_rpcs_preserve_retry_authority_and_reject_replacement( + #[case] mode: crate::backup_ownership::IdentityMode, +) -> TestResult { + let Some(mut fixture) = cluster_fixture().await? else { + return Ok(()); + }; + let pool = fixture.service.pool.clone(); + let sandbox = seed_sandbox(&pool).await?; + let node: NodeId = sqlx::query_scalar(sql!("SELECT active_node_id FROM sandboxes WHERE id=$1")) + .bind(sandbox) + .fetch_one(&pool) + .await?; + sqlx::query(sql!("UPDATE nodes SET managed_by=$2 WHERE id=$1")) + .bind(node) + .bind(pplx_space_common_core::types::SubstrateKind::Kubernetes) + .execute(&pool) + .await?; + sqlx::query(sql!("INSERT INTO node_k8s(node_id,k8s_node_name,k8s_node_uid) VALUES($1,'worker-a','node-original')")).bind(node).execute(&pool).await?; + sqlx::query(sql!("INSERT INTO sandbox_k8s_executions(sandbox_id,generation,pod_name,pod_namespace,state,active,k8s_node_name,k8s_node_uid) VALUES($1,1,'sandbox-pod','space','retired',false,'worker-a','node-original')")).bind(sandbox).execute(&pool).await?; + let pod_status = Arc::new(AtomicU16::new(200)); + let status = pod_status.clone(); + let app = token_review_service(|_| { + let mut review = service_account_status("space-cluster-api", "system:serviceaccount:space:workers"); + review.user.as_mut().unwrap().extra = Some([ + ("authentication.kubernetes.io/pod-name".into(), vec!["worker-daemon".into()]), + ("authentication.kubernetes.io/pod-uid".into(), vec!["daemon-original".into()]), + ("authentication.kubernetes.io/node-name".into(), vec!["worker-a".into()]), + ("authentication.kubernetes.io/node-uid".into(), vec!["node-original".into()]), + ].into()); review + }).route("/api/v1/namespaces/space/pods/worker-daemon", get(move || { + let status = status.load(Ordering::SeqCst); + async move { (axum::http::StatusCode::from_u16(status).unwrap(), Json(if status == 200 { + serde_json::json!({"apiVersion":"v1","kind":"Pod","metadata":{"name":"worker-daemon","uid":"daemon-original"},"spec":{"nodeName":"worker-a","serviceAccountName":"workers"},"status":{"phase":"Running","podIPs":[{"ip":"127.0.0.1"}]}}) + } else { serde_json::json!({"apiVersion":"v1","kind":"Status","status":"Failure","message":"synthetic verification failure","reason":"Unknown","code":status}) })) } + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let kube_client = kube::Client::try_from(kube::Config::new( + format!("http://{}", listener.local_addr()?).parse()?, + ))?; + let api = tokio_util::task::AbortOnDropHandle::new(tokio::spawn(async move { + axum::serve(listener, app).await + })); + let mut env = pplx_space_test_env::TestEnv::default(); + for (key, value) in [ + ("SPACE_K8S_TOKEN_AUTH_ENABLED", "true"), + ("KUBERNETES_SERVICE_HOST", "synthetic"), + ("SPACE_K8S_TOKEN_AUTH_NAMESPACE", "space"), + ("SPACE_CLUSTER_API_AUTH_ALLOWED_SERVICE_ACCOUNTS", "workers"), + ] { + env.set(key, value); + } + let config = auth::NodeApiAuthConfig::from_environment( + &env, + auth::NodeRole::Cluster, + "synthetic-static".into(), + None, + )?; + let layer = auth::NodeApiAuthLayer::with_k8s_client_for_test(config, kube_client.clone()); + fixture.service.backup_identity = auth::NodeIdentityVerifier::new(kube_client); + fixture.service.backup_identity_mode = mode; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let endpoint = format!("http://{}", listener.local_addr()?); + let service = fixture.service.clone(); + let rpc = tokio_util::task::AbortOnDropHandle::new(tokio::spawn(async move { + tonic::transport::Server::builder() + .layer(layer) + .add_service(ClusterApiServer::new(service)) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await + })); + let mut client = ClusterApiClient::connect(endpoint).await?; + let token = projected_token("system:serviceaccount:space:workers"); + let artifact = LocalArtifactId::new(); + let first = SnapshotId::new(); + let request = |id: SnapshotId| BeginArtifactBackupRequest { + sandbox_id: sandbox.to_string(), + local_artifact_id: artifact.to_string(), + backup_attempt_id: id.to_string(), + storage_uri: test_storage_uri(sandbox, id), + artifact_kind: pplx_space_sbx_mgr_client::proto::ArtifactKind::Full as i32, + artifact_origin: pplx_space_sbx_mgr_client::proto::ArtifactOrigin::Suspend as i32, + execution: Some(test_snapshot_execution()), + execution_generation: Some(1), + ..Default::default() + }; + sqlx::query(sql!("UPDATE sandbox_k8s_executions SET k8s_node_uid='older-incarnation' WHERE sandbox_id=$1 AND generation=1")).bind(sandbox).execute(&pool).await?; + if mode == crate::backup_ownership::IdentityMode::Shadow { + // Reader rollout must tolerate historical inventory and verification + // outages without changing the existing legacy admission policy. + pod_status.store(503, Ordering::SeqCst); + sqlx::query(sql!( + "UPDATE node_k8s SET k8s_node_uid=NULL WHERE node_id=$1" + )) + .bind(node) + .execute(&pool) + .await?; + } else { + assert_eq!( + client + .begin_artifact_backup(authenticated(request(first), &token)) + .await + .unwrap_err() + .code(), + Code::PermissionDenied + ); + } + sqlx::query(sql!("UPDATE sandbox_k8s_executions SET k8s_node_uid='node-original' WHERE sandbox_id=$1 AND generation=1")).bind(sandbox).execute(&pool).await?; + client + .begin_artifact_backup(authenticated(request(first), &token)) + .await?; + let source_uid: Option = + sqlx::query_scalar(sql!("SELECT source_node_uid FROM snapshots WHERE id=$1")) + .bind(first) + .fetch_one(&pool) + .await?; + if mode == crate::backup_ownership::IdentityMode::Shadow { + assert_eq!( + source_uid, None, + "initial rollout must not issue proof before reader rollout" + ); + sqlx::query(sql!( + "UPDATE snapshots SET source_node_uid='node-original' WHERE id=$1" + )) + .bind(first) + .execute(&pool) + .await?; + sqlx::query(sql!( + "UPDATE node_k8s SET k8s_node_uid='node-original' WHERE node_id=$1" + )) + .bind(node) + .execute(&pool) + .await?; + assert_eq!( + client + .touch_artifact_backup(authenticated( + TouchArtifactBackupRequest { + backup_attempt_id: first.to_string() + }, + &token + )) + .await + .unwrap_err() + .code(), + Code::Unavailable + ); + pod_status.store(200, Ordering::SeqCst); + let retry = SnapshotId::new(); + client + .begin_artifact_backup(authenticated(request(retry), &token)) + .await?; + let retry_uid: Option = + sqlx::query_scalar(sql!("SELECT source_node_uid FROM snapshots WHERE id=$1")) + .bind(retry) + .fetch_one(&pool) + .await?; + assert_eq!( + retry_uid.as_deref(), + Some("node-original"), + "shadow rollback must not downgrade verified retry provenance" + ); + return Ok(()); + } + assert_eq!(source_uid.as_deref(), Some("node-original")); + // Move placement and materialize a successor. Prior verified bytes remain retryable. + sqlx::query(sql!( + "UPDATE snapshots SET timestamp=NOW()-INTERVAL '1 hour' WHERE id=$1" + )) + .bind(first) + .execute(&pool) + .await?; + sqlx::query(sql!( + "UPDATE sandboxes SET active_node_id=NULL,target_node_id=NULL WHERE id=$1" + )) + .bind(sandbox) + .execute(&pool) + .await?; + sqlx::query(sql!("INSERT INTO sandbox_k8s_executions(sandbox_id,generation,pod_name,pod_namespace,state,active,pod_materialized_at) VALUES($1,2,'successor-pod','space','running',true,NOW())")).bind(sandbox).execute(&pool).await?; + let retry = SnapshotId::new(); + client + .begin_artifact_backup(authenticated(request(retry), &token)) + .await?; + let clocks: (bool,bool) = sqlx::query_as(sql!("SELECT timestamp=(SELECT timestamp FROM snapshots WHERE id=$1),last_attempt_at>timestamp FROM snapshots WHERE id=$2")).bind(first).bind(retry).fetch_one(&pool).await?; + assert_eq!(clocks, (true, true)); + let touch = || TouchArtifactBackupRequest { + backup_attempt_id: retry.to_string(), + }; + let finalize = || FinalizeArtifactBackupRequest { + backup_attempt_id: retry.to_string(), + target_state: SuspendSnapshotTerminalState::Ready as i32, + bytes_uploaded: 10, + ..Default::default() + }; + client + .touch_artifact_backup(authenticated(touch(), &token)) + .await?; + for (status, expected) in [(404, Code::PermissionDenied), (503, Code::Unavailable)] { + pod_status.store(status, Ordering::SeqCst); + assert_eq!( + client + .touch_artifact_backup(authenticated(touch(), &token)) + .await + .unwrap_err() + .code(), + expected + ); + assert_eq!( + client + .finalize_artifact_backup(authenticated(finalize(), &token)) + .await + .unwrap_err() + .code(), + expected + ); + } + pod_status.store(200, Ordering::SeqCst); + // Static authentication cannot downgrade an already verified attempt. + assert_eq!( + client + .touch_artifact_backup(authenticated(touch(), "synthetic-static")) + .await + .unwrap_err() + .code(), + Code::PermissionDenied + ); + sqlx::query(sql!( + "UPDATE node_k8s SET k8s_node_uid='node-replacement' WHERE node_id=$1" + )) + .bind(node) + .execute(&pool) + .await?; + assert_eq!( + client + .begin_artifact_backup(authenticated(request(SnapshotId::new()), &token)) + .await + .unwrap_err() + .code(), + Code::PermissionDenied + ); + assert_eq!( + client + .finalize_artifact_backup(authenticated(finalize(), &token)) + .await + .unwrap_err() + .code(), + Code::PermissionDenied + ); + sqlx::query(sql!( + "UPDATE node_k8s SET k8s_node_uid='node-original' WHERE node_id=$1" + )) + .bind(node) + .execute(&pool) + .await?; + client + .finalize_artifact_backup(authenticated(finalize(), &token)) + .await?; + drop(client); + drop(rpc); + drop(api); + Ok(()) +} diff --git a/crates/libs/snapshot/store/src/backup_authority.rs b/crates/libs/snapshot/store/src/backup_authority.rs new file mode 100644 index 0000000000..5d7a37fd7f --- /dev/null +++ b/crates/libs/snapshot/store/src/backup_authority.rs @@ -0,0 +1,135 @@ +//! Historical publisher evidence. Inventory can veto a claim, never create it. +use super::*; +use pplx_space_common_core::types::{K8sNodeName, SubstrateKind}; + +#[derive(sqlx::FromRow)] +pub struct BackupPlacement { + pub source_node_id: Option, +} + +/// Serializes registration with placement changes and execution retirement. +pub async fn lock_placement( + conn: &mut sqlx::PgConnection, + sandbox: SandboxId, +) -> Result, sqlx::Error> { + sqlx::query_as(sql!("SELECT COALESCE(active_node_id,target_node_id) AS source_node_id FROM sandboxes WHERE id=$1 FOR UPDATE")) + .bind(sandbox).fetch_optional(conn).await +} + +#[derive(sqlx::FromRow)] +pub struct Registration { + pub id: SnapshotId, + pub source_node_id: NodeId, + pub source_node_uid: K8sNodeUid, +} + +/// Only a registration carrying original verified identity grants retry authority. +/// A legacy attempt remains unknown even if today's inventory has a Node UID. +pub async fn prior_registration( + conn: &mut sqlx::PgConnection, + sandbox: SandboxId, + artifact: LocalArtifactId, +) -> Result, sqlx::Error> { + sqlx::query_as(sql!("SELECT id,source_node_id,source_node_uid FROM snapshots + WHERE sandbox_id=$1 AND local_artifact_id=$2 AND source_node_id IS NOT NULL AND source_node_uid IS NOT NULL + ORDER BY timestamp,id LIMIT 1")) + .bind(sandbox).bind(artifact).fetch_optional(conn).await +} + +pub async fn provenance_conflicts( + conn: &mut sqlx::PgConnection, + sandbox: SandboxId, + artifact: &SnapshotArtifact, +) -> Result { + sqlx::query_scalar(sql!("SELECT EXISTS(SELECT 1 FROM snapshots WHERE sandbox_id=$1 AND local_artifact_id=$2 + AND (artifact_kind IS DISTINCT FROM $3 OR artifact_origin IS DISTINCT FROM $4 OR artifact_snapshot_name IS DISTINCT FROM $5)) + OR EXISTS(SELECT 1 FROM sandboxes WHERE id=$1 AND current_local_artifact_id=$2 + AND (current_local_artifact_kind IS DISTINCT FROM $3 OR current_local_artifact_origin IS DISTINCT FROM $4 OR $5::TEXT IS NOT NULL))")) + .bind(sandbox).bind(artifact.local_artifact_id()).bind(artifact.kind().as_str()).bind(artifact.origin().as_str()) + .bind(artifact.snapshot_name().map(SnapshotName::as_str)).fetch_one(conn).await +} + +#[derive(sqlx::FromRow)] +pub struct NodeIncarnation { + pub managed_by: SubstrateKind, + pub node_name: Option, + pub node_uid: Option, +} + +pub async fn node_incarnation( + conn: &mut sqlx::PgConnection, + node: NodeId, +) -> Result, sqlx::Error> { + // Lock each relation separately: an outer join cannot lock its nullable side. + let managed_by = sqlx::query_scalar(sql!( + "SELECT managed_by FROM nodes WHERE id=$1 AND role='worker' AND state<>'deleted' FOR SHARE" + )) + .bind(node) + .fetch_optional(&mut *conn) + .await?; + let Some(managed_by) = managed_by else { + return Ok(None); + }; + let binding: Option<(Option, Option)> = sqlx::query_as(sql!("SELECT k8s_node_name,NULLIF(BTRIM(k8s_node_uid),'') FROM node_k8s WHERE node_id=$1 FOR SHARE")) + .bind(node).fetch_optional(conn).await?; + let (node_name, node_uid) = binding.unwrap_or_default(); + Ok(Some(NodeIncarnation { + managed_by, + node_name, + node_uid, + })) +} + +/// Retirement alone does not erase the source's authority. The caller must +/// also apply the generation supersession fence and locked placement check. +#[derive(sqlx::FromRow)] +pub struct ExecutionIncarnation { + pub node_name: Option, + pub node_uid: Option, +} + +pub async fn execution_incarnation( + conn: &mut sqlx::PgConnection, + sandbox: SandboxId, + generation: ExecutionGeneration, +) -> Result, sqlx::Error> { + sqlx::query_as(sql!( + "SELECT k8s_node_name AS node_name, NULLIF(BTRIM(k8s_node_uid),'') AS node_uid FROM sandbox_k8s_executions + WHERE sandbox_id=$1 AND generation=$2 AND state<>'replaced' FOR SHARE" + )) + .bind(sandbox).bind(generation.get()).fetch_optional(conn).await +} + +/// Called only for the row this transaction inserted after proving its source. +/// Preserve capture freshness across attempts; heartbeat time describes liveness. +pub async fn certify( + conn: &mut sqlx::PgConnection, + attempt: SnapshotId, + node: NodeId, + uid: &K8sNodeUid, + prior: Option, +) -> Result { + Ok(sqlx::query(sql!("UPDATE snapshots SET source_node_uid=$3, + timestamp=COALESCE((SELECT timestamp FROM snapshots WHERE id=$4),timestamp), last_attempt_at=NOW() + WHERE id=$1 AND source_node_id=$2 AND source_node_uid IS NULL AND state='creating'")) + .bind(attempt).bind(node).bind(uid).bind(prior).execute(conn).await?.rows_affected()==1) +} + +/// Authentication and incarnation checks run before this guarded heartbeat. +pub async fn touch( + conn: &mut sqlx::PgConnection, + attempt: SnapshotId, + node: NodeId, + uid: Option<&K8sNodeUid>, +) -> Result<(), sqlx::Error> { + sqlx::query(sql!( + "UPDATE snapshots SET last_attempt_at=NOW() WHERE id=$1 AND state='creating' + AND source_node_id=$2 AND source_node_uid IS NOT DISTINCT FROM $3" + )) + .bind(attempt) + .bind(node) + .bind(uid) + .execute(conn) + .await?; + Ok(()) +} diff --git a/crates/libs/snapshot/store/src/lib.rs b/crates/libs/snapshot/store/src/lib.rs index 621cd35691..b7d1ebf002 100644 --- a/crates/libs/snapshot/store/src/lib.rs +++ b/crates/libs/snapshot/store/src/lib.rs @@ -6,10 +6,10 @@ use jiff::Timestamp; use pplx_space_common_core::sql; use pplx_space_common_core::types::{ ArtifactKind, ArtifactKindParseError, ArtifactOrigin, ArtifactOriginParseError, - ExactStorageUri, ExecutionGeneration, ExecutionGenerationParseError, LocalArtifactId, NodeId, - OrgId, SandboxId, SnapshotContentHash, SnapshotExecutionMetadata, SnapshotId, SnapshotKind, - SnapshotKindParseError, SnapshotName, SnapshotParseError, SnapshotState, - SnapshotStateParseError, TemplateName, TemplateParseError, + ExactStorageUri, ExecutionGeneration, ExecutionGenerationParseError, K8sNodeUid, + LocalArtifactId, NodeId, OrgId, SandboxId, SnapshotContentHash, SnapshotExecutionMetadata, + SnapshotId, SnapshotKind, SnapshotKindParseError, SnapshotName, SnapshotParseError, + SnapshotState, SnapshotStateParseError, TemplateName, TemplateParseError, }; use pplx_space_common_core::types::{ Digest, DigestParseError, SnapshotArtifact as SnapshotArtifactFile, SnapshotArtifactParseError, @@ -19,6 +19,8 @@ use sqlx::types::Json; use time::OffsetDateTime; use uuid::Uuid; +pub mod backup_authority; + /// Reading a snapshot catalog row failed. #[derive(Debug, thiserror::Error)] pub enum GetSnapshotError { @@ -228,6 +230,7 @@ pub struct SnapshotRecord { state: SnapshotState, timestamp: Timestamp, source_node_id: Option, + source_node_uid: Option, label: Option, retention_tier: Option, vcpu: Option, @@ -269,6 +272,10 @@ impl SnapshotRecord { self.source_node_id } + pub fn source_node_uid(&self) -> Option<&K8sNodeUid> { + self.source_node_uid.as_ref() + } + pub fn label(&self) -> Option<&str> { self.label.as_deref() } @@ -448,6 +455,7 @@ struct SnapshotRow { state: String, timestamp: OffsetDateTime, source_node_id: Option, + source_node_uid: Option, label: Option, retention_tier: Option, vcpu: Option, @@ -853,7 +861,7 @@ pub async fn get( ) -> Result, GetSnapshotError> { let row: Option = sqlx::query_as(sql!( "SELECT sn.id, sn.sandbox_id, sn.type AS snapshot_kind, sn.state, - sn.timestamp, sn.source_node_id, sn.label, sn.retention_tier, + sn.timestamp, sn.source_node_id, sn.source_node_uid, sn.label, sn.retention_tier, sn.vcpu, sn.memory_mib, sn.size_bytes, sn.template_id, sn.content_hash, sn.org_id, sn.failure_reason, sn.last_attempt_at, sn.local_artifact_id, sn.artifact_kind, @@ -947,6 +955,7 @@ impl TryFrom for SnapshotRecord { state, timestamp, source_node_id: row.source_node_id, + source_node_uid: row.source_node_uid, label: row.label, retention_tier: row.retention_tier, vcpu: row.vcpu, diff --git a/docs/architecture.md b/docs/architecture.md index d68b8b0994..f6412127ee 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -191,6 +191,28 @@ token-aware compatibility binary; pre-token binaries are no longer safe writers. The placement store requires an explicit creation mode and must receive the same policy when a controller adopts it. +### Backup publisher identity + +Backup publisher identity is rolled out with `SPACE_BACKUP_IDENTITY_MODE`, +initially `shadow`, which records no new verified source identity. Deploy all +identity-aware readers before enabling `compatibility` to certify new attempts; +rollback then requires an identity-aware binary. Pod-bound callers are checked against a live daemon +pod before acquiring database locks; a verified attempt always requires its +original worker incarnation, including in compatibility mode. Static Direct +workers retain address-based admission. Static Kubernetes fallback and legacy +execution records cannot mint historical Node UID evidence. Enable `strict` +only after all identity writers and workers have rolled out and retained legacy +artifacts have an operator recovery path. Verification failures use the existing +bounded, attempt-scoped upload retries; exhaustion retains the local source. + +A verified retry preserves the first registration's timestamp. `last_attempt_at` +is its liveness clock for the stuck-upload reaper and node termination gate; +restore selection and portability continue to use the original timestamp. +Retirement alone does not supersede a capture. Main's generation fence still +rejects a replaced execution or a newer active/materialized execution on first +registration. A verified prior registration can authorize an exact artifact +retry after relocation, without changing placement or clearing quarantine. + ### Lifecycle policy at activation Timeout updates use durable execution ownership rather than a lifecycle label