oauth2: harden configuration, JWKS refresh, and token lifetimes

include/fluent-bit/flb_input.h

@@ -74,6 +74,7 @@
                                      * private key and certificate are required.
                                      */
 #define FLB_INPUT_HTTP_SERVER 4096  /* input uses the generic HTTP server     */
+#define FLB_INPUT_OAUTH2_JWT  8192  /* input supports OAuth2 JWT validation  */
 
 /* Input status */
 #define FLB_INPUT_RUNNING     1

include/fluent-bit/flb_output.h

@@ -85,6 +85,7 @@ int flb_chunk_trace_output(struct flb_chunk_trace *trace, struct flb_output_inst
 #define FLB_OUTPUT_PRIVATE      1024
 #define FLB_OUTPUT_SYNCHRONOUS  2048  /* run one task at a time, no flush cycle limit */
 #define FLB_OUTPUT_HTTP_SERVER  4096  /* output uses the generic HTTP server  */
+#define FLB_OUTPUT_OAUTH2_CLIENT 8192 /* output supports OAuth2 authentication */
 
 
 /*

plugins/in_http/http.c

@@ -192,5 +192,6 @@ struct flb_input_plugin in_http_plugin = {
     .cb_resume    = NULL,
     .cb_exit      = in_http_exit,
     .config_map   = config_map,
-    .flags        = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS
+    .flags        = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER |
+                    FLB_INPUT_OAUTH2_JWT | FLB_IO_OPT_TLS
 };

plugins/in_opentelemetry/opentelemetry.c

@@ -201,5 +201,6 @@ struct flb_input_plugin in_opentelemetry_plugin = {
     .cb_resume    = NULL,
     .cb_exit      = in_opentelemetry_exit,
     .config_map   = config_map,
-    .flags        = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS
+    .flags        = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER |
+                    FLB_INPUT_OAUTH2_JWT | FLB_IO_OPT_TLS
 };

plugins/out_http/http.c

@@ -943,6 +943,6 @@ struct flb_output_plugin out_http_plugin = {
     /* for testing */
     .test_formatter.callback = cb_http_format_test,
 
-    .flags       = FLB_OUTPUT_NET | FLB_IO_OPT_TLS,
+    .flags       = FLB_OUTPUT_NET | FLB_OUTPUT_OAUTH2_CLIENT | FLB_IO_OPT_TLS,
     .workers     = 2
 };

plugins/out_opentelemetry/opentelemetry.c

@@ -1288,7 +1288,7 @@ struct flb_output_plugin out_opentelemetry_plugin = {
     .cb_exit     = cb_opentelemetry_exit,
     .config_map  = config_map,
     .event_type  = FLB_OUTPUT_LOGS | FLB_OUTPUT_METRICS | FLB_OUTPUT_TRACES | FLB_OUTPUT_PROFILES,
-    .flags       = FLB_OUTPUT_NET | FLB_IO_OPT_TLS,
+    .flags       = FLB_OUTPUT_NET | FLB_OUTPUT_OAUTH2_CLIENT | FLB_IO_OPT_TLS,
 
     .test_formatter.callback = opentelemetry_format_test,
 };

src/flb_input.c

@@ -771,7 +771,14 @@ int flb_input_set_property(struct flb_input_instance *ins,
         }
         kv->val = tmp;
     }
