Add support for user impersonation

kube-client/src/client/config_ext.rs

@@ -1,11 +1,14 @@
+use std::sync::Arc;
+
+use http::{header::HeaderName, HeaderValue};
 use secrecy::ExposeSecret;
 use tower::{filter::AsyncFilterLayer, util::Either};
 
 #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "openssl-tls"))]
 use super::tls;
 use super::{
     auth::Auth,
-    middleware::{AddAuthorizationLayer, AuthLayer, BaseUriLayer},
+    middleware::{AddAuthorizationLayer, AuthLayer, BaseUriLayer, ExtraHeadersLayer},
 };
 use crate::{Config, Error, Result};
 
@@ -21,6 +24,9 @@ pub trait ConfigExt: private::Sealed {
     /// Optional layer to set up `Authorization` header depending on the config.
     fn auth_layer(&self) -> Result<Option<AuthLayer>>;
 
+    /// Layer to add non-authn HTTP headers depending on the config.
+    fn extra_headers_layer(&self) -> Result<ExtraHeadersLayer>;
+
     /// Create [`hyper_tls::HttpsConnector`] based on config.
     ///
     /// # Example
@@ -182,6 +188,31 @@ impl ConfigExt for Config {
         })
     }
 
+    fn extra_headers_layer(&self) -> Result<ExtraHeadersLayer> {
+        let mut headers = Vec::new();
+        if let Some(impersonate_user) = &self.auth_info.impersonate {
+            headers.push((
+                HeaderName::from_static("impersonate-user"),
+                HeaderValue::from_str(impersonate_user)
+                    .map_err(http::Error::from)
+                    .map_err(Error::HttpError)?,
+            ));
+        }
+        if let Some(impersonate_groups) = &self.auth_info.impersonate_groups {
+            for group in impersonate_groups {
+                headers.push((
+                    HeaderName::from_static("impersonate-group"),
+                    HeaderValue::from_str(group)
+                        .map_err(http::Error::from)
+                        .map_err(Error::HttpError)?,
+                ));
+            }
+        }
+        Ok(ExtraHeadersLayer {
+            headers: Arc::new(headers),
+        })
+    }
+
     #[cfg(feature = "native-tls")]
     fn native_tls_connector(&self) -> Result<tokio_native_tls::native_tls::TlsConnector> {
         tls::native_tls::native_tls_connector(

kube-client/src/client/middleware/extra_headers.rs

@@ -0,0 +1,46 @@
+use std::sync::Arc;
+
+use http::{header::HeaderName, request::Request, HeaderValue};
+use tower::{Layer, Service};
+
+#[derive(Clone)]
+/// Layer that adds a static set of extra headers to each request
+pub struct ExtraHeadersLayer {
+    pub(crate) headers: Arc<Vec<(HeaderName, HeaderValue)>>,
+}
+
+impl<S> Layer<S> for ExtraHeadersLayer {
+    type Service = ExtraHeaders<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        ExtraHeaders {
+            inner,
+            headers: self.headers.clone(),
+        }
+    }
+}
+
+#[derive(Clone)]
+/// Service that adds a static set of extra headers to each request
+pub struct ExtraHeaders<S> {
+    inner: S,
+    headers: Arc<Vec<(HeaderName, HeaderValue)>>,
+}
+
+impl<S, ReqBody> Service<Request<ReqBody>> for ExtraHeaders<S>
+where
+    S: Service<Request<ReqBody>>,
+{
+    type Error = S::Error;
+    type Future = S::Future;
+    type Response = S::Response;
+
+    fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut req: Request<ReqBody>) -> Self::Future {
+        req.headers_mut().extend(self.headers.iter().cloned());
+        self.inner.call(req)
+    }
+}

kube-client/src/client/middleware/mod.rs

@@ -3,8 +3,10 @@ use tower::{filter::AsyncFilterLayer, util::Either, Layer};
 pub(crate) use tower_http::auth::AddAuthorizationLayer;
 
 mod base_uri;
+mod extra_headers;
 
 pub use base_uri::{BaseUri, BaseUriLayer};
+pub use extra_headers::{ExtraHeaders, ExtraHeadersLayer};
 
 use super::auth::RefreshableToken;
 /// Layer to set up `Authorization` header depending on the config.
