kube_client/client/auth/
mod.rs

1use futures::future::BoxFuture;
2use http::{
3    HeaderValue, Request,
4    header::{AUTHORIZATION, InvalidHeaderValue},
5};
6use jiff::{SignedDuration, Timestamp};
7use jsonpath_rust::JsonPath;
8use secrecy::{ExposeSecret, SecretString};
9use serde::{Deserialize, Serialize};
10use std::{
11    path::{Path, PathBuf},
12    process::Command,
13    sync::Arc,
14};
15use thiserror::Error;
16use tokio::sync::{Mutex, RwLock};
17use tower::{BoxError, filter::AsyncPredicate};
18
19use crate::config::{AuthInfo, AuthProviderConfig, ExecAuthCluster, ExecConfig, ExecInteractiveMode};
20
21#[cfg(feature = "oauth")] mod oauth;
22#[cfg(feature = "oauth")] pub use oauth::Error as OAuthError;
23#[cfg(feature = "oidc")] mod oidc;
24#[cfg(feature = "oidc")] pub use oidc::errors as oidc_errors;
25#[cfg(target_os = "windows")] use std::os::windows::process::CommandExt;
26
27#[derive(Error, Debug)]
28/// Client auth errors
29pub enum Error {
30    /// Invalid basic auth
31    #[error("invalid basic auth: {0}")]
32    InvalidBasicAuth(#[source] InvalidHeaderValue),
33
34    /// Invalid bearer token
35    #[error("invalid bearer token: {0}")]
36    InvalidBearerToken(#[source] InvalidHeaderValue),
37
38    /// Tried to refresh a token and got a non-refreshable token response
39    #[error("tried to refresh a token and got a non-refreshable token response")]
40    UnrefreshableTokenResponse,
41
42    /// Exec plugin response did not contain a status
43    #[error("exec-plugin response did not contain a status")]
44    ExecPluginFailed,
45
46    /// Malformed token expiration date
47    #[error("malformed token expiration date: {0}")]
48    MalformedTokenExpirationDate(#[source] jiff::Error),
49
50    /// Failed to start auth exec
51    #[error("unable to run auth exec: {0}")]
52    AuthExecStart(#[source] std::io::Error),
53
54    /// Failed to run auth exec command
55    #[error("auth exec command '{cmd}' failed with status {}: {out:?}", .out.status)]
56    AuthExecRun {
57        /// The failed command
58        cmd: String,
59        /// Exit status and stdout/stderr of the failed command
60        out: Box<std::process::Output>,
61    },
62
63    /// Failed to parse auth exec output
64    #[error("failed to parse auth exec output: {0}")]
65    AuthExecParse(#[source] Box<serde_saphyr::Error>),
66
67    /// Fail to serialize input
68    #[error("failed to serialize input: {0}")]
69    AuthExecSerialize(#[source] serde_json::Error),
70
71    /// Failed to exec auth
72    #[error("failed exec auth: {0}")]
73    AuthExec(String),
74
75    /// Failed to read token file
76    #[error("failed to read token file '{1:?}': {0}")]
77    ReadTokenFile(#[source] std::io::Error, PathBuf),
78
79    /// Failed to parse token-key
80    #[error("failed to parse token-key: {0}")]
81    ParseTokenKey(#[source] serde_json::Error),
82
83    /// command was missing from exec config
84    #[error("command must be specified to use exec authentication plugin")]
85    MissingCommand,
86
87    /// OAuth error
88    #[cfg(feature = "oauth")]
89    #[cfg_attr(docsrs, doc(cfg(feature = "oauth")))]
90    #[error("failed OAuth: {0}")]
91    OAuth(#[source] OAuthError),
92
93    /// OIDC error
94    #[cfg(feature = "oidc")]
95    #[cfg_attr(docsrs, doc(cfg(feature = "oidc")))]
96    #[error("failed OIDC: {0}")]
97    Oidc(#[source] oidc_errors::Error),
98
99    /// cluster spec missing while `provideClusterInfo` is true
100    #[error("Cluster spec must be populated when `provideClusterInfo` is true")]
101    ExecMissingClusterInfo,
102
103    /// No valid native root CA certificates found
104    #[error("No valid native root CA certificates found: {0}")]
105    NoValidNativeRootCA(#[source] std::io::Error),
106}
107
108#[derive(Debug, Clone)]
109pub(crate) enum Auth {
110    None,
111    Basic(String, SecretString),
112    Bearer(SecretString),
113    RefreshableToken(RefreshableToken),
114    Certificate(String, SecretString, Option<Timestamp>),
115}
116
117// Token file reference. Reloads at least once per minute.
118#[derive(Debug)]
119pub struct TokenFile {
120    path: PathBuf,
121    token: SecretString,
122    expires_at: Timestamp,
123}
124
125impl TokenFile {
126    fn new<P: AsRef<Path>>(path: P) -> Result<TokenFile, Error> {
127        let token = std::fs::read_to_string(&path)
128            .map_err(|source| Error::ReadTokenFile(source, path.as_ref().to_owned()))?;
129        Ok(Self {
130            path: path.as_ref().to_owned(),
131            token: SecretString::from(token),
132            // Try to reload at least once a minute
133            expires_at: Timestamp::now() + SIXTY_SEC,
134        })
135    }
136
137    fn is_expiring(&self) -> bool {
138        Timestamp::now() + TEN_SEC > self.expires_at
139    }
140
141    /// Get the cached token. Returns `None` if it's expiring.
142    fn cached_token(&self) -> Option<&str> {
143        (!self.is_expiring()).then(|| self.token.expose_secret())
144    }
145
146    /// Get a token. Reloads from file if the cached token is expiring.
147    fn token(&mut self) -> &str {
148        if self.is_expiring() {
149            // > If reload from file fails, the last-read token should be used to avoid breaking
150            // > clients that make token files available on process start and then remove them to
151            // > limit credential exposure.
152            // > https://github.com/kubernetes/kubernetes/issues/68164
153            if let Ok(token) = std::fs::read_to_string(&self.path) {
154                self.token = SecretString::from(token);
155            }
156            self.expires_at = Timestamp::now() + SIXTY_SEC;
157        }
158        self.token.expose_secret()
159    }
160}
161
162/// Common constant for checking if an auth token is close to expiring
163pub const TEN_SEC: SignedDuration = SignedDuration::from_secs(10);
164/// Common duration for time between reloads
165const SIXTY_SEC: SignedDuration = SignedDuration::from_secs(60);
166
167// See https://github.com/kubernetes/kubernetes/tree/master/staging/src/k8s.io/client-go/plugin/pkg/client/auth
168// for the list of auth-plugins supported by client-go.
169// We currently support the following:
170// - exec
171// - token-file refreshed at least once per minute
172// - gcp: command based token source (exec)
173// - gcp: application credential based token source (requires `oauth` feature)
174//
175// Note that the visibility must be `pub` for `impl Layer for AuthLayer`, but this is not exported from the crate.
176// It's not accessible from outside and not shown on docs.
177#[derive(Debug, Clone)]
178pub enum RefreshableToken {
179    Exec(Arc<Mutex<(SecretString, Timestamp, AuthInfo)>>),
180    File(Arc<RwLock<TokenFile>>),
181    #[cfg(feature = "oauth")]
182    GcpOauth(Arc<Mutex<oauth::Gcp>>),
183    #[cfg(feature = "oidc")]
184    Oidc(Arc<Mutex<oidc::Oidc>>),
185}
186
187// For use with `AsyncFilterLayer` to add `Authorization` header with a refreshed token.
188impl<B> AsyncPredicate<Request<B>> for RefreshableToken
189where
190    B: http_body::Body + Send + 'static,
191{
192    type Future = BoxFuture<'static, Result<Request<B>, BoxError>>;
193    type Request = Request<B>;
194
195    fn check(&mut self, mut request: Self::Request) -> Self::Future {
196        let refreshable = self.clone();
197        Box::pin(async move {
198            refreshable.to_header().await.map_err(Into::into).map(|value| {
199                request.headers_mut().insert(AUTHORIZATION, value);
200                request
201            })
202        })
203    }
204}
205
206impl RefreshableToken {
207    async fn to_header(&self) -> Result<HeaderValue, Error> {
208        match self {
209            RefreshableToken::Exec(data) => {
210                let mut locked_data = data.lock().await;
211                // Add some wiggle room onto the current timestamp so we don't get any race
212                // conditions where the token expires while we are refreshing
213                if Timestamp::now() + SIXTY_SEC >= locked_data.1 {
214                    // Run blocking exec command on the blocking threadpool to avoid
215                    // stalling the tokio worker during token refresh.
216                    // TODO Improve refreshing exec to avoid `Auth::try_from`
217                    let auth_info = locked_data.2.clone();
218                    let auth = tokio::task::spawn_blocking(move || Auth::try_from(&auth_info))
219                        .await
220                        .map_err(|e| Error::AuthExec(format!("failed to spawn blocking auth task: {e}")))??;
221                    match auth {
222                        Auth::None | Auth::Basic(_, _) | Auth::Bearer(_) | Auth::Certificate(_, _, _) => {
223                            return Err(Error::UnrefreshableTokenResponse);
224                        }
225
226                        Auth::RefreshableToken(RefreshableToken::Exec(d)) => {
227                            let (new_token, new_expire, new_info) = Arc::try_unwrap(d)
228                                .expect("Unable to unwrap Arc, this is likely a programming error")
229                                .into_inner();
230                            locked_data.0 = new_token;
231                            locked_data.1 = new_expire;
232                            locked_data.2 = new_info;
233                        }
234
235                        // Unreachable because the token source does not change
236                        Auth::RefreshableToken(RefreshableToken::File(_)) => unreachable!(),
237                        #[cfg(feature = "oauth")]
238                        Auth::RefreshableToken(RefreshableToken::GcpOauth(_)) => unreachable!(),
239                        #[cfg(feature = "oidc")]
240                        Auth::RefreshableToken(RefreshableToken::Oidc(_)) => unreachable!(),
241                    }
242                }
243
244                bearer_header(locked_data.0.expose_secret())
245            }
246
247            RefreshableToken::File(token_file) => {
248                let guard = token_file.read().await;
249                if let Some(header) = guard.cached_token().map(bearer_header) {
250                    return header;
251                }
252                // Drop the read guard before a write lock attempt to prevent deadlock.
253                drop(guard);
254                // Note that `token()` only reloads if the cached token is expiring.
255                // A separate method to conditionally reload minimizes the need for an exclusive access.
256                bearer_header(token_file.write().await.token())
257            }
258
259            #[cfg(feature = "oauth")]
260            RefreshableToken::GcpOauth(data) => {
261                let gcp_oauth = data.lock().await;
262                let token = (*gcp_oauth).token().await.map_err(Error::OAuth)?;
263                bearer_header(&token.access_token)
264            }
265
266            #[cfg(feature = "oidc")]
267            RefreshableToken::Oidc(oidc) => {
268                let token = oidc.lock().await.id_token().await.map_err(Error::Oidc)?;
269                bearer_header(&token)
270            }
271        }
272    }
273}
274
275fn bearer_header(token: &str) -> Result<HeaderValue, Error> {
276    let mut value = HeaderValue::try_from(format!("Bearer {token}")).map_err(Error::InvalidBearerToken)?;
277    value.set_sensitive(true);
278    Ok(value)
279}
280
281impl TryFrom<&AuthInfo> for Auth {
282    type Error = Error;
283
284    /// Loads the authentication header from the credentials available in the kubeconfig. This supports
285    /// exec plugins as well as specified in
286    /// https://kubernetes.io/docs/reference/access-authn-authz/authentication/#client-go-credential-plugins
287    fn try_from(auth_info: &AuthInfo) -> Result<Self, Self::Error> {
288        if let Some(provider) = &auth_info.auth_provider {
289            match token_from_provider(provider)? {
290                #[cfg(feature = "oidc")]
291                ProviderToken::Oidc(oidc) => {
292                    return Ok(Self::RefreshableToken(RefreshableToken::Oidc(Arc::new(
293                        Mutex::new(*oidc),
294                    ))));
295                }
296
297                #[cfg(not(feature = "oidc"))]
298                ProviderToken::Oidc(token) => {
299                    return Ok(Self::Bearer(SecretString::from(token)));
300                }
301
302                ProviderToken::GcpCommand(token, Some(expiry)) => {
303                    let mut info = auth_info.clone();
304                    let mut provider = provider.clone();
305                    provider.config.insert("access-token".into(), token.clone());
306                    // `jiff::Timestamp` provides RFC3339 via `Display`, docs: https://docs.rs/jiff/latest/jiff/struct.Timestamp.html#impl-Display-for-Timestamp
307                    provider.config.insert("expiry".into(), expiry.to_string());
308                    info.auth_provider = Some(provider);
309                    return Ok(Self::RefreshableToken(RefreshableToken::Exec(Arc::new(
310                        Mutex::new((SecretString::from(token), expiry, info)),
311                    ))));
312                }
313
314                ProviderToken::GcpCommand(token, None) => {
315                    return Ok(Self::Bearer(SecretString::from(token)));
316                }
317
318                #[cfg(feature = "oauth")]
319                ProviderToken::GcpOauth(gcp) => {
320                    return Ok(Self::RefreshableToken(RefreshableToken::GcpOauth(Arc::new(
321                        Mutex::new(gcp),
322                    ))));
323                }
324            }
325        }
326
327        if let (Some(u), Some(p)) = (&auth_info.username, &auth_info.password) {
328            return Ok(Self::Basic(u.to_owned(), p.to_owned()));
329        }
330
331        // Inline token. Has precedence over `token_file`.
332        if let Some(token) = &auth_info.token {
333            return Ok(Self::Bearer(token.clone()));
334        }
335
336        // Token file reference. Must be reloaded at least once a minute.
337        if let Some(file) = &auth_info.token_file {
338            return Ok(Self::RefreshableToken(RefreshableToken::File(Arc::new(
339                RwLock::new(TokenFile::new(file)?),
340            ))));
341        }
342
343        if let Some(exec) = &auth_info.exec {
344            let creds = auth_exec(exec)?;
345            let status = creds.status.ok_or(Error::ExecPluginFailed)?;
346            let expiration = status
347                .expiration_timestamp
348                .map(|ts| ts.parse())
349                .transpose()
350                .map_err(Error::MalformedTokenExpirationDate)?;
351
352            if let (Some(client_certificate_data), Some(client_key_data)) =
353                (status.client_certificate_data, status.client_key_data)
354            {
355                return Ok(Self::Certificate(
356                    client_certificate_data,
357                    client_key_data.into(),
358                    expiration,
359                ));
360            }
361
362            match (status.token.map(SecretString::from), expiration) {
363                (Some(token), Some(expire)) => Ok(Self::RefreshableToken(RefreshableToken::Exec(Arc::new(
364                    Mutex::new((token, expire, auth_info.clone())),
365                )))),
366                (Some(token), None) => Ok(Self::Bearer(token)),
367                _ => Ok(Self::None),
368            }
369        } else {
370            Ok(Self::None)
371        }
372    }
373}
374
375// We need to differentiate providers because the keys/formats to store token expiration differs.
376enum ProviderToken {
377    #[cfg(feature = "oidc")]
378    Oidc(Box<oidc::Oidc>), // boxed: largest variant, keeps the enum small (clippy::large_enum_variant)
379    #[cfg(not(feature = "oidc"))]
380    Oidc(String),
381    // "access-token", "expiry" (RFC3339)
382    GcpCommand(String, Option<Timestamp>),
383    #[cfg(feature = "oauth")]
384    GcpOauth(oauth::Gcp),
385    // "access-token", "expires-on" (timestamp)
386    // Azure(String, Option<DateTime<Utc>>),
387}
388
389fn token_from_provider(provider: &AuthProviderConfig) -> Result<ProviderToken, Error> {
390    match provider.name.as_ref() {
391        "oidc" => token_from_oidc_provider(provider),
392        "gcp" => token_from_gcp_provider(provider),
393        "azure" => Err(Error::AuthExec(
394            "The azure auth plugin is not supported; use https://github.com/Azure/kubelogin instead".into(),
395        )),
396        _ => Err(Error::AuthExec(format!(
397            "Authentication with provider {:} not supported",
398            provider.name
399        ))),
400    }
401}
402
403#[cfg(feature = "oidc")]
404fn token_from_oidc_provider(provider: &AuthProviderConfig) -> Result<ProviderToken, Error> {
405    oidc::Oidc::from_config(&provider.config)
406        .map_err(Error::Oidc)
407        .map(|t| ProviderToken::Oidc(Box::new(t)))
408}
409
410#[cfg(not(feature = "oidc"))]
411fn token_from_oidc_provider(provider: &AuthProviderConfig) -> Result<ProviderToken, Error> {
412    match provider.config.get("id-token") {
413        Some(id_token) => Ok(ProviderToken::Oidc(id_token.clone())),
414        None => Err(Error::AuthExec(
415            "No id-token for oidc Authentication provider".into(),
416        )),
417    }
418}
419
420fn token_from_gcp_provider(provider: &AuthProviderConfig) -> Result<ProviderToken, Error> {
421    if let Some(id_token) = provider.config.get("id-token") {
422        return Ok(ProviderToken::GcpCommand(id_token.clone(), None));
423    }
424
425    // Return cached access token if it's still valid
426    if let Some(access_token) = provider.config.get("access-token")
427        && let Some(expiry) = provider.config.get("expiry")
428    {
429        let expiry_date = expiry
430            .parse::<Timestamp>()
431            .map_err(Error::MalformedTokenExpirationDate)?;
432        if Timestamp::now() + SIXTY_SEC < expiry_date {
433            return Ok(ProviderToken::GcpCommand(access_token.clone(), Some(expiry_date)));
434        }
435    }
436
437    // Command-based token source
438    if let Some(cmd) = provider.config.get("cmd-path") {
439        let params = provider.config.get("cmd-args").cloned().unwrap_or_default();
440        // NB: This property does currently not exist upstream in client-go
441        // See https://github.com/kube-rs/kube/issues/1060
442        let drop_env = provider.config.get("cmd-drop-env").cloned().unwrap_or_default();
443        // TODO splitting args by space is not safe
444        let mut command = Command::new(cmd);
445        // Do not pass the following env vars to the command
446        for env in drop_env.trim().split(' ') {
447            command.env_remove(env);
448        }
449        let output = command
450            .args(params.trim().split(' '))
451            .output()
452            .map_err(|e| Error::AuthExec(format!("Executing {cmd:} failed: {e:?}")))?;
453
454        if !output.status.success() {
455            return Err(Error::AuthExecRun {
456                cmd: format!("{cmd} {params}"),
457                out: Box::new(output),
458            });
459        }
460
461        if let Some(field) = provider.config.get("token-key") {
462            let json_output: serde_json::Value =
463                serde_json::from_slice(&output.stdout).map_err(Error::ParseTokenKey)?;
464            let token = extract_value(&json_output, "token-key", field)?;
465            if let Some(field) = provider.config.get("expiry-key") {
466                let expiry = extract_value(&json_output, "expiry-key", field)?;
467                let expiry = expiry
468                    .parse::<Timestamp>()
469                    .map_err(Error::MalformedTokenExpirationDate)?;
470                return Ok(ProviderToken::GcpCommand(token, Some(expiry)));
471            } else {
472                return Ok(ProviderToken::GcpCommand(token, None));
473            }
474        } else {
475            let token = std::str::from_utf8(&output.stdout)
476                .map_err(|e| Error::AuthExec(format!("Result is not a string {e:?} ")))?
477                .to_owned();
478            return Ok(ProviderToken::GcpCommand(token, None));
479        }
480    }
481
482    // Google Application Credentials-based token source
483    #[cfg(feature = "oauth")]
484    {
485        Ok(ProviderToken::GcpOauth(
486            oauth::Gcp::default_credentials_with_scopes(provider.config.get("scopes"))
487                .map_err(Error::OAuth)?,
488        ))
489    }
490    #[cfg(not(feature = "oauth"))]
491    {
492        Err(Error::AuthExec(
493            "Enable oauth feature to use Google Application Credentials-based token source".into(),
494        ))
495    }
496}
497
498fn extract_value(json: &serde_json::Value, context: &str, path: &str) -> Result<String, Error> {
499    let path = {
500        let p = path.trim_matches(|c| c == '"' || c == '{' || c == '}');
501        if p.starts_with('$') {
502            p
503        } else if p.starts_with('.') {
504            &format!("${p}")
505        } else {
506            &format!("$.{p}")
507        }
508    };
509
510    let res = json.query(path).map_err(|err| {
511        Error::AuthExec(format!(
512            "Failed to query {context:?} as a JsonPath: {path}\n
513             Error: {err}"
514        ))
515    })?;
516
517    let Some(jval) = res.into_iter().next() else {
518        return Err(Error::AuthExec(format!(
519            "Target {context:?} value {path:?} not found"
520        )));
521    };
522
523    let val = jval.as_str().ok_or(Error::AuthExec(format!(
524        "Target {context:?} value {path:?} is not a string"
525    )))?;
526
527    Ok(val.to_string())
528}
529
530/// ExecCredentials is used by exec-based plugins to communicate credentials to
531/// HTTP transports.
532#[derive(Clone, Debug, Serialize, Deserialize)]
533pub struct ExecCredential {
534    pub kind: Option<String>,
535    #[serde(rename = "apiVersion")]
536    pub api_version: Option<String>,
537    pub spec: Option<ExecCredentialSpec>,
538    #[serde(skip_serializing_if = "Option::is_none")]
539    pub status: Option<ExecCredentialStatus>,
540}
541
542/// ExecCredenitalSpec holds request and runtime specific information provided
543/// by transport.
544#[derive(Clone, Debug, Serialize, Deserialize)]
545pub struct ExecCredentialSpec {
546    #[serde(skip_serializing_if = "Option::is_none")]
547    interactive: Option<bool>,
548
549    #[serde(skip_serializing_if = "Option::is_none")]
550    cluster: Option<ExecAuthCluster>,
551}
552
553/// ExecCredentialStatus holds credentials for the transport to use.
554#[derive(Clone, Debug, Serialize, Deserialize)]
555pub struct ExecCredentialStatus {
556    #[serde(rename = "expirationTimestamp")]
557    pub expiration_timestamp: Option<String>,
558    pub token: Option<String>,
559    #[serde(rename = "clientCertificateData")]
560    pub client_certificate_data: Option<String>,
561    #[serde(rename = "clientKeyData")]
562    pub client_key_data: Option<String>,
563}
564
565fn auth_exec(auth: &ExecConfig) -> Result<ExecCredential, Error> {
566    let mut cmd = match &auth.command {
567        Some(cmd) => Command::new(cmd),
568        None => return Err(Error::MissingCommand),
569    };
570
571    if let Some(args) = &auth.args {
572        cmd.args(args);
573    }
574    if let Some(env) = &auth.env {
575        let envs = env
576            .iter()
577            .flat_map(|env| match (env.get("name"), env.get("value")) {
578                (Some(name), Some(value)) => Some((name, value)),
579                _ => None,
580            });
581        cmd.envs(envs);
582    }
583
584    let interactive = auth.interactive_mode != Some(ExecInteractiveMode::Never);
585    if interactive {
586        cmd.stdin(std::process::Stdio::inherit());
587        cmd.stderr(std::process::Stdio::inherit());
588    } else {
589        cmd.stdin(std::process::Stdio::piped());
590    }
591
592    let mut exec_credential_spec = ExecCredentialSpec {
593        interactive: Some(interactive),
594        cluster: None,
595    };
596
597    if auth.provide_cluster_info {
598        exec_credential_spec.cluster = Some(auth.cluster.clone().ok_or(Error::ExecMissingClusterInfo)?);
599    }
600
601    // Provide exec info to child process
602    let exec_info = serde_json::to_string(&ExecCredential {
603        api_version: auth.api_version.clone(),
604        kind: "ExecCredential".to_string().into(),
605        spec: Some(exec_credential_spec),
606        status: None,
607    })
608    .map_err(Error::AuthExecSerialize)?;
609    cmd.env("KUBERNETES_EXEC_INFO", exec_info);
610
611    if let Some(envs) = &auth.drop_env {
612        for env in envs {
613            cmd.env_remove(env);
614        }
615    }
616
617    #[cfg(target_os = "windows")]
618    {
619        // Opt-in via env var; see https://github.com/kube-rs/kube/issues/1901 for why
620        // this is gated (CREATE_NO_WINDOW breaks stderr inheritance for interactive exec
621        // plugins like kubelogin.exe).
622        if std::env::var("KUBE_RS_UNSTABLE_CREATE_NO_WINDOW")
623            .map(|s| s == "1")
624            .unwrap_or(false)
625        {
626            const CREATE_NO_WINDOW: u32 = 0x08000000;
627            cmd.creation_flags(CREATE_NO_WINDOW);
628        }
629    }
630
631    let out = cmd.output().map_err(Error::AuthExecStart)?;
632    if !out.status.success() {
633        return Err(Error::AuthExecRun {
634            cmd: format!("{cmd:?}"),
635            out: Box::new(out),
636        });
637    }
638    parse_exec_credentials(&out.stdout)
639}
640
641/// Exec plugins are expected to emit JSON, but client-go decodes their output with a
642/// YAML-tolerant codec, so some plugins emit YAML. `serde_saphyr` handles both because
643/// YAML 1.2 is a superset of JSON.
644fn parse_exec_credentials(stdout: &[u8]) -> Result<ExecCredential, Error> {
645    serde_saphyr::from_slice(stdout).map_err(|e| Error::AuthExecParse(Box::new(e)))
646}
647
648#[cfg(test)]
649mod test {
650    use crate::config::Kubeconfig;
651    use std::time::{Duration, Instant};
652
653    use super::*;
654
655    /// Build an `AuthInfo` whose gcp auth-provider runs `cmd_path cmd_args` to emit credential JSON.
656    fn gcp_auth_info(cmd_path: &str, cmd_args: &str) -> AuthInfo {
657        let test_file = format!(
658            r#"
659        apiVersion: v1
660        clusters:
661        - cluster:
662            certificate-authority-data: XXXXXXX
663            server: https://36.XXX.XXX.XX
664          name: generic-name
665        contexts:
666        - context:
667            cluster: generic-name
668            user: generic-name
669          name: generic-name
670        current-context: generic-name
671        kind: Config
672        preferences: {{}}
673        users:
674        - name: generic-name
675          user:
676            auth-provider:
677              config:
678                cmd-args: {cmd_args}
679                cmd-path: {cmd_path}
680                expiry-key: '{{.credential.token_expiry}}'
681                token-key: '{{.credential.access_token}}'
682              name: gcp
683        "#
684        );
685        let config: Kubeconfig = serde_saphyr::from_str(&test_file).unwrap();
686        config.auth_infos[0].auth_info.clone().unwrap()
687    }
688
689    fn cred_json(token: &str, expiry: &str) -> String {
690        format!(
691            r#"{{"something": "else", "credential": {{"access_token": "{token}", "token_expiry": "{expiry}"}}}}"#
692        )
693    }
694
695    #[tokio::test]
696    #[ignore = "fails on windows mysteriously"]
697    async fn exec_auth_command() -> Result<(), Error> {
698        let expiry = (Timestamp::now() + SIXTY_SEC).to_string();
699        let auth_info = gcp_auth_info("echo", &format!("'{}'", cred_json("my_token", &expiry)));
700        match Auth::try_from(&auth_info).unwrap() {
701            Auth::RefreshableToken(RefreshableToken::Exec(refreshable)) => {
702                let (token, _expire, info) = Arc::try_unwrap(refreshable).unwrap().into_inner();
703                assert_eq!(token.expose_secret(), &"my_token".to_owned());
704                let config = info.auth_provider.unwrap().config;
705                assert_eq!(config.get("access-token"), Some(&"my_token".to_owned()));
706            }
707            _ => unreachable!(),
708        }
709        Ok(())
710    }
711
712    #[tokio::test]
713    #[ignore = "shells out to echo/sh; skipped on windows"]
714    async fn exec_token_refresh_via_to_header() -> Result<(), Error> {
715        let fresh_expiry = (Timestamp::now() + SignedDuration::from_secs(3600)).to_string();
716        let auth_info = gcp_auth_info("echo", &format!("'{}'", cred_json("my_token", &fresh_expiry)));
717        // Seed with a stale token + past expiry to force the refresh branch.
718        let stale_expiry = Timestamp::now() - SIXTY_SEC;
719        let refreshable = RefreshableToken::Exec(Arc::new(Mutex::new((
720            SecretString::from("stale"),
721            stale_expiry,
722            auth_info,
723        ))));
724
725        let header = refreshable.to_header().await?;
726        assert_eq!(header, HeaderValue::from_static("Bearer my_token"));
727
728        // Cached state should have been updated in place.
729        if let RefreshableToken::Exec(data) = &refreshable {
730            let locked = data.lock().await;
731            assert_eq!(locked.0.expose_secret(), "my_token");
732            assert!(locked.1 > Timestamp::now(), "expiry should be in the future");
733        } else {
734            unreachable!();
735        }
736        Ok(())
737    }
738
739    #[tokio::test(flavor = "current_thread")]
740    #[ignore = "shells out to echo/sh; skipped on windows"]
741    async fn exec_token_refresh_does_not_block_runtime() -> Result<(), Error> {
742        use std::io::Write;
743        let fresh_expiry = (Timestamp::now() + SignedDuration::from_secs(3600)).to_string();
744        // Write a script that sleeps ~300ms before emitting credentials; the gcp provider
745        // splits cmd-args on spaces so we can't inline this via `sh -c`.
746        let mut script = tempfile::NamedTempFile::new().unwrap();
747        writeln!(script, "#!/bin/sh").unwrap();
748        writeln!(script, "sleep 0.3").unwrap();
749        writeln!(script, "echo '{}'", cred_json("my_token", &fresh_expiry)).unwrap();
750        script.flush().unwrap();
751        let script_path = script.path().to_str().unwrap().to_owned();
752
753        let auth_info = gcp_auth_info("sh", &script_path);
754        let stale_expiry = Timestamp::now() - SIXTY_SEC;
755        let refreshable = RefreshableToken::Exec(Arc::new(Mutex::new((
756            SecretString::from("stale"),
757            stale_expiry,
758            auth_info,
759        ))));
760
761        // On a current_thread runtime, if to_header() blocked the worker the 50ms
762        // timer below could not fire until the 300ms exec completed.
763        let start = Instant::now();
764        let (header, timer_elapsed) = tokio::join!(refreshable.to_header(), async {
765            tokio::time::sleep(Duration::from_millis(50)).await;
766            start.elapsed()
767        });
768
769        assert!(
770            timer_elapsed < Duration::from_millis(200),
771            "timer took {timer_elapsed:?}; to_header likely blocked the runtime"
772        );
773        assert_eq!(header?, HeaderValue::from_static("Bearer my_token"));
774        Ok(())
775    }
776
777    #[test]
778    fn exec_credentials_json_and_yaml_parse() {
779        let json = br#"{"kind": "ExecCredential", "apiVersion": "client.authentication.k8s.io/v1", "status": {"token": "json_token"}}"#;
780        let creds = parse_exec_credentials(json).unwrap();
781        assert_eq!(creds.status.unwrap().token.unwrap(), "json_token");
782
783        let yaml = br#"
784kind: ExecCredential
785apiVersion: client.authentication.k8s.io/v1
786status:
787  token: yaml_token
788  expirationTimestamp: "2030-01-01T00:00:00Z"
789"#;
790        let creds = parse_exec_credentials(yaml).unwrap();
791        let status = creds.status.unwrap();
792        assert_eq!(status.token.unwrap(), "yaml_token");
793        assert_eq!(status.expiration_timestamp.unwrap(), "2030-01-01T00:00:00Z");
794
795        let invalid = b"not: [valid";
796        assert!(matches!(
797            parse_exec_credentials(invalid),
798            Err(Error::AuthExecParse(_))
799        ));
800    }
801
802    #[test]
803    fn token_file() {
804        let file = tempfile::NamedTempFile::new().unwrap();
805        std::fs::write(file.path(), "token1").unwrap();
806        let mut token_file = TokenFile::new(file.path()).unwrap();
807        assert_eq!(token_file.cached_token().unwrap(), "token1");
808        assert!(!token_file.is_expiring());
809        assert_eq!(token_file.token(), "token1");
810        // Doesn't reload unless expiring
811        std::fs::write(file.path(), "token2").unwrap();
812        assert_eq!(token_file.token(), "token1");
813
814        token_file.expires_at = Timestamp::now();
815        assert!(token_file.is_expiring());
816        assert_eq!(token_file.cached_token(), None);
817        assert_eq!(token_file.token(), "token2");
818        assert!(!token_file.is_expiring());
819        assert_eq!(token_file.cached_token().unwrap(), "token2");
820    }
821}
kube_client/client/auth/mod.rs

kube_client/client/auth/
mod.rs
