From 42d163cdd228595dd5929547126f0e476b521be9 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 18:52:40 +0900 Subject: [PATCH 01/12] config: expose HTTP client response limits Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_config.h | 10 ++++++ src/flb_config.c | 54 ++++++++++++++++++++++++++++++++- 2 files changed, 63 insertions(+), 1 deletion(-) diff --git a/include/fluent-bit/flb_config.h b/include/fluent-bit/flb_config.h index ab39188197d..6ca9f93c40c 100644 --- a/include/fluent-bit/flb_config.h +++ b/include/fluent-bit/flb_config.h @@ -99,6 +99,11 @@ struct flb_config { int flush_adaptive_direction; double flush_adaptive_current_interval; + size_t http_client_max_response_size; + size_t http_client_max_header_size; + int http_client_max_header_count; + int http_client_response_timeout; + /* * Maximum grace time on shutdown. If set to -1, the engine will * shutdown when all remaining tasks are flushed @@ -415,6 +420,7 @@ struct flb_service_config { enum conf_type { FLB_CONF_TYPE_INT, + FLB_CONF_TYPE_SIZE, FLB_CONF_TYPE_DOUBLE, FLB_CONF_TYPE_BOOL, FLB_CONF_TYPE_STR, @@ -422,6 +428,10 @@ enum conf_type { }; #define FLB_CONF_STR_FLUSH "Flush" +#define FLB_CONF_STR_HTTP_CLIENT_MAX_RESPONSE_SIZE "http_client.max_response_size" +#define FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_SIZE "http_client.max_header_size" +#define FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_COUNT "http_client.max_header_count" +#define FLB_CONF_STR_HTTP_CLIENT_RESPONSE_TIMEOUT "http_client.response_timeout" #define FLB_CONF_STR_FLUSH_ADAPTIVE "flush.adaptive" #define FLB_CONF_STR_FLUSH_ADAPTIVE_MIN "flush.adaptive.min_interval" #define FLB_CONF_STR_FLUSH_ADAPTIVE_MAX "flush.adaptive.max_interval" diff --git a/src/flb_config.c b/src/flb_config.c index 8427a3d74f5..4007176c6fc 100644 --- a/src/flb_config.c +++ b/src/flb_config.c @@ -54,6 +54,22 @@ const char *FLB_CONF_ENV_LOGLEVEL = "FLB_LOG_LEVEL"; int flb_regex_init(); struct flb_service_config service_configs[] = { + {FLB_CONF_STR_HTTP_CLIENT_MAX_RESPONSE_SIZE, + FLB_CONF_TYPE_SIZE, + offsetof(struct flb_config, http_client_max_response_size)}, + + {FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_SIZE, + FLB_CONF_TYPE_SIZE, + offsetof(struct flb_config, http_client_max_header_size)}, + + {FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_COUNT, + FLB_CONF_TYPE_INT, + offsetof(struct flb_config, http_client_max_header_count)}, + + {FLB_CONF_STR_HTTP_CLIENT_RESPONSE_TIMEOUT, + FLB_CONF_TYPE_INT, + offsetof(struct flb_config, http_client_response_timeout)}, + {FLB_CONF_STR_FLUSH, FLB_CONF_TYPE_DOUBLE, offsetof(struct flb_config, flush)}, @@ -321,6 +337,10 @@ struct flb_config *flb_config_init() /* Flush */ config->flush = FLB_CONFIG_FLUSH_SECS; + config->http_client_max_response_size = 4 * 1024 * 1024; + config->http_client_max_header_size = 65536; + config->http_client_max_header_count = 100; + config->http_client_response_timeout = 30; config->flush_adaptive = FLB_FALSE; config->flush_adaptive_min_interval = 0.5; config->flush_adaptive_max_interval = 2.0; @@ -858,6 +878,10 @@ int flb_config_set_property(struct flb_config *config, int i=0; int ret = -1; int *i_val; + size_t *size_val; + int64_t parsed_size; + long parsed_int; + char *endptr; double *d_val; char **s_val; size_t len = strnlen(k, 256); @@ -912,8 +936,36 @@ int flb_config_set_property(struct flb_config *config, switch(service_configs[i].type) { case FLB_CONF_TYPE_INT: i_val = (int*)((char*)config + service_configs[i].offset); - *i_val = atoi(tmp); + if (strncasecmp(key, "http_client.", 12) == 0) { + errno = 0; + parsed_int = strtol(tmp, &endptr, 10); + if (errno != 0 || endptr == tmp || *endptr != '\0' || + parsed_int <= 0 || parsed_int > INT_MAX) { + ret = -1; + } + else { + *i_val = (int) parsed_int; + } + } + else { + *i_val = atoi(tmp); + } + flb_sds_destroy(tmp); + tmp = NULL; + break; + case FLB_CONF_TYPE_SIZE: + parsed_size = flb_utils_size_to_bytes(tmp); + if (parsed_size <= 0 || parsed_size > UINT32_MAX || + (strcmp(key, FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_SIZE) == 0 && + parsed_size < 32)) { + ret = -1; + } + else { + size_val = (size_t *) ((char *) config + service_configs[i].offset); + *size_val = (size_t) parsed_size; + } flb_sds_destroy(tmp); + tmp = NULL; break; case FLB_CONF_TYPE_DOUBLE: d_val = (double*)((char*)config + service_configs[i].offset); From e6ce2d9193f607c3ac3d23bc56da45a41d0a48fa Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 18:52:46 +0900 Subject: [PATCH 02/12] http_client: bound response size and deadline Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_http_client.h | 15 ++++++ src/flb_http_client.c | 70 ++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/include/fluent-bit/flb_http_client.h b/include/fluent-bit/flb_http_client.h index 613f20efafb..d4c15225a76 100644 --- a/include/fluent-bit/flb_http_client.h +++ b/include/fluent-bit/flb_http_client.h @@ -42,6 +42,12 @@ /* Buffer size */ #define FLB_HTTP_BUF_SIZE 2048 #define FLB_HTTP_DATA_SIZE_MAX 4096 + +/* Maximum response bytes received by a client-ng session. */ +#define FLB_HTTP_CLIENT_NG_RESPONSE_SIZE_MAX (4 * 1024 * 1024) +#define FLB_HTTP_CLIENT_NG_HEADER_SIZE_MAX 65536 +#define FLB_HTTP_CLIENT_NG_HEADER_COUNT_MAX 100 +#define FLB_HTTP_CLIENT_NG_RESPONSE_TIMEOUT 30 #define FLB_HTTP_DATA_CHUNK 32768 /* HTTP Methods */ @@ -276,6 +282,10 @@ struct flb_http_client_ng { uint16_t port; uint64_t flags; int protocol_version; + size_t max_response_size; + size_t max_header_size; + size_t max_header_count; + int response_timeout; cfl_sds_t temporary_buffer; int releasable; @@ -293,6 +303,11 @@ struct flb_http_client_session { struct cfl_list streams; int protocol_version; + size_t received_bytes; + int response_limit_exceeded; + time_t response_start; + struct flb_net_setup request_net_setup; + struct flb_net_setup *original_net_setup; cfl_sds_t incoming_data; cfl_sds_t outgoing_data; diff --git a/src/flb_http_client.c b/src/flb_http_client.c index b625a99d1f0..67b0518eb30 100644 --- a/src/flb_http_client.c +++ b/src/flb_http_client.c @@ -2385,6 +2385,10 @@ int flb_http_client_ng_init(struct flb_http_client_ng *client, } client->protocol_version = protocol_version; + client->max_response_size = FLB_HTTP_CLIENT_NG_RESPONSE_SIZE_MAX; + client->max_header_size = FLB_HTTP_CLIENT_NG_HEADER_SIZE_MAX; + client->max_header_count = FLB_HTTP_CLIENT_NG_HEADER_COUNT_MAX; + client->response_timeout = FLB_HTTP_CLIENT_NG_RESPONSE_TIMEOUT; client->upstream_ha = upstream_ha; client->upstream = upstream; client->flags = flags; @@ -2700,6 +2704,14 @@ void flb_http_client_session_destroy(struct flb_http_client_session *session) } if (session->connection != NULL) { + if (session->original_net_setup != NULL) { + session->connection->net = session->original_net_setup; + if (session->connection->fd > 0) { + flb_net_socket_set_rcvtimeout( + session->connection->fd, + session->original_net_setup->io_timeout); + } + } flb_upstream_conn_release(session->connection); } @@ -2792,6 +2804,35 @@ struct flb_http_response *flb_http_client_request_execute_step( return response; } + if (session->response_start == 0) { + session->response_start = time(NULL); + session->original_net_setup = session->connection->net; + if (session->original_net_setup != NULL) { + session->request_net_setup = *session->original_net_setup; + session->connection->net = &session->request_net_setup; + } + } + + result = session->parent->response_timeout - + (int) (time(NULL) - session->response_start); + if (result <= 0) { + flb_upstream_conn_recycle(session->connection, FLB_FALSE); + request->stream->status = HTTP_STREAM_STATUS_ERROR; + return NULL; + } + + if (session->original_net_setup != NULL) { + if (session->original_net_setup->io_timeout > 0 && + session->original_net_setup->io_timeout < result) { + result = session->original_net_setup->io_timeout; + } + + session->request_net_setup.io_timeout = result; + if (session->connection->fd > 0) { + flb_net_socket_set_rcvtimeout(session->connection->fd, result); + } + } + if (session->outgoing_data != NULL && cfl_sds_len(session->outgoing_data) > 0) { @@ -2885,6 +2926,7 @@ static int flb_http_client_session_read(struct flb_http_client_session *session) result); if (result < 0) { + flb_upstream_conn_recycle(session->connection, FLB_FALSE); return -2; } @@ -2960,6 +3002,34 @@ int flb_http_client_session_ingest(struct flb_http_client_session *session, size_t length) { #ifndef __EMSCRIPTEN__ + struct cfl_list *iterator; + struct flb_http_stream *stream; + size_t max_wire_size; + + max_wire_size = session->parent->max_response_size; + if (session->parent->max_header_size <= SIZE_MAX - max_wire_size) { + max_wire_size += session->parent->max_header_size; + } + else { + max_wire_size = SIZE_MAX; + } + + if (session->received_bytes > max_wire_size || + length > max_wire_size - session->received_bytes) { + session->response_limit_exceeded = FLB_TRUE; + cfl_list_foreach(iterator, &session->streams) { + stream = cfl_list_entry(iterator, struct flb_http_stream, _head); + stream->status = HTTP_STREAM_STATUS_ERROR; + } + + if (session->connection != NULL) { + flb_upstream_conn_recycle(session->connection, FLB_FALSE); + } + return -1; + } + + session->received_bytes += length; + if (session->protocol_version == HTTP_PROTOCOL_VERSION_11 || session->protocol_version == HTTP_PROTOCOL_VERSION_10) { return flb_http1_client_session_ingest(&session->http1, From d2938c47199247c40a997bf2453a792ff6a1fd31 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 18:52:52 +0900 Subject: [PATCH 03/12] http_client_http1: reject oversized responses Signed-off-by: Hiroshi Hatake --- src/flb_http_client_http1.c | 46 +++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/src/flb_http_client_http1.c b/src/flb_http_client_http1.c index 64657b1d033..1e5e042a384 100644 --- a/src/flb_http_client_http1.c +++ b/src/flb_http_client_http1.c @@ -132,6 +132,11 @@ static int flb_http1_client_session_process_headers(struct flb_http_client_sessi header_block_begining = strstr(status_line, "\r\n"); if (header_block_begining == NULL) { + if (cfl_sds_len(status_line) > session->parent->max_header_size) { + session->response_limit_exceeded = FLB_TRUE; + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } return 0; } @@ -143,9 +148,21 @@ static int flb_http1_client_session_process_headers(struct flb_http_client_sessi header_block_end = strstr(header_block_begining, "\r\n\r\n"); if (header_block_end == NULL) { + if (cfl_sds_len(status_line) > session->parent->max_header_size) { + session->response_limit_exceeded = FLB_TRUE; + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } return 0; } + if ((size_t) (header_block_end - status_line) + 4 > + session->parent->max_header_size) { + session->response_limit_exceeded = FLB_TRUE; + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } + header_block_length = (size_t) ((uintptr_t) header_block_end - (uintptr_t) header_block_begining); @@ -194,6 +211,7 @@ static int flb_http1_client_session_process_data(struct flb_http_client_session size_t chunk_length; size_t required_size; + result = 0; body = session->incoming_data; body_streaming_flag = (session->parent->flags & @@ -215,6 +233,15 @@ static int flb_http1_client_session_process_data(struct flb_http_client_session if (chunked_transfer == FLB_FALSE) { if (response->content_length > 0) { + if (body_length > session->parent->max_response_size || + (response->body != NULL && + cfl_sds_len(response->body) > + session->parent->max_response_size - body_length)) { + session->response_limit_exceeded = FLB_TRUE; + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } + if ((response->body_read_offset + body_length) < response->content_length) { if (body_streaming_flag == FLB_FALSE) { @@ -266,6 +293,12 @@ static int flb_http1_client_session_process_data(struct flb_http_client_session chunk_length_length = (size_t) ((uintptr_t) chunk_length_end - (uintptr_t) chunk_header); + if (chunk_length > session->parent->max_response_size) { + session->response_limit_exceeded = FLB_TRUE; + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } + required_size = chunk_length_length + 2 + chunk_length + 2; if (body_remainder < required_size) { @@ -275,6 +308,14 @@ static int flb_http1_client_session_process_data(struct flb_http_client_session chunk_data = chunk_header + chunk_length_length + 2; if (chunk_length > 0) { + if (response->body != NULL && + cfl_sds_len(response->body) > + session->parent->max_response_size - chunk_length) { + session->response_limit_exceeded = FLB_TRUE; + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } + result = flb_http_response_append_to_body( response, (unsigned char *) chunk_data, @@ -297,6 +338,11 @@ static int flb_http1_client_session_process_data(struct flb_http_client_session } } + if (result != 0) { + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } + return 0; } From 7f3d930c92d5533bc37ec972d652b7c9134ac577 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 18:53:05 +0900 Subject: [PATCH 04/12] http_client_http2: bound headers and body Signed-off-by: Hiroshi Hatake --- src/flb_http_client_http2.c | 33 +++++++++++++++++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/src/flb_http_client_http2.c b/src/flb_http_client_http2.c index fbfd6349ea8..728158af429 100644 --- a/src/flb_http_client_http2.c +++ b/src/flb_http_client_http2.c @@ -209,6 +209,7 @@ static int http2_header_callback(nghttp2_session *inner_session, { char temporary_buffer[16]; struct flb_http_stream *stream; + struct flb_http_client_ng *client; int result; stream = nghttp2_session_get_stream_user_data(inner_session, @@ -218,6 +219,21 @@ static int http2_header_callback(nghttp2_session *inner_session, return 0; } + client = ((struct flb_http_client_session *) stream->parent)->parent; + + if (stream->incoming_header_count >= client->max_header_count || + name_length > client->max_header_size - 32 || + value_length > client->max_header_size - 32 - name_length || + stream->incoming_header_size > + client->max_header_size - 32 - name_length - value_length) { + ((struct flb_http_client_session *) stream->parent)->response_limit_exceeded = FLB_TRUE; + stream->status = HTTP_STREAM_STATUS_ERROR; + return NGHTTP2_ERR_CALLBACK_FAILURE; + } + + stream->incoming_header_count++; + stream->incoming_header_size += name_length + value_length + 32; + if (flb_http_server_strncasecmp( name, name_length, ":status", 0) == 0) { strncpy(temporary_buffer, @@ -364,6 +380,16 @@ static int http2_data_chunk_recv_callback(nghttp2_session *inner_session, return -1; } + parent_session = (struct flb_http_client_session *) stream->parent; + if (len > parent_session->parent->max_response_size || + (stream->response.body != NULL && + cfl_sds_len(stream->response.body) > + parent_session->parent->max_response_size - len)) { + parent_session->response_limit_exceeded = FLB_TRUE; + stream->status = HTTP_STREAM_STATUS_ERROR; + return NGHTTP2_ERR_CALLBACK_FAILURE; + } + if (stream->response.body == NULL) { stream->response.body = cfl_sds_create_size(len); @@ -473,7 +499,7 @@ static ssize_t http2_data_source_read_callback(nghttp2_session *session, int flb_http2_client_session_init(struct flb_http2_client_session *session) { - nghttp2_settings_entry session_settings[3]; + nghttp2_settings_entry session_settings[4]; nghttp2_session_callbacks *callbacks; int result; @@ -514,11 +540,14 @@ int flb_http2_client_session_init(struct flb_http2_client_session *session) session_settings[2].settings_id = NGHTTP2_SETTINGS_ENABLE_PUSH; session_settings[2].value = 0; + session_settings[3].settings_id = NGHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE; + session_settings[3].value = session->parent->parent->max_header_size; + result = nghttp2_submit_settings(session->inner_session, NGHTTP2_FLAG_NONE, session_settings, - 3); + 4); if (result != 0) { return -3; From ed0118420142bfb1a0a7979de0ccea8323f26756 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 18:53:09 +0900 Subject: [PATCH 05/12] out_opentelemetry: reject oversized responses Signed-off-by: Hiroshi Hatake --- plugins/out_opentelemetry/opentelemetry.c | 14 +++++++++++--- plugins/out_opentelemetry/opentelemetry_conf.c | 7 +++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/plugins/out_opentelemetry/opentelemetry.c b/plugins/out_opentelemetry/opentelemetry.c index 4ca2d4eb382..0bb4a2d7fac 100644 --- a/plugins/out_opentelemetry/opentelemetry.c +++ b/plugins/out_opentelemetry/opentelemetry.c @@ -607,11 +607,19 @@ int opentelemetry_post(struct opentelemetry_context *ctx, response = flb_http_client_request_execute(request); if (response == NULL) { - flb_plg_warn(ctx->ins, "error performing HTTP request, remote host=%s:%i connection error", - ctx->host, ctx->port); + result = ((struct flb_http_client_session *) + request->stream->parent)->response_limit_exceeded; + if (result) { + flb_plg_error(ctx->ins, "HTTP response exceeded configured client limit, " + "remote host=%s:%i", ctx->host, ctx->port); + } + else { + flb_plg_warn(ctx->ins, "error performing HTTP request, " + "remote host=%s:%i connection error", ctx->host, ctx->port); + } flb_http_client_request_destroy(request, FLB_TRUE); - return FLB_RETRY; + return result ? FLB_ERROR : FLB_RETRY; } if (ctx->oauth2_ctx != NULL && response->status == 401) { diff --git a/plugins/out_opentelemetry/opentelemetry_conf.c b/plugins/out_opentelemetry/opentelemetry_conf.c index c09cbf9eb2a..4999285f0db 100644 --- a/plugins/out_opentelemetry/opentelemetry_conf.c +++ b/plugins/out_opentelemetry/opentelemetry_conf.c @@ -751,6 +751,13 @@ struct opentelemetry_context *flb_opentelemetry_context_create(struct flb_output http_protocol_version, http_client_flags); + if (ret == 0) { + ctx->http_client.max_response_size = config->http_client_max_response_size; + ctx->http_client.max_header_size = config->http_client_max_header_size; + ctx->http_client.max_header_count = config->http_client_max_header_count; + ctx->http_client.response_timeout = config->http_client_response_timeout; + } + if (ret != 0) { flb_plg_debug(ctx->ins, "http client creation error"); From f13a02a66818f0ed963a0ffe43f1ffe6e4c4a1ac Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 18:53:15 +0900 Subject: [PATCH 06/12] tests: integration: cover bounded OTLP responses Signed-off-by: Hiroshi Hatake --- .../out_otel_http2_http1_response_limit.yaml | 28 +++++++ .../out_otel_http2_ipv6_response_limit.yaml | 20 +++++ .../tests/test_out_opentelemetry_001.py | 81 +++++++++++++++---- 3 files changed, 115 insertions(+), 14 deletions(-) create mode 100644 tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml create mode 100644 tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml new file mode 100644 index 00000000000..658482f56ba --- /dev/null +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml @@ -0,0 +1,28 @@ +service: + flush: 1 + http_client.max_response_size: 64K + http_client.max_header_size: 16K + http_client.max_header_count: 50 + http_client.response_timeout: 10 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: response_limit + samples: 1 + + outputs: + - name: opentelemetry + match: response_limit + host: localhost + port: ${TEST_SUITE_HTTP_PORT} + http2: on + tls: on + tls.verify: on + tls.verify_hostname: on + tls.vhost: localhost + tls.ca_file: ${CERTIFICATE_TEST} + retry_limit: 1 diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml new file mode 100644 index 00000000000..3f8036a3501 --- /dev/null +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml @@ -0,0 +1,20 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + http_client.max_response_size: 16K + +pipeline: + inputs: + - name: dummy + tag: ipv6_http2 + samples: 1 + + outputs: + - name: opentelemetry + match: ipv6_http2 + host: "::1" + port: ${TEST_SUITE_HTTP_PORT} + http2: on + retry_limit: 1 diff --git a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py index 647131c6720..f870b8643c7 100644 --- a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py +++ b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py @@ -269,8 +269,10 @@ def _build_signal_payload_from_dict(self, payload_dict, signal_type): class IPv6Http2OtlpReceiver: - def __init__(self, port): + def __init__(self, port, oversized_headers=False, oversized_body=False): self.port = port + self.oversized_headers = oversized_headers + self.oversized_body = oversized_body self.server_socket = None self.thread = None self.requests = [] @@ -352,24 +354,35 @@ def _handle_connection(self, client_socket): client_socket.sendall(pending) def _send_response(self, connection, stream_id): - body = b"{}" + body = b"x" * (32 * 1024) if self.oversized_body else b"{}" - connection.send_headers( - stream_id, - [ - (":status", "200"), - ("content-type", "application/json"), - ("content-length", str(len(body))), - ], - ) - connection.send_data(stream_id, body, end_stream=True) + headers = [ + (":status", "200"), + ("content-type", "application/json"), + ("content-length", str(len(body))), + ] + if self.oversized_headers: + headers.extend((f"x-extra-{index}", "a" * 1024) for index in range(70)) + + connection.send_headers(stream_id, headers) + if self.oversized_body: + connection.send_data(stream_id, body[:16384]) + connection.send_data(stream_id, body[16384:], end_stream=True) + else: + connection.send_data(stream_id, body, end_stream=True) class Http2IPv6Service: - def __init__(self): - self.config_file = _repo_relative("../config", "out_otel_http2_ipv6_logs.yaml") + def __init__(self, oversized_headers=False, oversized_body=False): + config_name = ( + "out_otel_http2_ipv6_response_limit.yaml" + if oversized_body else "out_otel_http2_ipv6_logs.yaml" + ) + self.config_file = _repo_relative("../config", config_name) self.receiver = None self.test_suite_http_port = None + self.oversized_headers = oversized_headers + self.oversized_body = oversized_body self.service = FluentBitTestService( self.config_file, pre_start=self._start_receiver, @@ -378,7 +391,9 @@ def __init__(self): def _start_receiver(self, service): self.test_suite_http_port = service.test_suite_http_port - self.receiver = IPv6Http2OtlpReceiver(service.test_suite_http_port) + self.receiver = IPv6Http2OtlpReceiver( + service.test_suite_http_port, self.oversized_headers, self.oversized_body + ) self.receiver.start() def _stop_receiver(self, service): @@ -388,6 +403,7 @@ def _stop_receiver(self, service): def start(self): self.service.start() + self.flb = self.service.flb def stop(self): self.service.stop() @@ -903,6 +919,43 @@ def test_out_opentelemetry_http2_ipv6_authority_header(): assert request_seen["headers"][":authority"] == f"[::1]:{service.test_suite_http_port}" +def test_out_opentelemetry_http1_oversized_response_is_rejected(): + service = Service("out_otel_http2_http1_response_limit.yaml", use_tls=True) + service.start() + try: + configure_otlp_response(body="x" * (128 * 1024)) + service.wait_for_requests(1) + _wait_for_log_message(service, "HTTP response exceeded configured client limit") + finally: + service.stop() + + +def test_out_opentelemetry_http2_oversized_headers_are_rejected(): + if not ipv6_loopback_available(): + pytest.skip("IPv6 loopback is not available") + + service = Http2IPv6Service(oversized_headers=True) + try: + service.start() + service.wait_for_requests(1, timeout=30) + _wait_for_log_message(service, "HTTP response exceeded configured client limit") + finally: + service.stop() + + +def test_out_opentelemetry_http2_oversized_body_is_rejected(): + if not ipv6_loopback_available(): + pytest.skip("IPv6 loopback is not available") + + service = Http2IPv6Service(oversized_body=True) + try: + service.start() + service.wait_for_requests(1, timeout=30) + _wait_for_log_message(service, "HTTP response exceeded configured client limit") + finally: + service.stop() + + @pytest.mark.parametrize( "config_file,receiver_mode", [ From a39a20fd030e265d0ca4f23648f4c0a4e0dadf23 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 19:06:22 +0900 Subject: [PATCH 07/12] config: expose independent HTTP wire budget Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_config.h | 2 ++ src/flb_config.c | 5 +++++ 2 files changed, 7 insertions(+) diff --git a/include/fluent-bit/flb_config.h b/include/fluent-bit/flb_config.h index 6ca9f93c40c..ac13dc3f71b 100644 --- a/include/fluent-bit/flb_config.h +++ b/include/fluent-bit/flb_config.h @@ -101,6 +101,7 @@ struct flb_config { size_t http_client_max_response_size; size_t http_client_max_header_size; + size_t http_client_max_wire_size; int http_client_max_header_count; int http_client_response_timeout; @@ -430,6 +431,7 @@ enum conf_type { #define FLB_CONF_STR_FLUSH "Flush" #define FLB_CONF_STR_HTTP_CLIENT_MAX_RESPONSE_SIZE "http_client.max_response_size" #define FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_SIZE "http_client.max_header_size" +#define FLB_CONF_STR_HTTP_CLIENT_MAX_WIRE_SIZE "http_client.max_wire_size" #define FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_COUNT "http_client.max_header_count" #define FLB_CONF_STR_HTTP_CLIENT_RESPONSE_TIMEOUT "http_client.response_timeout" #define FLB_CONF_STR_FLUSH_ADAPTIVE "flush.adaptive" diff --git a/src/flb_config.c b/src/flb_config.c index 4007176c6fc..b3c9b53afaf 100644 --- a/src/flb_config.c +++ b/src/flb_config.c @@ -62,6 +62,10 @@ struct flb_service_config service_configs[] = { FLB_CONF_TYPE_SIZE, offsetof(struct flb_config, http_client_max_header_size)}, + {FLB_CONF_STR_HTTP_CLIENT_MAX_WIRE_SIZE, + FLB_CONF_TYPE_SIZE, + offsetof(struct flb_config, http_client_max_wire_size)}, + {FLB_CONF_STR_HTTP_CLIENT_MAX_HEADER_COUNT, FLB_CONF_TYPE_INT, offsetof(struct flb_config, http_client_max_header_count)}, @@ -339,6 +343,7 @@ struct flb_config *flb_config_init() config->flush = FLB_CONFIG_FLUSH_SECS; config->http_client_max_response_size = 4 * 1024 * 1024; config->http_client_max_header_size = 65536; + config->http_client_max_wire_size = 64 * 1024 * 1024; config->http_client_max_header_count = 100; config->http_client_response_timeout = 30; config->flush_adaptive = FLB_FALSE; From af72efdd52552656bc950aa597707a713c5e71c1 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 19:06:29 +0900 Subject: [PATCH 08/12] http_client: arm deadlines and separate wire budget Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_http_client.h | 2 + src/flb_http_client.c | 126 ++++++++++++++++++--------- 2 files changed, 88 insertions(+), 40 deletions(-) diff --git a/include/fluent-bit/flb_http_client.h b/include/fluent-bit/flb_http_client.h index d4c15225a76..9e4bc814087 100644 --- a/include/fluent-bit/flb_http_client.h +++ b/include/fluent-bit/flb_http_client.h @@ -46,6 +46,7 @@ /* Maximum response bytes received by a client-ng session. */ #define FLB_HTTP_CLIENT_NG_RESPONSE_SIZE_MAX (4 * 1024 * 1024) #define FLB_HTTP_CLIENT_NG_HEADER_SIZE_MAX 65536 +#define FLB_HTTP_CLIENT_NG_WIRE_SIZE_MAX (64 * 1024 * 1024) #define FLB_HTTP_CLIENT_NG_HEADER_COUNT_MAX 100 #define FLB_HTTP_CLIENT_NG_RESPONSE_TIMEOUT 30 #define FLB_HTTP_DATA_CHUNK 32768 @@ -284,6 +285,7 @@ struct flb_http_client_ng { int protocol_version; size_t max_response_size; size_t max_header_size; + size_t max_wire_size; size_t max_header_count; int response_timeout; cfl_sds_t temporary_buffer; diff --git a/src/flb_http_client.c b/src/flb_http_client.c index 67b0518eb30..4819d3f62b6 100644 --- a/src/flb_http_client.c +++ b/src/flb_http_client.c @@ -2387,6 +2387,7 @@ int flb_http_client_ng_init(struct flb_http_client_ng *client, client->protocol_version = protocol_version; client->max_response_size = FLB_HTTP_CLIENT_NG_RESPONSE_SIZE_MAX; client->max_header_size = FLB_HTTP_CLIENT_NG_HEADER_SIZE_MAX; + client->max_wire_size = FLB_HTTP_CLIENT_NG_WIRE_SIZE_MAX; client->max_header_count = FLB_HTTP_CLIENT_NG_HEADER_COUNT_MAX; client->response_timeout = FLB_HTTP_CLIENT_NG_RESPONSE_TIMEOUT; client->upstream_ha = upstream_ha; @@ -2711,6 +2712,12 @@ void flb_http_client_session_destroy(struct flb_http_client_session *session) session->connection->fd, session->original_net_setup->io_timeout); } + if (session->original_net_setup->io_timeout > 0) { + flb_connection_reset_io_timeout(session->connection); + } + else { + flb_connection_unset_io_timeout(session->connection); + } } flb_upstream_conn_release(session->connection); } @@ -2784,6 +2791,66 @@ struct flb_http_request *flb_http_client_request_begin(struct flb_http_client_se return &stream->request; } +#ifndef __EMSCRIPTEN__ +static int flb_http_client_session_arm_timeout(struct flb_http_client_session *session) +{ + struct flb_connection *connection; + time_t elapsed; + time_t now; + int timeout; + + connection = session->connection; + now = time(NULL); + + if (session->response_start == 0) { + session->response_start = now; + session->original_net_setup = connection->net; + if (session->original_net_setup == NULL) { + return -1; + } + + session->request_net_setup = *session->original_net_setup; + connection->net = &session->request_net_setup; + } + + elapsed = now - session->response_start; + if (elapsed >= session->parent->response_timeout) { + flb_upstream_conn_recycle(connection, FLB_FALSE); + return -1; + } + + if (elapsed < 0) { + elapsed = 0; + } + + timeout = session->parent->response_timeout - (int) elapsed; + if (session->original_net_setup->io_timeout > 0 && + session->original_net_setup->io_timeout < timeout) { + timeout = session->original_net_setup->io_timeout; + } + + session->request_net_setup.io_timeout = timeout; + if (connection->fd > 0) { + flb_net_socket_set_rcvtimeout(connection->fd, timeout); + } + + /* Async reads and writes use this timestamp while their coroutine yields. */ + flb_connection_reset_io_timeout(connection); + + return 0; +} + +static int flb_http_client_session_deadline_reached(struct flb_http_client_session *session) +{ + if (time(NULL) - session->response_start >= session->parent->response_timeout) { + flb_upstream_conn_recycle(session->connection, FLB_FALSE); + return FLB_TRUE; + } + + return FLB_FALSE; +} +#endif + struct flb_http_response *flb_http_client_request_execute_step( struct flb_http_request *request) { @@ -2804,35 +2871,6 @@ struct flb_http_response *flb_http_client_request_execute_step( return response; } - if (session->response_start == 0) { - session->response_start = time(NULL); - session->original_net_setup = session->connection->net; - if (session->original_net_setup != NULL) { - session->request_net_setup = *session->original_net_setup; - session->connection->net = &session->request_net_setup; - } - } - - result = session->parent->response_timeout - - (int) (time(NULL) - session->response_start); - if (result <= 0) { - flb_upstream_conn_recycle(session->connection, FLB_FALSE); - request->stream->status = HTTP_STREAM_STATUS_ERROR; - return NULL; - } - - if (session->original_net_setup != NULL) { - if (session->original_net_setup->io_timeout > 0 && - session->original_net_setup->io_timeout < result) { - result = session->original_net_setup->io_timeout; - } - - session->request_net_setup.io_timeout = result; - if (session->connection->fd > 0) { - flb_net_socket_set_rcvtimeout(session->connection->fd, result); - } - } - if (session->outgoing_data != NULL && cfl_sds_len(session->outgoing_data) > 0) { @@ -2912,6 +2950,10 @@ static int flb_http_client_session_read(struct flb_http_client_session *session) { ssize_t result; + if (flb_http_client_session_arm_timeout(session) != 0) { + return -1; + } + result = flb_io_net_read(session->connection, (void *) session->parent->temporary_buffer, cfl_sds_avail(session->parent->temporary_buffer)); @@ -2920,6 +2962,10 @@ static int flb_http_client_session_read(struct flb_http_client_session *session) return -1; } + if (flb_http_client_session_deadline_reached(session)) { + return -1; + } + result = (ssize_t) flb_http_client_session_ingest( session, (unsigned char *) session->parent->temporary_buffer, @@ -2970,6 +3016,10 @@ static int flb_http_client_session_write(struct flb_http_client_session *session data_length = cfl_sds_len(session->outgoing_data); if (data_length > 0) { + if (flb_http_client_session_arm_timeout(session) != 0) { + return -2; + } + result = flb_io_net_write(session->connection, (void *) session->outgoing_data, data_length, @@ -2979,6 +3029,10 @@ static int flb_http_client_session_write(struct flb_http_client_session *session return -2; } + if (flb_http_client_session_deadline_reached(session)) { + return -2; + } + if (data_sent < data_length) { memmove(session->outgoing_data, @@ -3004,18 +3058,10 @@ int flb_http_client_session_ingest(struct flb_http_client_session *session, #ifndef __EMSCRIPTEN__ struct cfl_list *iterator; struct flb_http_stream *stream; - size_t max_wire_size; - - max_wire_size = session->parent->max_response_size; - if (session->parent->max_header_size <= SIZE_MAX - max_wire_size) { - max_wire_size += session->parent->max_header_size; - } - else { - max_wire_size = SIZE_MAX; - } - if (session->received_bytes > max_wire_size || - length > max_wire_size - session->received_bytes) { + /* This independent budget includes chunk and HTTP/2 frame overhead. */ + if (session->received_bytes > session->parent->max_wire_size || + length > session->parent->max_wire_size - session->received_bytes) { session->response_limit_exceeded = FLB_TRUE; cfl_list_foreach(iterator, &session->streams) { stream = cfl_list_entry(iterator, struct flb_http_stream, _head); From 43a6ace19fe7b1a6764a1b0eb9999f615fc6077c Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 19:06:41 +0900 Subject: [PATCH 09/12] out_opentelemetry: apply HTTP wire budget Signed-off-by: Hiroshi Hatake --- plugins/out_opentelemetry/opentelemetry_conf.c | 1 + 1 file changed, 1 insertion(+) diff --git a/plugins/out_opentelemetry/opentelemetry_conf.c b/plugins/out_opentelemetry/opentelemetry_conf.c index 4999285f0db..eb90ae48070 100644 --- a/plugins/out_opentelemetry/opentelemetry_conf.c +++ b/plugins/out_opentelemetry/opentelemetry_conf.c @@ -754,6 +754,7 @@ struct opentelemetry_context *flb_opentelemetry_context_create(struct flb_output if (ret == 0) { ctx->http_client.max_response_size = config->http_client_max_response_size; ctx->http_client.max_header_size = config->http_client_max_header_size; + ctx->http_client.max_wire_size = config->http_client_max_wire_size; ctx->http_client.max_header_count = config->http_client_max_header_count; ctx->http_client.response_timeout = config->http_client_response_timeout; } From f220bfa9f72903b2c5abdbf9ff39847c9552a32a Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 19:06:48 +0900 Subject: [PATCH 10/12] tests: integration: cover HTTP framing and deadlines Signed-off-by: Hiroshi Hatake --- .../out_otel_http2_http1_response_limit.yaml | 7 +- .../out_otel_http2_http1_stalled_write.yaml | 26 +++ .../out_otel_http2_ipv6_response_limit.yaml | 2 + .../tests/test_out_opentelemetry_001.py | 163 +++++++++++++++++- 4 files changed, 189 insertions(+), 9 deletions(-) create mode 100644 tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_stalled_write.yaml diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml index 658482f56ba..8b31d68333a 100644 --- a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml @@ -1,9 +1,10 @@ service: flush: 1 - http_client.max_response_size: 64K - http_client.max_header_size: 16K + http_client.max_response_size: 16K + http_client.max_header_size: 1K + http_client.max_wire_size: 256K http_client.max_header_count: 50 - http_client.response_timeout: 10 + http_client.response_timeout: ${TEST_SUITE_HTTP_CLIENT_RESPONSE_TIMEOUT} log_level: info http_server: on http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_stalled_write.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_stalled_write.yaml new file mode 100644 index 00000000000..46fb64cbfb4 --- /dev/null +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_stalled_write.yaml @@ -0,0 +1,26 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + http_client.response_timeout: 2 + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + logs_max_size: 32M + http_server.buffer_max_size: 32M + + outputs: + - name: opentelemetry + match: "*" + host: localhost + port: ${TEST_SUITE_HTTP_PORT} + http2: on + tls: on + tls.verify: on + tls.verify_hostname: on + tls.vhost: localhost + tls.ca_file: ${CERTIFICATE_TEST} + retry_limit: 1 diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml index 3f8036a3501..c71aa1c7952 100644 --- a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_ipv6_response_limit.yaml @@ -4,6 +4,8 @@ service: http_server: on http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} http_client.max_response_size: 16K + http_client.max_header_size: 1K + http_client.max_wire_size: 256K pipeline: inputs: diff --git a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py index f870b8643c7..86afbb8d6a8 100644 --- a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py +++ b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py @@ -5,6 +5,7 @@ import re import shutil import socket +import ssl import threading import time from collections import Counter @@ -86,6 +87,7 @@ def __init__( use_tls=False, grpc_methods=None, use_oauth_server=False, + response_timeout=10, ): self.config_file = _repo_relative("../config", config_file) cert_dir = _repo_relative("../../in_splunk/certificate") @@ -103,6 +105,7 @@ def __init__( extra_env={ "CERTIFICATE_TEST": self.tls_crt_file, "PRIVATE_KEY_TEST": self.tls_key_file, + "TEST_SUITE_HTTP_CLIENT_RESPONSE_TIMEOUT": response_timeout, }, pre_start=self._start_receiver, post_stop=self._stop_receiver, @@ -269,10 +272,12 @@ def _build_signal_payload_from_dict(self, payload_dict, signal_type): class IPv6Http2OtlpReceiver: - def __init__(self, port, oversized_headers=False, oversized_body=False): + def __init__(self, port, oversized_headers=False, oversized_body=False, + fragmented_body=False): self.port = port self.oversized_headers = oversized_headers self.oversized_body = oversized_body + self.fragmented_body = fragmented_body self.server_socket = None self.thread = None self.requests = [] @@ -354,13 +359,19 @@ def _handle_connection(self, client_socket): client_socket.sendall(pending) def _send_response(self, connection, stream_id): - body = b"x" * (32 * 1024) if self.oversized_body else b"{}" + if self.oversized_body: + body = b"x" * (32 * 1024) + elif self.fragmented_body: + body = b"x" * 8192 + else: + body = b"{}" headers = [ (":status", "200"), ("content-type", "application/json"), - ("content-length", str(len(body))), ] + if not self.fragmented_body: + headers.append(("content-length", str(len(body)))) if self.oversized_headers: headers.extend((f"x-extra-{index}", "a" * 1024) for index in range(70)) @@ -368,21 +379,27 @@ def _send_response(self, connection, stream_id): if self.oversized_body: connection.send_data(stream_id, body[:16384]) connection.send_data(stream_id, body[16384:], end_stream=True) + elif self.fragmented_body: + for offset in range(len(body)): + connection.send_data(stream_id, body[offset:offset + 1], + end_stream=offset + 1 == len(body)) else: connection.send_data(stream_id, body, end_stream=True) class Http2IPv6Service: - def __init__(self, oversized_headers=False, oversized_body=False): + def __init__(self, oversized_headers=False, oversized_body=False, + fragmented_body=False): config_name = ( "out_otel_http2_ipv6_response_limit.yaml" - if oversized_body else "out_otel_http2_ipv6_logs.yaml" + if oversized_body or fragmented_body else "out_otel_http2_ipv6_logs.yaml" ) self.config_file = _repo_relative("../config", config_name) self.receiver = None self.test_suite_http_port = None self.oversized_headers = oversized_headers self.oversized_body = oversized_body + self.fragmented_body = fragmented_body self.service = FluentBitTestService( self.config_file, pre_start=self._start_receiver, @@ -392,7 +409,8 @@ def __init__(self, oversized_headers=False, oversized_body=False): def _start_receiver(self, service): self.test_suite_http_port = service.test_suite_http_port self.receiver = IPv6Http2OtlpReceiver( - service.test_suite_http_port, self.oversized_headers, self.oversized_body + service.test_suite_http_port, self.oversized_headers, + self.oversized_body, self.fragmented_body ) self.receiver.start() @@ -419,6 +437,51 @@ def wait_for_requests(self, minimum_count, timeout=10): ) +class StalledTlsReceiver: + def __init__(self, port, cert_file, key_file): + self.port = port + self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + self.context.load_cert_chain(cert_file, key_file) + self.context.set_alpn_protocols(["http/1.1"]) + self.accepted = threading.Event() + self.stop_event = threading.Event() + self.thread = None + self.server_socket = None + + def start(self): + self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.server_socket.bind(("127.0.0.1", self.port)) + self.server_socket.listen(1) + self.server_socket.settimeout(0.5) + self.thread = threading.Thread(target=self._serve, daemon=True) + self.thread.start() + + def _serve(self): + while not self.stop_event.is_set(): + try: + client_socket, _ = self.server_socket.accept() + except socket.timeout: + continue + except OSError: + return + + with client_socket: + client_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096) + client_socket.settimeout(5) + with self.context.wrap_socket(client_socket, server_side=True): + self.accepted.set() + self.stop_event.wait(15) + return + + def stop(self): + self.stop_event.set() + if self.server_socket is not None: + self.server_socket.close() + if self.thread is not None: + self.thread.join(timeout=5) + + def ipv6_loopback_available(): sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) try: @@ -930,6 +993,78 @@ def test_out_opentelemetry_http1_oversized_response_is_rejected(): service.stop() +def test_out_opentelemetry_http1_chunk_framing_does_not_count_as_body(): + service = Service("out_otel_http2_http1_response_limit.yaml", use_tls=True) + service.start() + try: + configure_otlp_response(body=(b"x" for _ in range(8192))) + service.wait_for_requests(1) + _wait_for_log_message(service, "HTTP status=200") + finally: + service.stop() + + with open(service.flb.log_file, encoding="utf-8") as log_file: + assert "HTTP response exceeded configured client limit" not in log_file.read() + + +def test_out_opentelemetry_http1_response_deadline(): + service = Service("out_otel_http2_http1_response_limit.yaml", use_tls=True, + response_timeout=1) + service.start() + try: + configure_otlp_response(delay_seconds=3) + service.wait_for_requests(1) + _wait_for_log_message(service, "timed out after 1 seconds (IO timeout)", timeout=8) + finally: + service.stop() + + +def test_out_opentelemetry_stalled_first_write_hits_deadline(): + cert_dir = _repo_relative("../../in_splunk/certificate") + cert_file = os.path.join(cert_dir, "certificate.pem") + key_file = os.path.join(cert_dir, "private_key.pem") + receiver = None + + def start_receiver(harness): + nonlocal receiver + receiver = StalledTlsReceiver(harness.test_suite_http_port, cert_file, key_file) + receiver.start() + + def stop_receiver(harness): + if receiver is not None: + receiver.stop() + + service = FluentBitTestService( + _repo_relative("../config", "out_otel_http2_http1_stalled_write.yaml"), + extra_env={"CERTIFICATE_TEST": cert_file}, + pre_start=start_receiver, + post_stop=stop_receiver, + ) + service.start() + try: + payload = ExportLogsServiceRequest() + record = payload.resource_logs.add().scope_logs.add().log_records.add() + record.body.string_value = "x" * (16 * 1024 * 1024) + response = requests.post( + f"http://127.0.0.1:{service.flb_listener_port}/v1/logs", + data=payload.SerializeToString(), + headers={"Content-Type": "application/x-protobuf"}, + timeout=15, + ) + response.raise_for_status() + assert receiver.accepted.wait(timeout=10) + + def request_timed_out(): + with open(service.flb.log_file, encoding="utf-8") as log_file: + return "timed out after 2 seconds (IO timeout)" in log_file.read() + + service.wait_for_condition( + request_timed_out, timeout=8, description="stalled HTTP request timeout" + ) + finally: + service.stop() + + def test_out_opentelemetry_http2_oversized_headers_are_rejected(): if not ipv6_loopback_available(): pytest.skip("IPv6 loopback is not available") @@ -956,6 +1091,22 @@ def test_out_opentelemetry_http2_oversized_body_is_rejected(): service.stop() +def test_out_opentelemetry_http2_frame_overhead_does_not_count_as_body(): + if not ipv6_loopback_available(): + pytest.skip("IPv6 loopback is not available") + + service = Http2IPv6Service(fragmented_body=True) + try: + service.start() + service.wait_for_requests(1, timeout=30) + _wait_for_log_message(service, "HTTP status=200") + finally: + service.stop() + + with open(service.flb.log_file, encoding="utf-8") as log_file: + assert "HTTP response exceeded configured client limit" not in log_file.read() + + @pytest.mark.parametrize( "config_file,receiver_mode", [ From b2a421d53faa96e8c6857711837f4cd27a7b7160 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 19:15:39 +0900 Subject: [PATCH 11/12] http_client_http1: enforce response header count limit Signed-off-by: Hiroshi Hatake --- src/flb_http_client_http1.c | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/flb_http_client_http1.c b/src/flb_http_client_http1.c index 1e5e042a384..af670e4a896 100644 --- a/src/flb_http_client_http1.c +++ b/src/flb_http_client_http1.c @@ -97,9 +97,11 @@ static int flb_http1_client_session_process_headers(struct flb_http_client_sessi { char *header_block_begining; size_t header_block_length; + size_t header_count; size_t status_line_length; char *header_block_end; char *status_line; + char *cursor; int result; status_line = session->incoming_data; @@ -163,6 +165,24 @@ static int flb_http1_client_session_process_headers(struct flb_http_client_sessi return -1; } + header_count = 0; + if (header_block_end > header_block_begining) { + header_count = 1; + for (cursor = header_block_begining; + cursor + 1 < header_block_end; + cursor++) { + if (cursor[0] == '\r' && cursor[1] == '\n') { + header_count++; + } + } + } + + if (header_count > session->parent->max_header_count) { + session->response_limit_exceeded = FLB_TRUE; + response->stream->status = HTTP_STREAM_STATUS_ERROR; + return -1; + } + header_block_length = (size_t) ((uintptr_t) header_block_end - (uintptr_t) header_block_begining); From 20d6419bf2b6f4e80a99b7613873883c568076a5 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 26 Oct 2026 19:15:46 +0900 Subject: [PATCH 12/12] tests: integration: cover HTTP/1 response header count limit Signed-off-by: Hiroshi Hatake --- .../out_otel_http2_http1_response_limit.yaml | 2 +- .../tests/test_out_opentelemetry_001.py | 29 +++++++++++++++++++ tests/integration/src/server/otlp_server.py | 6 ++++ 3 files changed, 36 insertions(+), 1 deletion(-) diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml index 8b31d68333a..8e6d28e492c 100644 --- a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http2_http1_response_limit.yaml @@ -3,7 +3,7 @@ service: http_client.max_response_size: 16K http_client.max_header_size: 1K http_client.max_wire_size: 256K - http_client.max_header_count: 50 + http_client.max_header_count: 12 http_client.response_timeout: ${TEST_SUITE_HTTP_CLIENT_RESPONSE_TIMEOUT} log_level: info http_server: on diff --git a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py index 86afbb8d6a8..e4521dfb993 100644 --- a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py +++ b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py @@ -88,6 +88,7 @@ def __init__( grpc_methods=None, use_oauth_server=False, response_timeout=10, + response_options=None, ): self.config_file = _repo_relative("../config", config_file) cert_dir = _repo_relative("../../in_splunk/certificate") @@ -97,6 +98,7 @@ def __init__( self.use_tls = use_tls self.grpc_methods = grpc_methods or {} self.use_oauth_server = use_oauth_server + self.response_options = response_options self.oauth_server_port = None self.service = FluentBitTestService( self.config_file, @@ -142,6 +144,8 @@ def _start_receiver(self, service): tls_key_file=self.tls_key_file, use_grpc=self.receiver_mode == "grpc", ) + if self.response_options is not None: + configure_otlp_response(**self.response_options) if self.receiver_mode == "grpc": self._wait_for_tcp_port(service.test_suite_http_port) @@ -993,6 +997,31 @@ def test_out_opentelemetry_http1_oversized_response_is_rejected(): service.stop() +@pytest.mark.parametrize( + "extra_header_count,rejected", + [(5, False), (15, True)], + ids=["within_header_count", "over_header_count"], +) +def test_out_opentelemetry_http1_header_count_limit(extra_header_count, rejected): + service = Service( + "out_otel_http2_http1_response_limit.yaml", + use_tls=True, + response_options={ + "body": "{}", + "headers": {f"X-Test-{index}": "x" for index in range(extra_header_count)}, + }, + ) + service.start() + try: + service.wait_for_requests(1) + if rejected: + _wait_for_log_message(service, "HTTP response exceeded configured client limit") + else: + _wait_for_log_message(service, "HTTP status=200") + finally: + service.stop() + + def test_out_opentelemetry_http1_chunk_framing_does_not_count_as_body(): service = Service("out_otel_http2_http1_response_limit.yaml", use_tls=True) service.start() diff --git a/tests/integration/src/server/otlp_server.py b/tests/integration/src/server/otlp_server.py index e4a755ac832..5a1f0776134 100644 --- a/tests/integration/src/server/otlp_server.py +++ b/tests/integration/src/server/otlp_server.py @@ -45,6 +45,7 @@ "status_codes": [], "body": {"status": "received"}, "content_type": "application/json", + "headers": {}, "delay_seconds": 0, } grpc_method_paths = { @@ -69,6 +70,7 @@ def reset_otlp_server_state(): "status_codes": [], "body": {"status": "received"}, "content_type": "application/json", + "headers": {}, "delay_seconds": 0, } ) @@ -88,6 +90,7 @@ def configure_otlp_response( status_codes=None, body=None, content_type=None, + headers=None, delay_seconds=None, ): if status_code is not None: @@ -98,6 +101,8 @@ def configure_otlp_response( response_config["body"] = body if content_type is not None: response_config["content_type"] = content_type + if headers is not None: + response_config["headers"] = dict(headers) if delay_seconds is not None: response_config["delay_seconds"] = delay_seconds @@ -128,6 +133,7 @@ def _build_response(): body, status=status_code, content_type=response_config["content_type"], + headers=response_config["headers"], )