From 9d96fb210c0d20f8c5860d50e14ce8f106787708 Mon Sep 17 00:00:00 2001 From: Md Nahid Hassan Bhuiyan Date: Tue, 4 Aug 2026 03:44:56 +0600 Subject: [PATCH 1/2] aws: copy cached credentials under a lock Every credential provider caches one flb_aws_credentials and hands out copies of it. get_credentials() made that copy without holding the provider lock, while the refresh path freed and replaced the cached struct under it. That was fine when flushes were coroutines on a single thread, but with `workers N` they run on N real pthreads, so the copy and the swap can overlap. Three ways it goes wrong: a reader dereferences a pointer the refresh already freed and takes a SIGSEGV in flb_sds_create(), a reader lands between `*creds = NULL` and the reassignment and bails out with no credentials, or a reader picks up the access key from one refresh and the secret from the next and gets InvalidSignatureException. All five providers (sts, eks, ec2, http, profile) had the same shape. The provider lock can't just be taken by readers: it is held across the network call that fetches new credentials, which yields the coroutine, so a second coroutine on the same thread blocking on it would deadlock. Add a separate cache lock that is only ever held for the copy and for the pointer swap, never across IO, and route both sides through flb_aws_cache_get_credentials() and flb_aws_cache_set_credentials(). The swap now publishes the new credentials first and frees the old ones afterwards, so there is no window where the cache is empty. next_refresh is read and written under the same lock since it is part of the same cached state. Fixes #12206 Signed-off-by: Md Nahid Hassan Bhuiyan --- include/fluent-bit/flb_aws_credentials.h | 51 ++++++++- src/aws/flb_aws_credentials.c | 104 +++++++++++++++++ src/aws/flb_aws_credentials_ec2.c | 86 ++++++-------- src/aws/flb_aws_credentials_http.c | 75 ++++-------- src/aws/flb_aws_credentials_profile.c | 77 +++++-------- src/aws/flb_aws_credentials_sts.c | 139 ++++++++--------------- 6 files changed, 284 insertions(+), 248 deletions(-) diff --git a/include/fluent-bit/flb_aws_credentials.h b/include/fluent-bit/flb_aws_credentials.h index d7e9eaacc0c..7512324adee 100644 --- a/include/fluent-bit/flb_aws_credentials.h +++ b/include/fluent-bit/flb_aws_credentials.h @@ -126,6 +126,18 @@ struct flb_aws_provider { */ pthread_mutex_t lock; + /* + * Protects the credentials cached by the provider implementation. + * + * The lock above is a trylock because it is held across the network call + * that fetches new credentials; a reader cannot take it without risking a + * deadlock between two coroutines running on the same thread. This lock is + * only ever held for the pointer swap that publishes new credentials and + * for the copy that readers make, so it never spans a yield point and can + * be taken with a blocking lock. + */ + pthread_mutex_t cache_lock; + struct flb_aws_provider_vtable *provider_vtable; void *implementation; @@ -338,12 +350,17 @@ int exec_credential_process(char* process, struct flb_aws_credentials** creds, #endif /* FLB_HAVE_AWS_CREDENTIAL_PROCESS */ /* - * Fluent Bit is single-threaded but asynchonous. Only one co-routine will - * be running at a time, and they only pause/resume for IO. + * A provider is shared by every flush thread of an output with `workers` set, + * so the cached credentials are read and replaced concurrently by real threads. * - * Thus, while synchronization is needed (to prevent multiple co-routines - * from duplicating effort and performing the same work), it can be obtained - * using a simple integer flag on the provider. + * The provider lock deduplicates refreshes: only the coroutine that wins it + * calls out to the credentials endpoint. It is a trylock because it is held + * across that network call, which yields the coroutine. + * + * The cache lock protects the cached credentials themselves. Use the + * flb_aws_cache_* helpers below instead of touching the cached pointer + * directly; they hold the cache lock for the copy and for the swap, so a + * reader can never end up with a pointer that the refresh path has freed. */ /* Like a traditional try lock- it does not block if the lock is not obtained */ @@ -351,6 +368,30 @@ int try_lock_provider(struct flb_aws_provider *provider); void unlock_provider(struct flb_aws_provider *provider); +/* + * Returns a copy of the credentials in *cache, or NULL if the cache is empty + * or the copy could not be allocated. The caller owns the returned copy. + */ +struct flb_aws_credentials *flb_aws_cache_get_credentials(struct flb_aws_provider + *provider, + struct flb_aws_credentials + **cache); + +/* Returns the refresh deadline stored in *next_refresh */ +time_t flb_aws_cache_get_refresh_time(struct flb_aws_provider *provider, + time_t *next_refresh); + +/* + * Publishes creds as the new contents of *cache and refresh_time as the new + * contents of *next_refresh. Takes ownership of creds and frees whatever was + * cached before. + */ +void flb_aws_cache_set_credentials(struct flb_aws_provider *provider, + struct flb_aws_credentials **cache, + struct flb_aws_credentials *creds, + time_t *next_refresh, + time_t refresh_time); + /* * HTTP Credentials Provider - retrieve credentials from a local http server diff --git a/src/aws/flb_aws_credentials.c b/src/aws/flb_aws_credentials.c index 37310676863..54a874be9c2 100644 --- a/src/aws/flb_aws_credentials.c +++ b/src/aws/flb_aws_credentials.c @@ -541,6 +541,7 @@ static struct flb_aws_provider *standard_chain_create(struct flb_config } pthread_mutex_init(&provider->lock, NULL); + pthread_mutex_init(&provider->cache_lock, NULL); implementation = flb_calloc(1, sizeof(struct flb_aws_provider_chain)); @@ -774,6 +775,7 @@ void flb_aws_provider_destroy(struct flb_aws_provider *provider) } pthread_mutex_destroy(&provider->lock); + pthread_mutex_destroy(&provider->cache_lock); /* free managed dependencies */ if (provider->base_aws_provider) { @@ -863,3 +865,105 @@ void unlock_provider(struct flb_aws_provider *provider) { pthread_mutex_unlock(&provider->lock); } + +/* + * The cache lock is never held across IO, so unlike the provider lock it is + * safe to block on it. + */ +static inline void lock_provider_cache(struct flb_aws_provider *provider) +{ + pthread_mutex_lock(&provider->cache_lock); +} + +static inline void unlock_provider_cache(struct flb_aws_provider *provider) +{ + pthread_mutex_unlock(&provider->cache_lock); +} + +struct flb_aws_credentials *flb_aws_cache_get_credentials(struct flb_aws_provider + *provider, + struct flb_aws_credentials + **cache) +{ + struct flb_aws_credentials *cached; + struct flb_aws_credentials *creds = NULL; + + lock_provider_cache(provider); + + cached = *cache; + if (!cached) { + unlock_provider_cache(provider); + return NULL; + } + + creds = flb_calloc(1, sizeof(struct flb_aws_credentials)); + if (!creds) { + goto error; + } + + creds->access_key_id = flb_sds_create(cached->access_key_id); + if (!creds->access_key_id) { + goto error; + } + + creds->secret_access_key = flb_sds_create(cached->secret_access_key); + if (!creds->secret_access_key) { + goto error; + } + + if (cached->session_token) { + creds->session_token = flb_sds_create(cached->session_token); + if (!creds->session_token) { + goto error; + } + } + else { + creds->session_token = NULL; + } + + unlock_provider_cache(provider); + return creds; + +error: + unlock_provider_cache(provider); + flb_errno(); + flb_aws_credentials_destroy(creds); + return NULL; +} + +time_t flb_aws_cache_get_refresh_time(struct flb_aws_provider *provider, + time_t *next_refresh) +{ + time_t refresh_time; + + lock_provider_cache(provider); + refresh_time = *next_refresh; + unlock_provider_cache(provider); + + return refresh_time; +} + +void flb_aws_cache_set_credentials(struct flb_aws_provider *provider, + struct flb_aws_credentials **cache, + struct flb_aws_credentials *creds, + time_t *next_refresh, + time_t refresh_time) +{ + struct flb_aws_credentials *previous; + + lock_provider_cache(provider); + + previous = *cache; + *cache = creds; + *next_refresh = refresh_time; + + unlock_provider_cache(provider); + + /* + * Free the old credentials after the swap and outside of the lock. + * Readers hold the cache lock for the whole copy, so none of them was + * halfway through reading these when we swapped them out, and anyone + * arriving now sees the new ones. + */ + flb_aws_credentials_destroy(previous); +} diff --git a/src/aws/flb_aws_credentials_ec2.c b/src/aws/flb_aws_credentials_ec2.c index d3cc2a4afaa..8744b13b167 100644 --- a/src/aws/flb_aws_credentials_ec2.c +++ b/src/aws/flb_aws_credentials_ec2.c @@ -34,9 +34,11 @@ #define AWS_IMDS_ROLE_PATH_LEN 43 struct flb_aws_provider_ec2; -static int get_creds_ec2(struct flb_aws_provider_ec2 *implementation); -static int ec2_credentials_request(struct flb_aws_provider_ec2 - *implementation, char *cred_path); +static int get_creds_ec2(struct flb_aws_provider *provider, + struct flb_aws_provider_ec2 *implementation); +static int ec2_credentials_request(struct flb_aws_provider *provider, + struct flb_aws_provider_ec2 *implementation, + char *cred_path); /* EC2 IMDS Provider */ @@ -58,25 +60,34 @@ struct flb_aws_credentials *get_credentials_fn_ec2(struct flb_aws_provider *provider) { struct flb_aws_credentials *creds; + time_t next_refresh; int refresh = FLB_FALSE; struct flb_aws_provider_ec2 *implementation = provider->implementation; flb_debug("[aws_credentials] Requesting credentials from the " "EC2 provider.."); + next_refresh = flb_aws_cache_get_refresh_time(provider, + &implementation->next_refresh); + /* a negative next_refresh means that auto-refresh is disabled */ - if (implementation->next_refresh > 0 - && time(NULL) > implementation->next_refresh) { + if (next_refresh > 0 && time(NULL) > next_refresh) { refresh = FLB_TRUE; } - if (!implementation->creds || refresh == FLB_TRUE) { + + creds = flb_aws_cache_get_credentials(provider, &implementation->creds); + if (!creds || refresh == FLB_TRUE) { if (try_lock_provider(provider)) { - get_creds_ec2(implementation); + get_creds_ec2(provider, implementation); unlock_provider(provider); + + flb_aws_credentials_destroy(creds); + creds = flb_aws_cache_get_credentials(provider, + &implementation->creds); } } - if (!implementation->creds) { + if (!creds) { /* * We failed to lock the provider and creds are unset. This means that * another co-routine is performing the refresh. @@ -88,40 +99,6 @@ struct flb_aws_credentials *get_credentials_fn_ec2(struct flb_aws_provider return NULL; } - creds = flb_calloc(1, sizeof(struct flb_aws_credentials)); - if (!creds) { - flb_errno(); - return NULL; - } - - creds->access_key_id = flb_sds_create(implementation->creds->access_key_id); - if (!creds->access_key_id) { - flb_errno(); - flb_aws_credentials_destroy(creds); - return NULL; - } - - creds->secret_access_key = flb_sds_create(implementation->creds-> - secret_access_key); - if (!creds->secret_access_key) { - flb_errno(); - flb_aws_credentials_destroy(creds); - return NULL; - } - - if (implementation->creds->session_token) { - creds->session_token = flb_sds_create(implementation->creds-> - session_token); - if (!creds->session_token) { - flb_errno(); - flb_aws_credentials_destroy(creds); - return NULL; - } - - } else { - creds->session_token = NULL; - } - return creds; } @@ -131,7 +108,7 @@ int refresh_fn_ec2(struct flb_aws_provider *provider) { flb_debug("[aws_credentials] Refresh called on the EC2 IMDS provider"); if (try_lock_provider(provider)) { - ret = get_creds_ec2(implementation); + ret = get_creds_ec2(provider, implementation); unlock_provider(provider); } return ret; @@ -145,7 +122,7 @@ int init_fn_ec2(struct flb_aws_provider *provider) { flb_debug("[aws_credentials] Init called on the EC2 IMDS provider"); if (try_lock_provider(provider)) { - ret = get_creds_ec2(implementation); + ret = get_creds_ec2(provider, implementation); unlock_provider(provider); } @@ -241,6 +218,7 @@ struct flb_aws_provider *flb_ec2_provider_create(struct flb_config *config, } pthread_mutex_init(&provider->lock, NULL); + pthread_mutex_init(&provider->cache_lock, NULL); implementation = flb_calloc(1, sizeof(struct flb_aws_provider_ec2)); @@ -296,7 +274,8 @@ struct flb_aws_provider *flb_ec2_provider_create(struct flb_config *config, } /* Requests creds from IMDSv1 and sets them on the provider */ -static int get_creds_ec2(struct flb_aws_provider_ec2 *implementation) +static int get_creds_ec2(struct flb_aws_provider *provider, + struct flb_aws_provider_ec2 *implementation) { int ret; flb_sds_t instance_role; @@ -337,7 +316,7 @@ static int get_creds_ec2(struct flb_aws_provider_ec2 *implementation) } /* request creds */ - ret = ec2_credentials_request(implementation, cred_path); + ret = ec2_credentials_request(provider, implementation, cred_path); flb_sds_destroy(instance_role); flb_free(cred_path); @@ -345,8 +324,9 @@ static int get_creds_ec2(struct flb_aws_provider_ec2 *implementation) } -static int ec2_credentials_request(struct flb_aws_provider_ec2 - *implementation, char *cred_path) +static int ec2_credentials_request(struct flb_aws_provider *provider, + struct flb_aws_provider_ec2 *implementation, + char *cred_path) { int ret; flb_sds_t credentials_response; @@ -370,12 +350,10 @@ static int ec2_credentials_request(struct flb_aws_provider_ec2 return -1; } - /* destroy existing credentials first */ - flb_aws_credentials_destroy(implementation->creds); - implementation->creds = NULL; - /* set new creds */ - implementation->creds = creds; - implementation->next_refresh = expiration - FLB_AWS_REFRESH_WINDOW; + /* publish the new credentials; the old ones are freed for us */ + flb_aws_cache_set_credentials(provider, &implementation->creds, creds, + &implementation->next_refresh, + expiration - FLB_AWS_REFRESH_WINDOW); flb_sds_destroy(credentials_response); return 0; diff --git a/src/aws/flb_aws_credentials_http.c b/src/aws/flb_aws_credentials_http.c index 90e764e6d9d..50c1686558f 100644 --- a/src/aws/flb_aws_credentials_http.c +++ b/src/aws/flb_aws_credentials_http.c @@ -51,7 +51,8 @@ /* Declarations */ -static int http_credentials_request(struct flb_aws_provider_http +static int http_credentials_request(struct flb_aws_provider *provider, + struct flb_aws_provider_http *implementation); @@ -87,27 +88,36 @@ struct flb_aws_credentials *get_credentials_fn_http(struct flb_aws_provider *provider) { struct flb_aws_credentials *creds = NULL; + time_t next_refresh; int refresh = FLB_FALSE; struct flb_aws_provider_http *implementation = provider->implementation; flb_debug("[aws_credentials] Retrieving credentials from the " "HTTP provider.."); + next_refresh = flb_aws_cache_get_refresh_time(provider, + &implementation->next_refresh); + /* a negative next_refresh means that auto-refresh is disabled */ - if (implementation->next_refresh > 0 - && time(NULL) > implementation->next_refresh) { + if (next_refresh > 0 && time(NULL) > next_refresh) { refresh = FLB_TRUE; } - if (!implementation->creds || refresh == FLB_TRUE) { + + creds = flb_aws_cache_get_credentials(provider, &implementation->creds); + if (!creds || refresh == FLB_TRUE) { if (try_lock_provider(provider)) { - http_credentials_request(implementation); + http_credentials_request(provider, implementation); unlock_provider(provider); + + flb_aws_credentials_destroy(creds); + creds = flb_aws_cache_get_credentials(provider, + &implementation->creds); } else { flb_error("try_lock_provider failed"); } } - if (!implementation->creds) { + if (!creds) { /* * We failed to lock the provider and creds are unset. This means that * another co-routine is performing the refresh. @@ -119,42 +129,7 @@ struct flb_aws_credentials *get_credentials_fn_http(struct flb_aws_provider return NULL; } - creds = flb_calloc(1, sizeof(struct flb_aws_credentials)); - if (!creds) { - flb_errno(); - goto error; - } - - creds->access_key_id = flb_sds_create(implementation->creds->access_key_id); - if (!creds->access_key_id) { - flb_errno(); - goto error; - } - - creds->secret_access_key = flb_sds_create(implementation->creds-> - secret_access_key); - if (!creds->secret_access_key) { - flb_errno(); - goto error; - } - - if (implementation->creds->session_token) { - creds->session_token = flb_sds_create(implementation->creds-> - session_token); - if (!creds->session_token) { - flb_errno(); - goto error; - } - - } else { - creds->session_token = NULL; - } - return creds; - -error: - flb_aws_credentials_destroy(creds); - return NULL; } int refresh_fn_http(struct flb_aws_provider *provider) { @@ -163,7 +138,7 @@ int refresh_fn_http(struct flb_aws_provider *provider) { flb_debug("[aws_credentials] Refresh called on the http provider"); if (try_lock_provider(provider)) { - ret = http_credentials_request(implementation); + ret = http_credentials_request(provider, implementation); unlock_provider(provider); } return ret; @@ -177,7 +152,7 @@ int init_fn_http(struct flb_aws_provider *provider) { implementation->client->debug_only = FLB_TRUE; if (try_lock_provider(provider)) { - ret = http_credentials_request(implementation); + ret = http_credentials_request(provider, implementation); unlock_provider(provider); } @@ -274,6 +249,7 @@ struct flb_aws_provider *flb_endpoint_provider_create(struct flb_config *config, } pthread_mutex_init(&provider->lock, NULL); + pthread_mutex_init(&provider->cache_lock, NULL); implementation = flb_calloc(1, sizeof(struct flb_aws_provider_http)); @@ -411,7 +387,8 @@ static void trim_newline(char *token) } } -static int http_credentials_request(struct flb_aws_provider_http +static int http_credentials_request(struct flb_aws_provider *provider, + struct flb_aws_provider_http *implementation) { char *response = NULL; @@ -499,12 +476,10 @@ static int http_credentials_request(struct flb_aws_provider_http return -1; } - /* destroy existing credentials */ - flb_aws_credentials_destroy(implementation->creds); - implementation->creds = NULL; - - implementation->creds = creds; - implementation->next_refresh = expiration - FLB_AWS_REFRESH_WINDOW; + /* publish the new credentials; the old ones are freed for us */ + flb_aws_cache_set_credentials(provider, &implementation->creds, creds, + &implementation->next_refresh, + expiration - FLB_AWS_REFRESH_WINDOW); flb_http_client_destroy(c); return 0; diff --git a/src/aws/flb_aws_credentials_profile.c b/src/aws/flb_aws_credentials_profile.c index fd667e0940f..2e411dd51d9 100644 --- a/src/aws/flb_aws_credentials_profile.c +++ b/src/aws/flb_aws_credentials_profile.c @@ -49,7 +49,8 @@ /* Declarations */ struct flb_aws_provider_profile; -static int refresh_credentials(struct flb_aws_provider_profile *implementation, +static int refresh_credentials(struct flb_aws_provider *provider, + struct flb_aws_provider_profile *implementation, int debug_only); static int get_aws_shared_file_path(flb_sds_t* field, char* env_var, char* home_aws_path); @@ -91,68 +92,46 @@ struct flb_aws_credentials *get_credentials_fn_profile(struct flb_aws_provider *provider) { struct flb_aws_credentials *creds; + time_t next_refresh; int ret; struct flb_aws_provider_profile *implementation = provider->implementation; + next_refresh = flb_aws_cache_get_refresh_time(provider, + &implementation->next_refresh); + + creds = flb_aws_cache_get_credentials(provider, &implementation->creds); + /* * If next_refresh <= 0, it means we don't know how long the credentials * are valid for. So we won't refresh them unless explicitly asked * via refresh_fn_profile. */ - if (!implementation->creds || (implementation->next_refresh > 0 && - time(NULL) >= implementation->next_refresh)) { + if (!creds || (next_refresh > 0 && time(NULL) >= next_refresh)) { AWS_CREDS_DEBUG("Retrieving credentials for AWS Profile %s", implementation->profile); if (try_lock_provider(provider) == FLB_TRUE) { - ret = refresh_credentials(implementation, FLB_FALSE); + ret = refresh_credentials(provider, implementation, FLB_FALSE); unlock_provider(provider); + + flb_aws_credentials_destroy(creds); + creds = NULL; + if (ret < 0) { AWS_CREDS_ERROR("Failed to retrieve credentials for AWS Profile %s", implementation->profile); return NULL; } + + creds = flb_aws_cache_get_credentials(provider, + &implementation->creds); } else { AWS_CREDS_WARN("Another thread is refreshing credentials, will retry"); + flb_aws_credentials_destroy(creds); return NULL; } } - creds = flb_calloc(1, sizeof(struct flb_aws_credentials)); - if (!creds) { - flb_errno(); - goto error; - } - - creds->access_key_id = flb_sds_create(implementation->creds->access_key_id); - if (!creds->access_key_id) { - flb_errno(); - goto error; - } - - creds->secret_access_key = flb_sds_create(implementation-> - creds->secret_access_key); - if (!creds->secret_access_key) { - flb_errno(); - goto error; - } - - if (implementation->creds->session_token) { - creds->session_token = flb_sds_create(implementation-> - creds->session_token); - if (!creds->session_token) { - flb_errno(); - goto error; - } - - } else { - creds->session_token = NULL; - } - return creds; - -error: - flb_aws_credentials_destroy(creds); - return NULL; } int refresh_fn_profile(struct flb_aws_provider *provider) @@ -161,7 +140,7 @@ int refresh_fn_profile(struct flb_aws_provider *provider) int ret = -1; AWS_CREDS_DEBUG("Refresh called on the profile provider"); if (try_lock_provider(provider) == FLB_TRUE) { - ret = refresh_credentials(implementation, FLB_FALSE); + ret = refresh_credentials(provider, implementation, FLB_FALSE); unlock_provider(provider); return ret; } @@ -174,7 +153,7 @@ int init_fn_profile(struct flb_aws_provider *provider) int ret = -1; AWS_CREDS_DEBUG("Init called on the profile provider"); if (try_lock_provider(provider) == FLB_TRUE) { - ret = refresh_credentials(implementation, FLB_TRUE); + ret = refresh_credentials(provider, implementation, FLB_TRUE); unlock_provider(provider); return ret; } @@ -253,6 +232,7 @@ struct flb_aws_provider *flb_profile_provider_create(char* profile) } pthread_mutex_init(&provider->lock, NULL); + pthread_mutex_init(&provider->cache_lock, NULL); implementation = flb_calloc(1, sizeof( @@ -694,7 +674,8 @@ static int get_shared_credentials(char* credentials_path, return result; } -static int refresh_credentials(struct flb_aws_provider_profile *implementation, +static int refresh_credentials(struct flb_aws_provider *provider, + struct flb_aws_provider_profile *implementation, int debug_only) { struct flb_aws_credentials *creds = NULL; @@ -735,16 +716,16 @@ static int refresh_credentials(struct flb_aws_provider_profile *implementation, expiration = 0; } - /* unset and free existing credentials */ - flb_aws_credentials_destroy(implementation->creds); - implementation->creds = creds; - if (expiration > 0) { - implementation->next_refresh = expiration - FLB_AWS_REFRESH_WINDOW; + expiration -= FLB_AWS_REFRESH_WINDOW; } else { - implementation->next_refresh = 0; + expiration = 0; } + /* publish the new credentials; the old ones are freed for us */ + flb_aws_cache_set_credentials(provider, &implementation->creds, creds, + &implementation->next_refresh, expiration); + return 0; error: diff --git a/src/aws/flb_aws_credentials_sts.c b/src/aws/flb_aws_credentials_sts.c index 0a819277b67..611b96ae81d 100644 --- a/src/aws/flb_aws_credentials_sts.c +++ b/src/aws/flb_aws_credentials_sts.c @@ -66,9 +66,11 @@ struct flb_aws_provider_eks; void bytes_to_string(unsigned char *data, char *buf, size_t len); -static int assume_with_web_identity(struct flb_aws_provider_eks +static int assume_with_web_identity(struct flb_aws_provider *provider, + struct flb_aws_provider_eks *implementation); -static int sts_assume_role_request(struct flb_aws_client *sts_client, +static int sts_assume_role_request(struct flb_aws_provider *provider, + struct flb_aws_client *sts_client, struct flb_aws_credentials **creds, char *uri, time_t *next_refresh); @@ -98,31 +100,41 @@ struct flb_aws_credentials *get_credentials_fn_sts(struct flb_aws_provider *provider) { struct flb_aws_credentials *creds; + time_t next_refresh; int refresh = FLB_FALSE; struct flb_aws_provider_sts *implementation = provider->implementation; flb_debug("[aws_credentials] Requesting credentials from the " "STS provider.."); + next_refresh = flb_aws_cache_get_refresh_time(provider, + &implementation->next_refresh); + /* a negative next_refresh means that auto-refresh is disabled */ - if (implementation->next_refresh > 0 - && time(NULL) > implementation->next_refresh) { + if (next_refresh > 0 && time(NULL) > next_refresh) { refresh = FLB_TRUE; } - if (!implementation->creds || refresh == FLB_TRUE) { + + /* return a copy of the existing cached credentials */ + creds = flb_aws_cache_get_credentials(provider, &implementation->creds); + if (!creds || refresh == FLB_TRUE) { /* credentials need to be refreshed/obtained */ if (try_lock_provider(provider)) { flb_debug("[aws_credentials] STS Provider: Refreshing credential " "cache."); - sts_assume_role_request(implementation->sts_client, + sts_assume_role_request(provider, implementation->sts_client, &implementation->creds, implementation->uri, &implementation->next_refresh); unlock_provider(provider); + + flb_aws_credentials_destroy(creds); + creds = flb_aws_cache_get_credentials(provider, + &implementation->creds); } } - if (!implementation->creds) { + if (!creds) { /* * We failed to lock the provider and creds are unset. This means that * another co-routine is performing the refresh. @@ -134,40 +146,7 @@ struct flb_aws_credentials *get_credentials_fn_sts(struct flb_aws_provider return NULL; } - /* return a copy of the existing cached credentials */ - creds = flb_calloc(1, sizeof(struct flb_aws_credentials)); - if (!creds) { - goto error; - } - - creds->access_key_id = flb_sds_create(implementation->creds->access_key_id); - if (!creds->access_key_id) { - goto error; - } - - creds->secret_access_key = flb_sds_create(implementation->creds-> - secret_access_key); - if (!creds->secret_access_key) { - goto error; - } - - if (implementation->creds->session_token) { - creds->session_token = flb_sds_create(implementation->creds-> - session_token); - if (!creds->session_token) { - goto error; - } - - } else { - creds->session_token = NULL; - } - return creds; - -error: - flb_errno(); - flb_aws_credentials_destroy(creds); - return NULL; } int refresh_fn_sts(struct flb_aws_provider *provider) { @@ -177,7 +156,7 @@ int refresh_fn_sts(struct flb_aws_provider *provider) { flb_debug("[aws_credentials] Refresh called on the STS provider"); if (try_lock_provider(provider)) { - ret = sts_assume_role_request(implementation->sts_client, + ret = sts_assume_role_request(provider, implementation->sts_client, &implementation->creds, implementation->uri, &implementation->next_refresh); unlock_provider(provider); @@ -198,7 +177,7 @@ int init_fn_sts(struct flb_aws_provider *provider) { implementation->sts_client->debug_only = FLB_TRUE; if (try_lock_provider(provider)) { - ret = sts_assume_role_request(implementation->sts_client, + ret = sts_assume_role_request(provider, implementation->sts_client, &implementation->creds, implementation->uri, &implementation->next_refresh); unlock_provider(provider); @@ -309,6 +288,7 @@ struct flb_aws_provider *flb_sts_provider_create(struct flb_config *config, } pthread_mutex_init(&provider->lock, NULL); + pthread_mutex_init(&provider->cache_lock, NULL); implementation = flb_calloc(1, sizeof(struct flb_aws_provider_sts)); if (!implementation) { @@ -407,27 +387,36 @@ struct flb_aws_credentials *get_credentials_fn_eks(struct flb_aws_provider *provider) { struct flb_aws_credentials *creds = NULL; + time_t next_refresh; int refresh = FLB_FALSE; struct flb_aws_provider_eks *implementation = provider->implementation; flb_debug("[aws_credentials] Requesting credentials from the " "EKS provider.."); + next_refresh = flb_aws_cache_get_refresh_time(provider, + &implementation->next_refresh); + /* a negative next_refresh means that auto-refresh is disabled */ - if (implementation->next_refresh > 0 - && time(NULL) > implementation->next_refresh) { + if (next_refresh > 0 && time(NULL) > next_refresh) { refresh = FLB_TRUE; } - if (!implementation->creds || refresh == FLB_TRUE) { + + creds = flb_aws_cache_get_credentials(provider, &implementation->creds); + if (!creds || refresh == FLB_TRUE) { if (try_lock_provider(provider)) { flb_debug("[aws_credentials] EKS Provider: Refreshing credential " "cache."); - assume_with_web_identity(implementation); + assume_with_web_identity(provider, implementation); unlock_provider(provider); + + flb_aws_credentials_destroy(creds); + creds = flb_aws_cache_get_credentials(provider, + &implementation->creds); } } - if (!implementation->creds) { + if (!creds) { /* * We failed to lock the provider and creds are unset. This means that * another co-routine is performing the refresh. @@ -439,40 +428,7 @@ struct flb_aws_credentials *get_credentials_fn_eks(struct flb_aws_provider return NULL; } - creds = flb_calloc(1, sizeof(struct flb_aws_credentials)); - if (!creds) { - goto error; - } - - creds->access_key_id = flb_sds_create(implementation->creds->access_key_id); - if (!creds->access_key_id) { - goto error; - } - - creds->secret_access_key = flb_sds_create(implementation->creds-> - secret_access_key); - if (!creds->secret_access_key) { - goto error; - } - - if (implementation->creds->session_token) { - creds->session_token = flb_sds_create(implementation->creds-> - session_token); - if (!creds->session_token) { - goto error; - } - - } - else { - creds->session_token = NULL; - } - return creds; - -error: - flb_errno(); - flb_aws_credentials_destroy(creds); - return NULL; } int refresh_fn_eks(struct flb_aws_provider *provider) { @@ -481,7 +437,7 @@ int refresh_fn_eks(struct flb_aws_provider *provider) { flb_debug("[aws_credentials] Refresh called on the EKS provider"); if (try_lock_provider(provider)) { - ret = assume_with_web_identity(implementation); + ret = assume_with_web_identity(provider, implementation); unlock_provider(provider); } return ret; @@ -495,7 +451,7 @@ int init_fn_eks(struct flb_aws_provider *provider) { flb_debug("[aws_credentials] Init called on the EKS provider"); if (try_lock_provider(provider)) { - ret = assume_with_web_identity(implementation); + ret = assume_with_web_identity(provider, implementation); unlock_provider(provider); } @@ -582,6 +538,7 @@ struct flb_aws_provider *flb_eks_provider_create(struct flb_config *config, } pthread_mutex_init(&provider->lock, NULL); + pthread_mutex_init(&provider->cache_lock, NULL); implementation = flb_calloc(1, sizeof(struct flb_aws_provider_eks)); @@ -709,7 +666,8 @@ void bytes_to_string(unsigned char *data, char *buf, size_t len) { } } -static int assume_with_web_identity(struct flb_aws_provider_eks +static int assume_with_web_identity(struct flb_aws_provider *provider, + struct flb_aws_provider_eks *implementation) { int ret; @@ -736,7 +694,7 @@ static int assume_with_web_identity(struct flb_aws_provider_eks return -1; } - ret = sts_assume_role_request(implementation->sts_client, + ret = sts_assume_role_request(provider, implementation->sts_client, &implementation->creds, uri, &implementation->next_refresh); flb_free(web_token); @@ -744,7 +702,8 @@ static int assume_with_web_identity(struct flb_aws_provider_eks return ret; } -static int sts_assume_role_request(struct flb_aws_client *sts_client, +static int sts_assume_role_request(struct flb_aws_provider *provider, + struct flb_aws_client *sts_client, struct flb_aws_credentials **creds, char *uri, time_t *next_refresh) @@ -773,12 +732,10 @@ static int sts_assume_role_request(struct flb_aws_client *sts_client, return -1; } - /* unset and free existing credentials first */ - flb_aws_credentials_destroy(*creds); - *creds = NULL; - - *next_refresh = expiration - FLB_AWS_REFRESH_WINDOW; - *creds = credentials; + /* publish the new credentials; the old ones are freed for us */ + flb_aws_cache_set_credentials(provider, creds, credentials, + next_refresh, + expiration - FLB_AWS_REFRESH_WINDOW); flb_http_client_destroy(c); return 0; } From 1492bd85dcfb8f94cb5fafb4df757fb30509313a Mon Sep 17 00:00:00 2001 From: Md Nahid Hassan Bhuiyan Date: Tue, 4 Aug 2026 03:45:03 +0600 Subject: [PATCH 2/2] tests: internal: cover concurrent credential refresh Six threads read credentials from an EKS provider while a seventh refreshes it in a loop. The mock STS response tags the access key, secret key and session token with the same generation number, so a reader that stitched together two different refreshes is detectable without a sanitizer. Against the unfixed providers this segfaults within a second, the same crash reported in #12206. Under ASan or TSan it also flags the use-after-free directly. Signed-off-by: Md Nahid Hassan Bhuiyan --- tests/internal/aws_credentials_sts.c | 280 +++++++++++++++++++++++++++ 1 file changed, 280 insertions(+) diff --git a/tests/internal/aws_credentials_sts.c b/tests/internal/aws_credentials_sts.c index d342e158aed..9b2ec6cad9a 100644 --- a/tests/internal/aws_credentials_sts.c +++ b/tests/internal/aws_credentials_sts.c @@ -7,6 +7,8 @@ #include #include +#include +#include #include #include @@ -221,6 +223,83 @@ struct flb_http_client *request_eks_test1(struct flb_aws_client *aws_client, return c; } +/* + * Mock for the concurrency test. Every call hands back a different set of + * credentials, with the same generation number embedded in all three fields. + * That lets a reader tell whether it copied a single, coherent set or mixed + * fields from two different refreshes. + * + * Only the writer thread ever reaches this function, so the counter does not + * need to be atomic. + */ +static int g_concurrency_generation; + +static char *build_eks_response_for_generation(int generation, size_t *out_len) +{ + time_t exp = time(NULL) + 3600; + struct tm gm; + char expbuf[32]; + const char *tmpl; + size_t need; + char *buf; + + gmtime_r(&exp, &gm); + strftime(expbuf, sizeof(expbuf), "%Y-%m-%dT%H:%M:%SZ", &gm); + + tmpl = + "\n" + " \n" + " \n" + " token_%d\n" + " skid_%d\n" + " %s\n" + " akid_%d\n" + " \n" + " \n" + ""; + + need = (size_t) snprintf(NULL, 0, tmpl, generation, generation, expbuf, + generation) + 1; + buf = flb_calloc(1, need); + if (!buf) { + flb_errno(); + return NULL; + } + snprintf(buf, need, tmpl, generation, generation, expbuf, generation); + + if (out_len) { + *out_len = need - 1; + } + return buf; +} + +struct flb_http_client *request_eks_concurrency(struct flb_aws_client *aws_client, + int method, const char *uri) +{ + struct flb_http_client *c; + char *payload = NULL; + size_t payload_len = 0; + + c = flb_calloc(1, sizeof(struct flb_http_client)); + if (!c) { + flb_errno(); + return NULL; + } + mk_list_init(&c->headers); + + payload = build_eks_response_for_generation(++g_concurrency_generation, + &payload_len); + if (!payload) { + flb_free(c); + return NULL; + } + + http_test_attach_owned_payload(c, payload, payload_len); + + return c; +} + struct flb_http_client *request_eks_flb_sts_session_name(struct flb_aws_client *aws_client, int method, @@ -363,6 +442,8 @@ struct flb_http_client *test_http_client_request(struct flb_aws_client *aws_clie */ if (strstr(uri, "test1") != NULL) { return request_eks_test1(aws_client, method, uri); + } else if (strstr(uri, "concurrency") != NULL) { + return request_eks_concurrency(aws_client, method, uri); } else if (strstr(uri, "randomsession") != NULL) { return request_eks_flb_sts_session_name(aws_client, method, uri); } else if (strstr(uri, "apierror") != NULL) { @@ -572,6 +653,203 @@ static void test_eks_provider() { flb_config_exit(config); } +/* + * Regression test for https://github.com/fluent/fluent-bit/issues/12206 + * + * With `workers N` an output runs its flushes on N real threads that all share + * one credential provider. get_credentials() used to copy the cached + * credentials without holding the provider lock while the refresh path freed + * and replaced them under it, so a reader could copy freed memory, see the + * NULL the refresh left behind mid-swap, or end up with an access key and a + * secret key from two different refreshes. + * + * Here one thread refreshes in a loop while several read in a loop. The mock + * STS response tags all three credential fields with the same generation + * number, so a torn read is detectable. Run under ASan or TSan to also catch + * the use-after-free itself. + */ + +#define CONCURRENCY_READERS 6 +#define CONCURRENCY_REFRESHES 3000 + +struct concurrency_ctx { + struct flb_aws_provider *provider; + int stop; + + /* results, one slot per reader so they need no synchronization */ + int null_creds[CONCURRENCY_READERS]; + int torn_creds[CONCURRENCY_READERS]; + int reads[CONCURRENCY_READERS]; +}; + +struct concurrency_reader_arg { + struct concurrency_ctx *ctx; + int id; +}; + +/* Returns the generation number in a "_" credential field */ +static int credential_generation(const char *value, const char *prefix) +{ + size_t prefix_len = strlen(prefix); + + if (!value || strncmp(value, prefix, prefix_len) != 0) { + return -1; + } + + return atoi(value + prefix_len); +} + +static void *concurrency_reader(void *arg) +{ + struct concurrency_reader_arg *reader = arg; + struct concurrency_ctx *ctx = reader->ctx; + struct flb_aws_provider *provider = ctx->provider; + struct flb_aws_credentials *creds; + int akid; + int skid; + int token; + + while (ctx->stop == FLB_FALSE) { + creds = provider->provider_vtable->get_credentials(provider); + if (!creds) { + ctx->null_creds[reader->id]++; + continue; + } + + akid = credential_generation(creds->access_key_id, "akid_"); + skid = credential_generation(creds->secret_access_key, "skid_"); + token = credential_generation(creds->session_token, "token_"); + + if (akid < 0 || akid != skid || akid != token) { + ctx->torn_creds[reader->id]++; + } + + ctx->reads[reader->id]++; + flb_aws_credentials_destroy(creds); + } + + return NULL; +} + +static void *concurrency_writer(void *arg) +{ + struct concurrency_ctx *ctx = arg; + struct flb_aws_provider *provider = ctx->provider; + int i; + + for (i = 0; i < CONCURRENCY_REFRESHES; i++) { + provider->provider_vtable->refresh(provider); + } + + ctx->stop = FLB_TRUE; + return NULL; +} + +static void test_eks_provider_concurrent_refresh() +{ + struct flb_config *config; + struct flb_aws_provider *provider; + struct flb_aws_credentials *creds; + struct concurrency_ctx ctx; + struct concurrency_reader_arg args[CONCURRENCY_READERS]; + pthread_t readers[CONCURRENCY_READERS]; + pthread_t writer; + int total_null = 0; + int total_torn = 0; + int total_reads = 0; + int ret; + int i; + + g_request_count = 0; + g_concurrency_generation = 0; + + config = flb_config_init(); + if (config == NULL) { + return; + } + + ret = setenv(ROLE_ARN_ENV_VAR, + "arn:aws:iam::123456789012:role/concurrency", 1); + if (ret < 0) { + flb_errno(); + flb_config_exit(config); + return; + } + ret = setenv(SESSION_NAME_ENV_VAR, "session_name", 1); + if (ret < 0) { + flb_errno(); + flb_config_exit(config); + return; + } + ret = setenv(TOKEN_FILE_ENV_VAR, WEB_TOKEN_FILE, 1); + if (ret < 0) { + flb_errno(); + flb_config_exit(config); + return; + } + + provider = flb_eks_provider_create(config, NULL, "us-west-2", + "https://sts.us-west-2.amazonaws.com", + NULL, generator_in_test()); + if (!TEST_CHECK(provider != NULL)) { + unsetenv_eks(); + flb_config_exit(config); + return; + } + + /* prime the cache so the readers never have to refresh themselves */ + creds = provider->provider_vtable->get_credentials(provider); + if (!TEST_CHECK(creds != NULL)) { + flb_aws_provider_destroy(provider); + unsetenv_eks(); + flb_config_exit(config); + return; + } + flb_aws_credentials_destroy(creds); + + memset(&ctx, 0, sizeof(ctx)); + ctx.provider = provider; + ctx.stop = FLB_FALSE; + + for (i = 0; i < CONCURRENCY_READERS; i++) { + args[i].ctx = &ctx; + args[i].id = i; + ret = pthread_create(&readers[i], NULL, concurrency_reader, &args[i]); + TEST_CHECK(ret == 0); + } + + ret = pthread_create(&writer, NULL, concurrency_writer, &ctx); + TEST_CHECK(ret == 0); + + pthread_join(writer, NULL); + for (i = 0; i < CONCURRENCY_READERS; i++) { + pthread_join(readers[i], NULL); + + total_null += ctx.null_creds[i]; + total_torn += ctx.torn_creds[i]; + total_reads += ctx.reads[i]; + } + + /* the readers should have actually run */ + TEST_CHECK(total_reads > 0); + TEST_MSG("reads=%d", total_reads); + + /* + * The cache is primed and the credentials are valid for an hour, so a + * reader has no reason to ever come back empty handed or with a set of + * credentials that was stitched together from two refreshes. + */ + TEST_CHECK(total_null == 0); + TEST_MSG("get_credentials returned NULL %d times", total_null); + + TEST_CHECK(total_torn == 0); + TEST_MSG("got %d torn credential sets", total_torn); + + flb_aws_provider_destroy(provider); + unsetenv_eks(); + flb_config_exit(config); +} + static void test_eks_provider_random_session_name() { struct flb_config *config; struct flb_aws_provider *provider; @@ -1019,6 +1297,8 @@ TEST_LIST = { { "test_sts_uri" , test_sts_uri}, { "process_sts_response" , test_process_sts_response}, { "eks_credential_provider" , test_eks_provider}, + { "eks_credential_provider_concurrent_refresh" , + test_eks_provider_concurrent_refresh}, { "eks_credential_provider_random_session_name" , test_eks_provider_random_session_name}, { "test_eks_provider_unexpected_api_response" ,