From dc2459ac7d82df1930c4c2111eca5ee86803d061 Mon Sep 17 00:00:00 2001 From: kimonus Date: Tue, 4 Aug 2026 19:51:08 +0200 Subject: [PATCH 1/8] http_server: propagate HTTP/1 parser errors The Monkey parser fix was merged upstream in monkey/monkey#444 and is bundled in Fluent Bit by #12212 as part of Monkey 1.8.9. Propagate MK_HTTP_PARSER_ERROR to the HTTP server provider so malformed requests are closed instead of resetting the parser and remaining pending. Keep this Fluent Bit-specific error handling separate from the bundled Monkey sources. Signed-off-by: kimonus --- src/http_server/flb_http_server_http1.c | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/http_server/flb_http_server_http1.c b/src/http_server/flb_http_server_http1.c index e9d930b7706..5c247c299b9 100644 --- a/src/http_server/flb_http_server_http1.c +++ b/src/http_server/flb_http_server_http1.c @@ -566,6 +566,12 @@ int flb_http1_server_session_ingest(struct flb_http1_server_session *session, * performance overhead in exchange for ensuring safety. */ } + else if (result == MK_HTTP_PARSER_ERROR) { + /* The caller closes the session when a parser error is returned. */ + session->stream.status = HTTP_STREAM_STATUS_ERROR; + + return HTTP_SERVER_PROVIDER_ERROR; + } dummy_mk_http_request_init(&session->inner_session, &session->inner_request); mk_http_parser_init(&session->inner_parser); From 7cdf3db6dd375b797a9c6cdfbe6cf50035a69156 Mon Sep 17 00:00:00 2001 From: kimonus Date: Tue, 4 Aug 2026 19:51:30 +0200 Subject: [PATCH 2/8] tests: runtime: cover empty HTTP header values Send real HTTP/1 requests containing empty and whitespace-only generic headers and require the accepted request to reach the input callback. Also verify that an invalid empty Upgrade field is not ingested. Use portable socket types, handle partial writes and reads, and reject timeouts or incomplete HTTP status lines so the regression runs on the supported runtime-test platforms. The test exercises Monkey 1.8.9 now bundled in Fluent Bit by #12212. Signed-off-by: kimonus --- tests/runtime/in_http.c | 170 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 167 insertions(+), 3 deletions(-) diff --git a/tests/runtime/in_http.c b/tests/runtime/in_http.c index ba745e24ed2..f9cdd1b0851 100644 --- a/tests/runtime/in_http.c +++ b/tests/runtime/in_http.c @@ -413,6 +413,158 @@ static void test_ctx_destroy(struct test_ctx *ctx) flb_free(ctx); } +static int send_raw_http_request(const char *request, + char *response, + size_t response_size) +{ + struct sockaddr_in address; + struct timeval timeout; + fd_set read_fds; + flb_sockfd_t fd; + size_t request_length; + size_t request_offset; + int ret; + int written; + int received; + int response_length; + + if (response_size < 2) { + return -1; + } + + response[0] = '\0'; + request_length = strlen(request); + fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd == FLB_INVALID_SOCKET) { + return -1; + } + + memset(&address, 0, sizeof(address)); + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + address.sin_port = htons(9880); + + ret = connect(fd, (struct sockaddr *) &address, sizeof(address)); + if (ret != 0) { + flb_socket_close(fd); + return -1; + } + + request_offset = 0; + while (request_offset < request_length) { + written = send(fd, + request + request_offset, + (int) (request_length - request_offset), + 0); + if (written <= 0) { + flb_socket_close(fd); + return -1; + } + request_offset += (size_t) written; + } + + response_length = 0; + while (response_length < (int) response_size - 1) { + FD_ZERO(&read_fds); + FD_SET(fd, &read_fds); + timeout.tv_sec = 2; + timeout.tv_usec = 0; + + ret = select((int) (fd + 1), &read_fds, NULL, NULL, &timeout); + if (ret <= 0) { + flb_socket_close(fd); + return -1; + } + + received = recv(fd, + response + response_length, + (int) response_size - 1 - response_length, + 0); + if (received <= 0) { + break; + } + + response_length += received; + response[response_length] = '\0'; + if (strstr(response, "\r\n") != NULL) { + break; + } + } + + flb_socket_close(fd); + response[response_length] = '\0'; + + if (response_length > 0 && strstr(response, "\r\n") == NULL) { + return -1; + } + + return response_length; +} + +static int send_empty_header_request(void) +{ + const char *body = "{\"test\":\"msg\"}"; + const char *request_template = + "POST / HTTP/1.1\r\n" + "Host: 127.0.0.1\r\n" + "Content-Length: %zu\r\n" + "Content-Type: application/json\r\n" + "X-Empty:\r\n" + "X-Empty-Whitespace: \t\r\n" + "\r\n" + "%s"; + char request[512]; + char response[256]; + int ret; + int response_length; + + ret = snprintf(request, sizeof(request), request_template, + strlen(body), body); + if (ret <= 0 || (size_t) ret >= sizeof(request)) { + return -1; + } + + response_length = send_raw_http_request(request, response, sizeof(response)); + if (response_length <= 0) { + return -1; + } + + return strstr(response, "HTTP/1.1 201") != NULL ? 0 : -1; +} + +static int send_invalid_header_request(void) +{ + const char *body = "{\"test\":\"msg\"}"; + const char *request_template = + "POST / HTTP/1.1\r\n" + "Host: 127.0.0.1\r\n" + "Content-Length: %zu\r\n" + "Content-Type: application/json\r\n" + "Upgrade:\r\n" + "\r\n" + "%s"; + char request[512]; + char response[256]; + int ret; + int response_length; + + ret = snprintf(request, sizeof(request), request_template, + strlen(body), body); + if (ret <= 0 || (size_t) ret >= sizeof(request)) { + return -1; + } + + response_length = send_raw_http_request(request, response, sizeof(response)); + if (response_length < 0) { + return -1; + } + if (response_length == 0) { + return 0; + } + + return strstr(response, "HTTP/1.1 4") != NULL ? 0 : -1; +} + void flb_test_http(void) { struct flb_lib_out_cb cb_data; @@ -469,13 +621,25 @@ void flb_test_http(void) TEST_MSG("http response code error. expect: 201, got: %d\n", c->resp.status); } + /* Ensure the first request has reached the callback before the regression. */ + flb_time_msleep(1500); + num = get_output_num(); + TEST_CHECK(num > 0); + + TEST_CHECK(send_empty_header_request() == 0); + /* waiting to flush */ flb_time_msleep(1500); - num = get_output_num(); - if (!TEST_CHECK(num > 0)) { - TEST_MSG("no outputs"); + if (!TEST_CHECK(get_output_num() > num)) { + TEST_MSG("empty-header request did not reach the callback"); } + + num = get_output_num(); + TEST_CHECK(send_invalid_header_request() == 0); + flb_time_msleep(500); + TEST_CHECK(get_output_num() == num); + flb_http_client_destroy(c); flb_upstream_conn_release(ctx->httpc->u_conn); test_ctx_destroy(ctx); From beab797f915f2c210187301fccc9d18ed4ed1ad8 Mon Sep 17 00:00:00 2001 From: kimonus Date: Tue, 4 Aug 2026 19:51:45 +0200 Subject: [PATCH 3/8] tests: integration: cover empty HTTP header values Add real-server POST coverage matching #12174 for empty and whitespace-only generic fields, and verify that the request body is forwarded successfully. Cover empty Connection and Transfer-Encoding values plus empty and whitespace-only Content-Length fields followed immediately by numeric header or body data. Rejected requests must close or return 400, never hang or forward a payload. These tests exercise Monkey 1.8.9 now bundled in Fluent Bit by #12212. Signed-off-by: kimonus --- .../in_http/tests/test_in_http_001.py | 105 ++++++++++++++++++ 1 file changed, 105 insertions(+) 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 a82d65307ea..d7939e50c45 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 @@ -2,6 +2,7 @@ import json import os import logging +import socket import time import pytest @@ -66,6 +67,28 @@ def send_requests(conn, num_requests, headers, json_payload): return responses +def send_raw_http1_request(port, request): + response = bytearray() + + with socket.create_connection(("127.0.0.1", port), timeout=2) as connection: + connection.settimeout(2) + connection.sendall(request) + + while len(response) < 4096 and b"\r\n" not in response: + try: + data = connection.recv(4096 - len(response)) + except ConnectionResetError: + break + except socket.timeout: + pytest.fail("HTTP/1 server did not respond or close the connection") + + if not data: + break + response.extend(data) + + return bytes(response) + + def test_send_data(): try: service = Service("in_http_config") @@ -150,6 +173,88 @@ def test_in_http_rejects_get_requests(): assert result["status_code"] >= 400 +def test_in_http_accepts_post_with_empty_generic_headers(): + service = Service("in_http_config") + body = b'{"message":"empty-header"}' + request = ( + b"POST / HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Type: application/json\r\n" + + f"Content-Length: {len(body)}\r\n".encode() + + b"X-Empty:\r\n" + b"X-Empty-Whitespace: \t\r\n" + b"Connection: close\r\n" + b"\r\n" + + body + ) + + try: + service.start() + response = send_raw_http1_request(service.flb_listener_port, request) + forwarded_payloads = service.read_forwarded_payloads() + finally: + service.stop() + + assert b"HTTP/1.1 201" in response + assert len(forwarded_payloads) == 1 + assert forwarded_payloads[0][0]["message"] == "empty-header" + + +def test_in_http_accepts_empty_connection_and_transfer_encoding(): + service = Service("in_http_config") + body = b'{"message":"empty-semantic-headers"}' + request = ( + b"POST / HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Type: application/json\r\n" + + f"Content-Length: {len(body)}\r\n".encode() + + b"Connection:\r\n" + b"Transfer-Encoding: \t\r\n" + b"Connection: close\r\n" + b"\r\n" + + body + ) + + try: + service.start() + response = send_raw_http1_request(service.flb_listener_port, request) + forwarded_payloads = service.read_forwarded_payloads() + finally: + service.stop() + + assert b"HTTP/1.1 201" in response + assert len(forwarded_payloads) == 1 + assert forwarded_payloads[0][0]["message"] == "empty-semantic-headers" + + +@pytest.mark.parametrize("header_value", [b"", b" \t"], ids=["empty", "whitespace"]) +@pytest.mark.parametrize( + "following_data", + [b"1-X: value\r\nConnection: close\r\n\r\n1", b"\r\n1"], + ids=["numeric-header", "numeric-body"], +) +def test_in_http_rejects_empty_content_length(header_value, following_data): + service = Service("in_http_config") + request = ( + b"POST / HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Type: application/json\r\n" + b"Content-Length:" + header_value + b"\r\n" + + following_data + ) + + try: + service.start() + response = send_raw_http1_request(service.flb_listener_port, request) + time.sleep(0.5) + forwarded_payloads = list(data_storage["payloads"]) + finally: + service.stop() + + assert response == b"" or b"HTTP/1.1 400" in response + assert forwarded_payloads == [] + + @pytest.mark.parametrize( "case", [ From 44cab5dd9846ab6242da8a55fb23f857b29b443a Mon Sep 17 00:00:00 2001 From: kimonus Date: Tue, 4 Aug 2026 20:30:07 +0200 Subject: [PATCH 4/8] tests: runtime: observe invalid-request flush window Wait for the same dispatch and flush observation window used by accepted raw requests before confirming that malformed requests do not reach the output callback. Signed-off-by: kimonus --- tests/runtime/in_http.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/runtime/in_http.c b/tests/runtime/in_http.c index f9cdd1b0851..989efd48b85 100644 --- a/tests/runtime/in_http.c +++ b/tests/runtime/in_http.c @@ -637,7 +637,8 @@ void flb_test_http(void) num = get_output_num(); TEST_CHECK(send_invalid_header_request() == 0); - flb_time_msleep(500); + /* Observe the same full dispatch and flush window as accepted requests. */ + flb_time_msleep(1500); TEST_CHECK(get_output_num() == num); flb_http_client_destroy(c); From 4b801a0830c6d34b49ca1e4ba67373bcb274a2c5 Mon Sep 17 00:00:00 2001 From: kimonus Date: Tue, 4 Aug 2026 20:34:08 +0200 Subject: [PATCH 5/8] tests: integration: observe invalid-request flush window Observe rejected requests for longer than one configured output flush interval, then check again after shutdown so delayed forwarding cannot make the regression pass incorrectly. Signed-off-by: kimonus --- .../scenarios/in_http/tests/test_in_http_001.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) 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 d7939e50c45..17a5cfc99de 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 @@ -246,11 +246,11 @@ def test_in_http_rejects_empty_content_length(header_value, following_data): try: service.start() response = send_raw_http1_request(service.flb_listener_port, request) - time.sleep(0.5) - forwarded_payloads = list(data_storage["payloads"]) + service.assert_no_forwarded_payloads_for() finally: service.stop() + forwarded_payloads = list(data_storage["payloads"]) assert response == b"" or b"HTTP/1.1 400" in response assert forwarded_payloads == [] @@ -510,5 +510,12 @@ def read_forwarded_payloads(self, timeout=10): time.sleep(0.5) raise TimeoutError("Timed out waiting for forwarded HTTP payloads") + def assert_no_forwarded_payloads_for(self, quiet_period=1.5): + deadline = time.time() + quiet_period + + while time.time() < deadline: + assert data_storage["payloads"] == [] + time.sleep(0.1) + def stop(self): self.service.stop() From 608dd5b1d8fd0039c4fb473379f9b825f9a1ebb8 Mon Sep 17 00:00:00 2001 From: kimonus Date: Tue, 4 Aug 2026 23:59:07 +0200 Subject: [PATCH 6/8] http_server: wait for complete HTTP/2 preface Keep protocol autodetection pending while a split HTTP/2 preface is incomplete. Feed all buffered bytes to nghttp2 after detection completes. Signed-off-by: kimonus --- src/http_server/flb_http_server.c | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/http_server/flb_http_server.c b/src/http_server/flb_http_server.c index 6f8e1ef6bc5..c80cfd02bc6 100644 --- a/src/http_server/flb_http_server.c +++ b/src/http_server/flb_http_server.c @@ -1399,7 +1399,11 @@ int flb_http_server_session_ingest(struct flb_http_server_session *session, } } - if (session->version <= HTTP_PROTOCOL_VERSION_11) { + if (session->version == HTTP_PROTOCOL_VERSION_AUTODETECT) { + /* Wait for the remainder of a split HTTP/2 connection preface. */ + return HTTP_SERVER_SUCCESS; + } + else if (session->version <= HTTP_PROTOCOL_VERSION_11) { result = flb_http1_server_session_init(&session->http1, session); if (result != 0) { @@ -1412,6 +1416,10 @@ int flb_http_server_session_ingest(struct flb_http_server_session *session, if (result != 0) { return -1; } + + /* Protocol detection may have accumulated the preface over multiple reads. */ + buffer = (unsigned char *) session->incoming_data; + length = cfl_sds_len(session->incoming_data); } } From 5eeaf9dd6fca37cdbcc3e5c6384c58627009adbd Mon Sep 17 00:00:00 2001 From: kimonus Date: Tue, 4 Aug 2026 23:59:36 +0200 Subject: [PATCH 7/8] tests: integration: cover split HTTP/2 preface Send the client connection preface across separate reads. Verify that the server initializes HTTP/2 and returns its SETTINGS frame. Signed-off-by: kimonus --- .../in_http/tests/test_in_http_001.py | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) 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 17a5cfc99de..dda7c848c57 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 @@ -89,6 +89,30 @@ def send_raw_http1_request(port, request): return bytes(response) +def send_split_http2_preface(port): + preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + settings_frame = b"\x00\x00\x00\x04\x00\x00\x00\x00\x00" + response = bytearray() + + with socket.create_connection(("127.0.0.1", port), timeout=2) as connection: + connection.settimeout(2) + connection.sendall(preface[:-2]) + time.sleep(0.1) + connection.sendall(preface[-2:] + settings_frame) + + while len(response) < 9: + try: + data = connection.recv(4096) + except socket.timeout: + pytest.fail("HTTP/2 server did not respond to a split connection preface") + + if not data: + break + response.extend(data) + + return bytes(response) + + def test_send_data(): try: service = Service("in_http_config") @@ -141,6 +165,20 @@ def test_in_http_protocol_matrix(case): assert forwarded_payloads[0][0]["message"] == "Este es un mensaje de prueba" +def test_in_http_accepts_split_http2_preface(): + service = Service("in_http_http2_cleartext.yaml") + + try: + service.start() + response = send_split_http2_preface(service.flb_listener_port) + finally: + service.stop() + + assert len(response) >= 9 + assert response[3] == 0x04 + assert data_storage["payloads"] == [] + + def test_in_http_rejects_bad_json(): service = Service("in_http_config") service.start() From 3e3214be7eeb52eb960391e9003e0fc85eecd25b Mon Sep 17 00:00:00 2001 From: kimonus Date: Wed, 5 Aug 2026 09:37:45 +0200 Subject: [PATCH 8/8] tests: integration: verify split protocol detection Make split request boundaries observable before sending the remainder. Cover HTTP/1 detection when the first read is shorter than four bytes. Signed-off-by: kimonus --- .../in_http/tests/test_in_http_001.py | 53 +++++++++++++++++-- 1 file changed, 50 insertions(+), 3 deletions(-) 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 dda7c848c57..063cd192975 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 @@ -67,12 +67,33 @@ def send_requests(conn, num_requests, headers, json_payload): return responses -def send_raw_http1_request(port, request): +def assert_connection_open_without_response(connection): + connection.settimeout(1) + + try: + data = connection.recv(1) + except socket.timeout: + return + + if not data: + pytest.fail("HTTP server closed the connection while the request was incomplete") + + pytest.fail("HTTP server responded before the request was complete") + + +def send_raw_http1_request(port, request, split_at=None): response = bytearray() with socket.create_connection(("127.0.0.1", port), timeout=2) as connection: connection.settimeout(2) - connection.sendall(request) + + if split_at is None: + connection.sendall(request) + else: + connection.sendall(request[:split_at]) + assert_connection_open_without_response(connection) + connection.settimeout(2) + connection.sendall(request[split_at:]) while len(response) < 4096 and b"\r\n" not in response: try: @@ -97,7 +118,8 @@ def send_split_http2_preface(port): with socket.create_connection(("127.0.0.1", port), timeout=2) as connection: connection.settimeout(2) connection.sendall(preface[:-2]) - time.sleep(0.1) + assert_connection_open_without_response(connection) + connection.settimeout(2) connection.sendall(preface[-2:] + settings_frame) while len(response) < 9: @@ -179,6 +201,31 @@ def test_in_http_accepts_split_http2_preface(): assert data_storage["payloads"] == [] +def test_in_http_accepts_http1_request_split_before_autodetect_boundary(): + service = Service("in_http_http2_cleartext.yaml") + body = b'{"message":"split-http1"}' + request = ( + b"POST / HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Type: application/json\r\n" + + f"Content-Length: {len(body)}\r\n".encode() + + b"Connection: close\r\n" + b"\r\n" + + body + ) + + try: + service.start() + response = send_raw_http1_request(service.flb_listener_port, request, split_at=1) + forwarded_payloads = service.read_forwarded_payloads() + finally: + service.stop() + + assert b"HTTP/1.1 201" in response + assert len(forwarded_payloads) == 1 + assert forwarded_payloads[0][0]["message"] == "split-http1" + + def test_in_http_rejects_bad_json(): service = Service("in_http_config") service.start()