-    else if (strncasecmp("oauth2", k, 6) == 0 && tmp) {
+    else if (strncasecmp("oauth2.", k, 7) == 0 && tmp) {
+        if ((ins->p->flags & FLB_INPUT_OAUTH2_JWT) == 0) {
+            flb_error("[config] input plugin '%s' does not support OAuth2 JWT validation",
+                      ins->p->name);
+            flb_sds_destroy(tmp);
+            return -1;
+        }
+
         kv = flb_kv_item_create(&ins->oauth2_jwt_properties, (char *) k, NULL);
         if (!kv) {
             if (tmp) {
@@ -1331,6 +1338,12 @@ int flb_input_oauth2_jwt_property_check(struct flb_input_instance *ins,
      * it might receive OAuth2 JWT settings.
      */
     if (mk_list_size(&ins->oauth2_jwt_properties) > 0) {
+        if ((ins->p->flags & FLB_INPUT_OAUTH2_JWT) == 0) {
+            flb_error("[config] input plugin '%s' does not support OAuth2 JWT validation",
+                      ins->p->name);
+            return -1;
+        }
+
         ret = flb_config_map_properties_check(ins->p->name,
                                               &ins->oauth2_jwt_properties,
                                               ins->oauth2_jwt_config_map);

src/flb_oauth2.c

@@ -469,6 +469,7 @@ int flb_oauth2_parse_json_response(const char *json_data, size_t json_size,
     jsmntok_t *tokens;
     char tmp_num[32];
     uint64_t new_expires_in = 0;
+    int expires_in_present = FLB_FALSE;
 
     jsmn_init(&parser);
     tokens = flb_calloc(1, sizeof(jsmntok_t) * tokens_size);
@@ -537,6 +538,8 @@ int flb_oauth2_parse_json_response(const char *json_data, size_t json_size,
             }
         }
         else if (key_cmp(key, key_len, "expires_in") == 0) {
+            expires_in_present = FLB_TRUE;
+
             if (val_len <= 0 || val_len >= sizeof(tmp_num)) {
                 break;
             }
@@ -557,18 +560,22 @@ int flb_oauth2_parse_json_response(const char *json_data, size_t json_size,
             }
 
             new_expires_in = parsed_expires_in;
-            new_expires_in -= (new_expires_in / 10);
         }
     }
 
     flb_free(tokens);
 
-    if (!new_access_token || !new_token_type || new_expires_in <= ctx->refresh_skew) {
+    if (!new_access_token || !new_token_type ||
+        (expires_in_present == FLB_TRUE && new_expires_in == 0)) {
         flb_sds_destroy(new_access_token);
         flb_sds_destroy(new_token_type);
         return -1;
     }
 
+    if (expires_in_present == FLB_FALSE) {
+        new_expires_in = FLB_OAUTH2_DEFAULT_EXPIRES;
+    }
+
     oauth2_reset_state(ctx);
 
     ctx->access_token = new_access_token;
@@ -1195,6 +1202,7 @@ static int oauth2_refresh_locked(struct flb_oauth2 *ctx)
 static int oauth2_token_needs_refresh(struct flb_oauth2 *ctx, int force_refresh)
 {
     time_t now;
+    time_t refresh_skew;
 
     if (force_refresh) {
         return FLB_TRUE;
@@ -1210,7 +1218,12 @@ static int oauth2_token_needs_refresh(struct flb_oauth2 *ctx, int force_refresh)
         return FLB_TRUE;
     }
 
-    if (now >= (ctx->expires_at - ctx->refresh_skew)) {
+    refresh_skew = ctx->refresh_skew;
+    if (ctx->expires_in <= (uint64_t) refresh_skew) {
+        refresh_skew = ctx->expires_in / 10;
+    }
+
+    if (now >= (ctx->expires_at - refresh_skew)) {
         return FLB_TRUE;
     }
 

src/flb_oauth2_jwt.c

@@ -125,47 +125,6 @@ static void oauth2_jwks_key_destroy(struct flb_oauth2_jwks_key *key)
     flb_free(key);
 }
 
-static int oauth2_jwks_cache_clear(struct flb_oauth2_jwks_cache *cache)
-{
-    int i;
-    struct mk_list *head;
-    struct mk_list *tmp;
-    struct flb_hash_table_entry *entry;
-    struct flb_hash_table_chain *table;
-    int refresh_interval;
-
-    if (!cache || !cache->entries) {
-        return 0;
-    }
-
-    /* Save refresh interval before destroying */
-    refresh_interval = cache->refresh_interval;
-
-    /* Iterate through all hash table chains and destroy keys */
-    for (i = 0; i < cache->entries->size; i++) {
-        table = &cache->entries->table[i];
-        mk_list_foreach_safe(head, tmp, &table->chains) {
-            entry = mk_list_entry(head, struct flb_hash_table_entry, _head);
-            if (entry->val) {
-                oauth2_jwks_key_destroy((struct flb_oauth2_jwks_key *)entry->val);
-                entry->val = NULL; /* Prevent double-free */
-            }
-        }
-    }
-
-    /* Destroy and recreate the hash table to clear all entries */
-    flb_hash_table_destroy(cache->entries);
-    cache->entries = flb_hash_table_create(FLB_HASH_TABLE_EVICT_NONE, 64, 0);
-    if (!cache->entries) {
-        flb_error("[oauth2_jwt] failed to recreate JWKS cache after clear");
-        return -1;
-    }
-
-    /* Restore refresh interval */
-    cache->refresh_interval = refresh_interval;
-    return 0;
-}
-
 static void oauth2_jwks_cache_destroy(struct flb_oauth2_jwks_cache *cache)
 {
     int i;
@@ -904,10 +863,24 @@ static int oauth2_jwks_parse_json(flb_sds_t jwks_json, struct flb_oauth2_jwks_ca
 
                 ret = oauth2_jwks_parse_key(key_obj, &jwks_key);
                 if (ret == 0 && jwks_key) {
+                    if (flb_hash_table_get_ptr(cache->entries, jwks_key->kid,
+                                               flb_sds_len(jwks_key->kid))) {
+                        flb_error("[oauth2_jwt] duplicate key ID in JWKS: %s",
+                                  jwks_key->kid);
+                        oauth2_jwks_key_destroy(jwks_key);
+                        keys_found = -1;
+                        break;
+                    }
+
                     /* Store key in cache using kid as hash key */
-                    flb_hash_table_add(cache->entries, jwks_key->kid,
-                                      flb_sds_len(jwks_key->kid),
-                                      jwks_key, 0);
+                    ret = flb_hash_table_add(cache->entries, jwks_key->kid,
+                                             flb_sds_len(jwks_key->kid),
+                                             jwks_key, 0);
+                    if (ret < 0) {
+                        oauth2_jwks_key_destroy(jwks_key);
+                        keys_found = -1;
+                        break;
+                    }
                     keys_found++;
                 }
             }
@@ -918,6 +891,11 @@ static int oauth2_jwks_parse_json(flb_sds_t jwks_json, struct flb_oauth2_jwks_ca
     flb_free(mp_buf);
     msgpack_unpacked_destroy(&result);
 
+    if (keys_found < 0) {
+        flb_error("[oauth2_jwt] failed to build JWKS cache");
+        return -1;
+    }
+
     if (keys_found == 0) {
         flb_error("[oauth2_jwt] No valid keys found in JWKS");
         return -1;
@@ -994,7 +972,9 @@ static int oauth2_jwks_fetch_keys(struct flb_oauth2_jwt_ctx *ctx)
     struct flb_connection *u_conn = NULL;
     struct flb_http_client *c = NULL;
     struct flb_tls *tls = NULL;
+    struct flb_oauth2_jwks_cache new_cache = {0};
     flb_sds_t jwks_json = NULL;
+    int new_cache_initialized = FLB_FALSE;
 
     if (!ctx || !ctx->cfg.jwks_url || !ctx->config) {
         return -1;
@@ -1092,26 +1072,31 @@ static int oauth2_jwks_fetch_keys(struct flb_oauth2_jwt_ctx *ctx)
         goto cleanup;
     }
 
-    /* Clear existing cache entries before refreshing to ensure revoked/rotated keys are removed */
-    ret = oauth2_jwks_cache_clear(&ctx->jwks_cache);
+    ret = oauth2_jwks_cache_init(&new_cache, ctx->jwks_cache.refresh_interval);
     if (ret != 0) {
-        flb_error("[oauth2_jwt] failed to clear JWKS cache");
+        flb_error("[oauth2_jwt] failed to initialize refreshed JWKS cache");
         goto cleanup;
     }
+    new_cache_initialized = FLB_TRUE;
 
-    /* Parse JWKS JSON and store keys in cache */
-    ret = oauth2_jwks_parse_json(jwks_json, &ctx->jwks_cache);
+    /* Build the replacement cache completely before discarding valid keys. */
+    ret = oauth2_jwks_parse_json(jwks_json, &new_cache);
     if (ret != 0) {
         flb_error("[oauth2_jwt] failed to parse JWKS JSON");
-        flb_sds_destroy(jwks_json);
-        jwks_json = NULL;
-    }
-    else {
-        ctx->jwks_cache.last_refresh = time(NULL);
-        ret_code = 0;
+        goto cleanup;
     }
 
+    new_cache.last_refresh = time(NULL);
+    oauth2_jwks_cache_destroy(&ctx->jwks_cache);
+    ctx->jwks_cache = new_cache;
+    memset(&new_cache, 0, sizeof(new_cache));
+    new_cache_initialized = FLB_FALSE;
+    ret_code = 0;
+
 cleanup:
+    if (new_cache_initialized == FLB_TRUE) {
+        oauth2_jwks_cache_destroy(&new_cache);
+    }
     if (jwks_json) {
         flb_sds_destroy(jwks_json);
     }
@@ -1392,8 +1377,12 @@ int flb_oauth2_jwt_validate(struct flb_oauth2_jwt_ctx *ctx,
             ret = oauth2_jwks_fetch_keys(ctx);
             if (ret != 0) {
                 flb_debug("[oauth2_jwt] Failed to fetch JWKS: %d", ret);
-                status = FLB_OAUTH2_JWT_ERR_INVALID_ARGUMENT;
-                goto jwt_end;
+                if (ctx->jwks_cache.last_refresh == 0) {
+                    status = FLB_OAUTH2_JWT_ERR_INVALID_ARGUMENT;
+                    goto jwt_end;
+                }
+
+                flb_warn("[oauth2_jwt] using cached keys after JWKS refresh failure");
             }
         }
 

src/flb_output.c

@@ -976,7 +976,14 @@ int flb_output_set_property(struct flb_output_instance *ins,
         }
         kv->val = tmp;
     }
-    else if (strncasecmp("oauth2", k, 6) == 0 && tmp) {
+    else if (strncasecmp("oauth2.", k, 7) == 0 && tmp) {
+        if ((ins->p->flags & FLB_OUTPUT_OAUTH2_CLIENT) == 0) {
+            flb_error("[config] output plugin '%s' does not support OAuth2 authentication",
+                      ins->p->name);
+            flb_sds_destroy(tmp);
+            return -1;
+        }
+
         kv = flb_kv_item_create(&ins->oauth2_properties, (char *) k, NULL);
         if (!kv) {
             if (tmp) {
@@ -1234,6 +1241,12 @@ int flb_output_oauth2_property_check(struct flb_output_instance *ins,
      * it might receive OAuth2 settings.
      */
     if (mk_list_size(&ins->oauth2_properties) > 0) {
+        if ((ins->p->flags & FLB_OUTPUT_OAUTH2_CLIENT) == 0) {
+            flb_error("[config] output plugin '%s' does not support OAuth2 authentication",
+                      ins->p->name);
+            return -1;
+        }
+
         ret = flb_config_map_properties_check(ins->p->name,
                                               &ins->oauth2_properties,
                                               ins->oauth2_config_map);

tests/integration/scenarios/in_http/config/in_http_oauth2_refresh.yaml

@@ -0,0 +1,25 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: http
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      oauth2.validate: true
+      oauth2.issuer: issuer
+      oauth2.jwks_url: http://127.0.0.1:${TEST_SUITE_HTTP_PORT}/jwks
+      oauth2.jwks_refresh_interval: 1
+      oauth2.allowed_audience: audience
+      oauth2.allowed_clients: client1
+
+  outputs:
+    - name: http
+      match: '*'
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      uri: /data
+      format: json
+      json_date_key: false

tests/integration/scenarios/in_http/tests/test_in_http_001.py

@@ -7,7 +7,7 @@
 import pytest
 import requests
 
-from server.http_server import data_storage, http_server_run
+from server.http_server import configure_jwks_response, data_storage, http_server_run
 from utils.http_matrix import PROTOCOL_CASES, run_curl_request
 from utils.test_service import FluentBitTestService
 
@@ -229,6 +229,36 @@ def test_in_http_oauth2_accepts_valid_jwt():
     assert forwarded_payloads[0][0]["message"] == "Este es un mensaje de prueba"
 
 
+def test_in_http_oauth2_keeps_cached_keys_after_bad_jwks_refresh():
+    service = Service("in_http_oauth2_refresh.yaml")
+    service.start()
+    headers = [
+        "Content-Type: application/json",
+        f"Authorization: Bearer {MOCK_VALID_JWT}",
+    ]
+
+    first_result = run_curl_request(
+        f"http://localhost:{service.flb_listener_port}/",
+        create_payload("sample_data.json"),
+        headers=headers,
+        http_mode="http1.1",
+    )
+
+    configure_jwks_response(body="{malformed", content_type="application/json")
+    time.sleep(1.2)
+
+    second_result = run_curl_request(
+        f"http://localhost:{service.flb_listener_port}/",
+        create_payload("sample_data.json"),
+        headers=headers,
+        http_mode="http1.1",
+    )
+    service.stop()
+
+    assert first_result["status_code"] == 201
+    assert second_result["status_code"] == 201
+
+
 class Service:
     def __init__(self, config_file):
         self.config_file = os.path.abspath(os.path.join(os.path.dirname(__file__), '../config/', config_file))