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; } 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" ,