From 6babb9819283c3d8f12a1a0897556ba305a0ce8b Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 19 Oct 2026 13:33:43 +0900 Subject: [PATCH 1/3] oauth2_jwt: reject duplicate JWT header and JWKS key fields Signed-off-by: Hiroshi Hatake --- src/flb_oauth2_jwt.c | 106 +++++++++++++++++++++++-------------------- 1 file changed, 58 insertions(+), 48 deletions(-) diff --git a/src/flb_oauth2_jwt.c b/src/flb_oauth2_jwt.c index 57557c7c1e1..314aff3401a 100644 --- a/src/flb_oauth2_jwt.c +++ b/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) From ab0ea7f8765890e455ed9687e976692a2b9acfc5 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 19 Oct 2026 13:33:55 +0900 Subject: [PATCH 2/3] tests: internal: cover duplicate and malformed JWT header fields Signed-off-by: Hiroshi Hatake --- tests/internal/oauth2_jwt.c | 64 +++++++++++++++++++++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/tests/internal/oauth2_jwt.c b/tests/internal/oauth2_jwt.c index 4f3d7c0b6b3..34efce849d4 100644 --- a/tests/internal/oauth2_jwt.c +++ b/tests/internal/oauth2_jwt.c @@ -113,6 +113,69 @@ static void test_bad_alg() TEST_CHECK(ret == FLB_OAUTH2_JWT_ERR_ALG_UNSUPPORTED); } +static void test_header_fields() +{ + const char *headers[] = { + "{\"alg\":\"RS256\",\"kid\":\"first\",\"kid\":\"test-key\"}", + "{\"kid\":\"test-key\",\"alg\":\"HS256\",\"alg\":\"RS256\"}", + "{\"alg\":\"RS256\",\"kid\":\"test-key\",\"kid\":\"test-key\"}", + "{\"kid\":\"test-key\",\"alg\":\"RS256\",\"alg\":\"RS256\"}", + "{\"alg\":\"RS256\",\"kid\":\"\",\"kid\":\"test-key\"}", + "{\"kid\":\"test-key\",\"alg\":\"\",\"alg\":\"RS256\"}", + "{\"alg\":\"RS256\",\"kid\":null,\"kid\":\"test-key\"}", + "{\"alg\":\"RS256\",\"kid\":\"test-key\",\"kid\":null}", + "{\"kid\":\"test-key\",\"alg\":null,\"alg\":\"RS256\"}", + "{\"kid\":\"test-key\",\"alg\":\"RS256\",\"alg\":null}", + "{\"alg\":\"RS256\",\"kid\":\"test-key\",\"k\\u0069d\":\"test-key\"}", + "{\"kid\":\"test-key\",\"alg\":\"RS256\",\"\\u0061lg\":\"RS256\"}", + "{\"alg\":\"RS256\",\"kid\":42}", + "{\"kid\":\"test-key\",\"alg\":[]}", + "{\"kid\":{},\"alg\":\"RS256\"}", + "{\"alg\":\"RS256\",\"kid\":\"test-key\",\"extra\":{\"kid\":\"nested\"}}" + }; + unsigned char encoded[512]; + char token[1024]; + const char *suffix; + size_t encoded_len; + size_t i; + size_t j; + int ret; + struct flb_oauth2_jwt jwt; + + suffix = strchr(VALID_JWT, '.'); + for (i = 0; i < sizeof(headers) / sizeof(headers[0]); i++) { + ret = flb_base64_encode(encoded, sizeof(encoded), &encoded_len, + (const unsigned char *) headers[i], strlen(headers[i])); + TEST_ASSERT(ret == 0); + while (encoded_len > 0 && encoded[encoded_len - 1] == '=') { + encoded_len--; + } + for (j = 0; j < encoded_len; j++) { + if (encoded[j] == '+') { + encoded[j] = '-'; + } + else if (encoded[j] == '/') { + encoded[j] = '_'; + } + } + encoded[encoded_len] = '\0'; + snprintf(token, sizeof(token), "%s%s", encoded, suffix); + + ret = flb_oauth2_jwt_parse(token, strlen(token), &jwt); + if (i == sizeof(headers) / sizeof(headers[0]) - 1) { + TEST_CHECK(ret == FLB_OAUTH2_JWT_OK); + } + else { + TEST_CHECK(ret == FLB_OAUTH2_JWT_ERR_JSON_HEADER); + } + TEST_MSG("header: %s, status: %d", headers[i], ret); + /* Parse failures already release all JWT allocations. */ + if (ret == FLB_OAUTH2_JWT_OK) { + flb_oauth2_jwt_destroy(&jwt); + } + } +} + static void test_static_key_validation() { int ret; @@ -267,6 +330,7 @@ TEST_LIST = { {"bad_base64", test_bad_base64}, {"missing_kid", test_missing_kid}, {"bad_alg", test_bad_alg}, + {"header_fields", test_header_fields}, {"static_key_validation", test_static_key_validation}, {0} }; From e80ab82cd94b51cbff87b9704969bd58b3a9774f Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 19 Oct 2026 13:33:56 +0900 Subject: [PATCH 3/3] tests: integration: cover duplicate JWT and JWKS field rejection Signed-off-by: Hiroshi Hatake --- .../in_http/tests/test_in_http_001.py | 53 ++++++++++++++ .../tests/test_in_opentelemetry_001.py | 73 +++++++++++++++++++ tests/integration/src/utils/oauth2.py | 38 ++++++++++ 3 files changed, 164 insertions(+) create mode 100644 tests/integration/src/utils/oauth2.py diff --git a/tests/integration/scenarios/in_http/tests/test_in_http_001.py b/tests/integration/scenarios/in_http/tests/test_in_http_001.py index bc66c795d83..94985210d5b 100644 --- a/tests/integration/scenarios/in_http/tests/test_in_http_001.py +++ b/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)) diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index 4c846a37465..75c5b08f4a1 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/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() diff --git a/tests/integration/src/utils/oauth2.py b/tests/integration/src/utils/oauth2.py new file mode 100644 index 00000000000..88224d467d6 --- /dev/null +++ b/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:] + "]}"