Skip to content

Bind backup RPCs to verified historical worker identity #345

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zibi-pplx wants to merge 1 commit into from
Closed

Bind backup RPCs to verified historical worker identity #345

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

83 changes: 66 additions & 17 deletions crates/libs/common/svc/src/auth/k8s.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down Expand Up @@ -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<CachedTokenReviewResult> {
Expand All @@ -330,10 +333,14 @@ impl K8sTokenReviewer {
cached: CachedTokenReviewResult,
) -> Result<NodeApiAuthenticatedCaller, Status> {
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");
Expand Down Expand Up @@ -445,7 +452,7 @@ impl K8sTokenReviewer {
&self,
token: &str,
audience: &str,
) -> Result<String, TokenReviewFailure> {
) -> Result<ReviewedToken, TokenReviewFailure> {
let review = TokenReview {
metadata: Default::default(),
spec: TokenReviewSpec {
Expand Down Expand Up @@ -860,8 +867,13 @@ struct CachedTokenReview {

#[derive(Clone, Debug, Eq, PartialEq)]
enum CachedTokenReviewResult {
Accepted { username: String },
Rejected { reason: TokenReviewRejectReason },
Accepted {
username: String,
pod: Option<super::KubernetesPodIdentity>,
},
Rejected {
reason: TokenReviewRejectReason,
},
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
Expand Down Expand Up @@ -1000,7 +1012,7 @@ fn jwt_exp_cache_deadline(token: &str, now: Instant, now_system: SystemTime) ->
fn validate_token_review_status(
status: Option<TokenReviewStatus>,
audience: &str,
) -> Result<String, TokenReviewFailure> {
) -> Result<ReviewedToken, TokenReviewFailure> {
let status = status.ok_or(TokenReviewFailure::Rejected {
reason: TokenReviewRejectReason::MissingStatus,
})?;
Expand All @@ -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<super::KubernetesPodIdentity>,
}

fn reviewed_pod_identity(
username: &str,
extra: Option<&std::collections::BTreeMap<String, Vec<String>>>,
) -> Option<super::KubernetesPodIdentity> {
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] {
Expand Down
52 changes: 51 additions & 1 deletion crates/libs/common/svc/src/auth/k8s/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 [
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand All @@ -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]
Expand Down
45 changes: 45 additions & 0 deletions crates/libs/common/svc/src/auth/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -288,13 +290,48 @@ struct NodeApiAuthenticator {
/// without re-parsing or trusting caller-controlled metadata.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct NodeApiAuthenticatedCaller {
kubernetes_pod: Option<KubernetesPodIdentity>,
wait_ready_lifecycle_authority: WaitReadyLifecycleAuthority,
network_policy_authority: NetworkPolicyMutationAuthority,
data_plane_session_refresh_authority: DataPlaneSessionRefreshAuthority,
template_release_read_authority: TemplateReleaseReadAuthority,
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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -389,6 +430,7 @@ impl NodeApiAuthenticatedCaller {
),
};
Self {
kubernetes_pod: None,
wait_ready_lifecycle_authority,
network_policy_authority,
data_plane_session_refresh_authority,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand Down
Loading
Loading