oauth2_jwt: mitigate leaks on duplicated JWT members

src/flb_oauth2_jwt.c

@@ -390,9 +390,8 @@ static int oauth2_jwt_parse_header(const char *json, size_t json_len,
     size_t i;
     size_t map_size;
     size_t key_len;
-    size_t val_len;
     const char *key_str;
-    const char *val_str;
+    flb_sds_t *field;
 
     /* Convert JSON to msgpack */
     ret = flb_pack_json(json, json_len, &mp_buf, &mp_size,
@@ -429,24 +428,42 @@ static int oauth2_jwt_parse_header(const char *json, size_t json_len,
             continue;
         }
 
-        if (v->type == MSGPACK_OBJECT_STR) {
-            key_len = k->via.str.size;
-            val_len = v->via.str.size;
-            key_str = (const char *) k->via.str.ptr;
-            val_str = (const char *) v->via.str.ptr;
+        key_len = k->via.str.size;
+        key_str = k->via.str.ptr;
 
-            if (key_len == 3 && strncmp(key_str, "kid", 3) == 0) {
-                claims->kid = flb_sds_create_len(val_str, val_len);
-            }
-            else if (key_len == 3 && strncmp(key_str, "alg", 3) == 0) {
-                claims->alg = flb_sds_create_len(val_str, val_len);
-            }
+        if (key_len == 3 && strncmp(key_str, "kid", 3) == 0) {
+            field = &claims->kid;
+        }
+        else if (key_len == 3 && strncmp(key_str, "alg", 3) == 0) {
+            field = &claims->alg;
+        }
+        else {
+            continue;
+        }
+
+        /* Reject duplicates before overwriting an owned value. */
+        if (*field || v->type != MSGPACK_OBJECT_STR) {
+            ret = FLB_OAUTH2_JWT_ERR_JSON_HEADER;
+            goto cleanup;
+        }
+
+        *field = flb_sds_create_len(v->via.str.ptr, v->via.str.size);
+        if (!*field) {
+            ret = FLB_OAUTH2_JWT_ERR_INVALID_ARGUMENT;
+            goto cleanup;
         }
     }
 
+    ret = FLB_OAUTH2_JWT_OK;
+
+cleanup:
     flb_free(mp_buf);
     msgpack_unpacked_destroy(&result);
 
+    if (ret != FLB_OAUTH2_JWT_OK) {
+        return ret;
+    }
+
     if (!claims->kid) {
         return FLB_OAUTH2_JWT_ERR_MISSING_KID;
     }
@@ -727,6 +744,7 @@ static int oauth2_jwks_parse_key(msgpack_object *key_obj,
     flb_sds_t kid = NULL;
     flb_sds_t n = NULL;
     flb_sds_t e = NULL;
+    flb_sds_t *field;
     msgpack_object *k;
     msgpack_object *v;
     size_t i;
@@ -764,58 +782,44 @@ static int oauth2_jwks_parse_key(msgpack_object *key_obj,
                 }
                 else {
                     /* Not an RSA key */
-                    if (kid) {
-                        flb_sds_destroy(kid);
-                    }
-                    if (n) {
-                        flb_sds_destroy(n);
-                    }
-                    if (e) {
-                        flb_sds_destroy(e);
-                    }
-                    return -1;
+                    goto error;
                 }
             }
+            continue;
         }
         else if (key_len == 3 && strncmp(key_str, "kid", 3) == 0) {
-            if (v->type == MSGPACK_OBJECT_STR) {
-                kid = flb_sds_create_len((const char *) v->via.str.ptr,
-                                         v->via.str.size);
-            }
+            field = &kid;
         }
         else if (key_len == 1 && strncmp(key_str, "n", 1) == 0) {
-            if (v->type == MSGPACK_OBJECT_STR) {
-                n = flb_sds_create_len((const char *) v->via.str.ptr, v->via.str.size);
-            }
+            field = &n;
         }
         else if (key_len == 1 && strncmp(key_str, "e", 1) == 0) {
-            if (v->type == MSGPACK_OBJECT_STR) {
-                e = flb_sds_create_len((const char *) v->via.str.ptr, v->via.str.size);
-            }
+            field = &e;
         }
-        /* Ignore other fields (e.g. x5c array) - msgpack handles nested structures automatically */
-    }
-
-    if (!kid || !n || !e) {
-        if (kid) {
-            flb_sds_destroy(kid);
+        else {
+            /* Ignore other fields (e.g. x5c arrays). */
+            continue;
         }
-        if (n) {
-            flb_sds_destroy(n);
+
+        /* Reject duplicates and invalid types without losing prior allocations. */
+        if (*field || v->type != MSGPACK_OBJECT_STR) {
+            goto error;
         }
-        if (e) {
-            flb_sds_destroy(e);
+
+        *field = flb_sds_create_len(v->via.str.ptr, v->via.str.size);
+        if (!*field) {
+            goto error;
         }
-        return -1;
+    }
+
+    if (!kid || !n || !e) {
+        goto error;
     }
 
     key = flb_calloc(1, sizeof(struct flb_oauth2_jwks_key));
     if (!key) {
         flb_errno();
-        flb_sds_destroy(kid);
-        flb_sds_destroy(n);
-        flb_sds_destroy(e);
-        return -1;
+        goto error;
     }
 
     key->kid = kid;
@@ -825,6 +829,12 @@ static int oauth2_jwks_parse_key(msgpack_object *key_obj,
 
     *key_out = key;
     return 0;
+
+error:
+    flb_sds_destroy(kid);
+    flb_sds_destroy(n);
+    flb_sds_destroy(e);
+    return -1;
 }
 
 static int oauth2_jwks_parse_json(flb_sds_t jwks_json, struct flb_oauth2_jwks_cache *cache)

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

@@ -12,6 +12,7 @@
 
 from server.http_server import data_storage, http_server_run
 from utils.http_matrix import PROTOCOL_CASES, run_curl_request
+from utils.oauth2 import duplicate_header_token, duplicate_jwks_body, track_jwks_requests
 from utils.input_pause_resume import (
     assert_connection_closed,
     assert_pause_resume_cycles,
@@ -753,6 +754,58 @@ def test_in_http_oauth2_accepts_valid_jwt():
     assert forwarded_payloads[0][0]["message"] == "Este es un mensaje de prueba"
 
 
+@pytest.mark.parametrize("field", ["kid", "alg"])
+def test_in_http_oauth2_rejects_duplicate_header(monkeypatch, field):
+    jwks_requests = track_jwks_requests(monkeypatch)
+    token = duplicate_header_token(MOCK_VALID_JWT, field)
+    service = Service("in_http_oauth2.yaml")
+    service.start()
+    try:
+        for _ in range(8):
+            result = run_curl_request(
+                f"http://localhost:{service.flb_listener_port}/",
+                create_payload("sample_data.json"),
+                headers=["Content-Type: application/json", f"Authorization: Bearer {token}"],
+                http_mode="http1.1",
+            )
+            assert result["status_code"] == 401
+        # Rejection must happen during parsing, before fetching a verification key.
+        assert jwks_requests == []
+        assert data_storage["payloads"] == []
+
+        result = run_curl_request(
+            f"http://localhost:{service.flb_listener_port}/",
+            create_payload("sample_data.json"),
+            headers=["Content-Type: application/json", f"Authorization: Bearer {MOCK_VALID_JWT}"],
+            http_mode="http1.1",
+        )
+        assert result["status_code"] == 201
+        assert len(service.read_forwarded_payloads()) == 1
+    finally:
+        service.stop()
+
+
+@pytest.mark.parametrize("field", ["kid", "n", "e"])
+@pytest.mark.parametrize("first_value", ["earlier-value", "", None, []])
+def test_in_http_oauth2_rejects_duplicate_jwks(monkeypatch, field, first_value):
+    jwks_requests = track_jwks_requests(monkeypatch, duplicate_jwks_body(field, first_value))
+    service = Service("in_http_oauth2.yaml")
+    service.start()
+    try:
+        for _ in range(3):
+            result = run_curl_request(
+                f"http://localhost:{service.flb_listener_port}/",
+                create_payload("sample_data.json"),
+                headers=["Content-Type: application/json", f"Authorization: Bearer {MOCK_VALID_JWT}"],
+                http_mode="http1.1",
+            )
+            assert result["status_code"] == 401
+        assert jwks_requests
+        assert data_storage["payloads"] == []
+    finally:
+        service.stop()
+
+
 class Service:
     def __init__(self, config_file, extra_env=None):
         self.config_file = os.path.abspath(os.path.join(os.path.dirname(__file__), '../config/', config_file))

tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py

@@ -37,6 +37,7 @@
 # local imports
 from utils.data_utils import read_json_file
 from utils.http_matrix import PROTOCOL_CASES, run_curl_request
+from utils.oauth2 import duplicate_header_token, track_jwks_requests
 from utils.input_pause_resume import (
     assert_pause_resume_cycles,
     assert_shutdown_while_paused,
@@ -1584,6 +1585,78 @@ def test_in_opentelemetry_grpc_oauth2_accepts_valid_jwt(case):
     assert len(response_payload["resourceLogs"]) > 0
 
 
+@pytest.mark.parametrize("field", ["kid", "alg"])
+@pytest.mark.parametrize("case", PROTOCOL_CASES, ids=[case["id"] for case in PROTOCOL_CASES])
+def test_in_opentelemetry_oauth2_rejects_duplicate_header(monkeypatch, case, field):
+    jwks_requests = track_jwks_requests(monkeypatch)
+    token = duplicate_header_token(MOCK_VALID_JWT, field)
+    service = Service(
+        IN_OPENTELEMETRY_OAUTH2_PROTOCOL_CONFIGS[case["config_key"]],
+        use_auth_server=True,
+    )
+    service.start()
+    try:
+        scheme = "https" if case["use_tls"] else "http"
+        payload = service.build_otel_payload("test_logs_001.in.json", "logs")
+        for _ in range(8):
+            result = run_curl_request(
+                f"{scheme}://localhost:{service.flb_listener_port}/v1/logs",
+                payload,
+                headers=["Content-Type: application/x-protobuf", f"Authorization: Bearer {token}"],
+                http_mode=case["http_mode"],
+                ca_cert_path=service.tls_crt_file if case["use_tls"] else None,
+            )
+            assert result["status_code"] == 401
+        assert jwks_requests == []
+        assert data_storage["logs"] == []
+
+        result = run_curl_request(
+            f"{scheme}://localhost:{service.flb_listener_port}/v1/logs",
+            payload,
+            headers=["Content-Type: application/x-protobuf", f"Authorization: Bearer {MOCK_VALID_JWT}"],
+            http_mode=case["http_mode"],
+            ca_cert_path=service.tls_crt_file if case["use_tls"] else None,
+        )
+        assert result["status_code"] == 201
+        assert service.read_response("logs")["resourceLogs"]
+    finally:
+        service.stop()
+
+
+@pytest.mark.parametrize("field", ["kid", "alg"])
+@pytest.mark.parametrize(
+    "case", IN_OPENTELEMETRY_GRPC_OAUTH2_CASES,
+    ids=[case["id"] for case in IN_OPENTELEMETRY_GRPC_OAUTH2_CASES],
+)
+def test_in_opentelemetry_grpc_oauth2_rejects_duplicate_header(monkeypatch, case, field):
+    jwks_requests = track_jwks_requests(monkeypatch)
+    token = duplicate_header_token(MOCK_VALID_JWT, field)
+    service = Service(case["config_file"], use_auth_server=True)
+    service.start()
+    try:
+        payload = json_format.Parse(
+            json.dumps(read_json_file(os.path.join(
+                os.path.dirname(__file__), "data_files/test_logs_001.in.json"
+            ))),
+            ExportLogsServiceRequest(),
+        )
+        for _ in range(8):
+            with pytest.raises(grpc.RpcError) as exc_info:
+                service.send_grpc_request(
+                    "logs", payload, authorization=f"Bearer {token}", use_tls=case["use_tls"],
+                )
+            assert exc_info.value.code() == grpc.StatusCode.UNAUTHENTICATED
+        assert jwks_requests == []
+        assert data_storage["logs"] == []
+
+        service.send_grpc_request(
+            "logs", payload, authorization=f"Bearer {MOCK_VALID_JWT}", use_tls=case["use_tls"],
+        )
+        assert service.read_response("logs")["resourceLogs"]
+    finally:
+        service.stop()
+
+
 def test_out_opentelemetry_receiver_error_is_observable():
     service = Service("001-fluent-bit.yaml")
     service.start()

tests/integration/src/utils/oauth2.py

@@ -0,0 +1,38 @@
+import base64
+import json
+
+from server.http_server import app, MOCK_JWKS_BODY
+
+
+def duplicate_header_token(token, field):
+    """Preserve duplicate members, which a Python dict would discard.
+
+    The signature is unchanged. Tests must assert that no JWKS fetch occurs
+    to distinguish header rejection from signature verification failure.
+    """
+    header = '{"alg":"RS256","kid":"test"'
+    value = "test" if field == "kid" else "RS256"
+    header += ("," + json.dumps(field) + ":" + json.dumps(value)) * 32 + "}"
+    encoded = base64.urlsafe_b64encode(header.encode()).rstrip(b"=").decode()
+    return encoded + "." + token.split(".", 1)[1]
+
+
+def track_jwks_requests(monkeypatch, body=None):
+    """Track fetches and optionally serve raw JSON containing duplicate members."""
+    requests_seen = []
+    if body is None:
+        body = json.dumps(MOCK_JWKS_BODY)
+
+    def jwks():
+        requests_seen.append(True)
+        return body, 200, {"Content-Type": "application/json"}
+
+    monkeypatch.setitem(app.view_functions, "jwks", jwks)
+    return requests_seen
+
+
+def duplicate_jwks_body(field, first_value):
+    key = MOCK_JWKS_BODY["keys"][0]
+    # The final value is correct, so the original last-wins parser accepts it.
+    prefix = json.dumps(field) + ":" + json.dumps(first_value)
+    return '{"keys":[{' + prefix + "," + json.dumps(key)[1:] + "]}"