@@ -21,6 +23,7 @@ impl<S> Layer<S> for AuthLayer {
     }
 }
 
+
 #[cfg(test)]
 mod tests {
     use super::*;

kube-client/src/client/mod.rs

@@ -504,6 +504,7 @@ impl TryFrom<Config> for Client {
         let service = ServiceBuilder::new()
             .layer(stack)
             .option_layer(config.auth_layer()?)
+            .layer(config.extra_headers_layer()?)
             .layer(
                 // Attribute names follow [Semantic Conventions].
                 // [Semantic Conventions]: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md

kube-client/src/config/mod.rs

@@ -166,22 +166,24 @@ impl Config {
     /// then if that fails, trying the local kubeconfig.
     ///
     /// Fails if inference from both sources fails
+    ///
+    /// Applies debug overrides, see [`Config::apply_debug_overrides`] for more details
     pub async fn infer() -> Result<Self, InferConfigError> {
-        match Self::from_cluster_env() {
+        let mut config = match Self::from_cluster_env() {
             Err(in_cluster_err) => {
                 tracing::trace!("No in-cluster config found: {}", in_cluster_err);
                 tracing::trace!("Falling back to local kubeconfig");
-                let config = Self::from_kubeconfig(&KubeConfigOptions::default())
+                Self::from_kubeconfig(&KubeConfigOptions::default())
                     .await
                     .map_err(|kubeconfig_err| InferConfigError {
                         in_cluster: in_cluster_err,
                         kubeconfig: kubeconfig_err,
-                    })?;
-
-                Ok(config)
+                    })?
             }
-            Ok(success) => Ok(success),
-        }
+            Ok(success) => success,
+        };
+        config.apply_debug_overrides();
+        Ok(config)
     }
 
     /// Create configuration from the cluster's environment variables
@@ -271,6 +273,45 @@ impl Config {
         })
     }
 
+    /// Override configuration based on environment variables
+    ///
+    /// This is only intended for use as a debugging aid, and the specific variables and their behaviour
+    /// should **not** be considered stable across releases.
+    ///
+    /// Currently, the following overrides are supported:
+    ///
+    /// - `KUBE_RS_DEBUG_IMPERSONATE_USER`: A Kubernetes user to impersonate, for example: `system:serviceaccount:default:foo` will impersonate the `ServiceAccount` `foo` in the `Namespace` `default`
+    /// - `KUBE_RS_DEBUG_IMPERSONATE_GROUP`: A Kubernetes group to impersonate, multiple groups may be specified by separating them with commas
+    /// - `KUBE_RS_DEBUG_OVERRIDE_URL`: A Kubernetes cluster URL to use rather than the one specified in the config, useful for proxying traffic through `kubectl proxy`
+    #[tracing::instrument(level = "warn")]
+    pub fn apply_debug_overrides(&mut self) {
+        // Log these overrides loudly, to emphasize that this is only a debugging aid, and should not be relied upon in production
+        if let Ok(impersonate_user) = std::env::var("KUBE_RS_DEBUG_IMPERSONATE_USER") {
+            tracing::warn!(?impersonate_user, "impersonating user");
+            self.auth_info.impersonate = Some(impersonate_user);
+        }
+        if let Ok(impersonate_groups) = std::env::var("KUBE_RS_DEBUG_IMPERSONATE_GROUP") {
+            let impersonate_groups = impersonate_groups.split(',').map(str::to_string).collect();
+            tracing::warn!(?impersonate_groups, "impersonating groups");
+            self.auth_info.impersonate_groups = Some(impersonate_groups);
+        }
+        if let Ok(url) = std::env::var("KUBE_RS_DEBUG_OVERRIDE_URL") {
+            tracing::warn!(?url, "overriding cluster URL");
+            match url.parse() {
+                Ok(uri) => {
+                    self.cluster_url = uri;
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        ?url,
+                        error = &err as &dyn std::error::Error,
+                        "failed to parse override cluster URL, ignoring"
+                    );
+                }
+            }
+        }
+    }
+
     /// Client certificate and private key in PEM.
     pub(crate) fn identity_pem(&self) -> Option<Vec<u8>> {
         self.auth_info.identity_pem().ok()