http_client: out_opentelemetry: mitigate unbounded http ng requests

include/fluent-bit/flb_config.h

@@ -99,6 +99,12 @@ 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;
+    size_t http_client_max_wire_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,13 +421,19 @@ 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,
     FLB_CONF_TYPE_OTHER,
 };
 
 #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"
 #define FLB_CONF_STR_FLUSH_ADAPTIVE_MIN "flush.adaptive.min_interval"
 #define FLB_CONF_STR_FLUSH_ADAPTIVE_MAX "flush.adaptive.max_interval"

include/fluent-bit/flb_http_client.h

@@ -42,6 +42,13 @@
 /* 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_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
 
 /* HTTP Methods */
@@ -276,6 +283,11 @@ 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_wire_size;
+    size_t                  max_header_count;
+    int                     response_timeout;
     cfl_sds_t               temporary_buffer;
 
     int                     releasable;
@@ -293,6 +305,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;

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) {

plugins/out_opentelemetry/opentelemetry_conf.c

@@ -751,6 +751,14 @@ 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_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;
+    }
+
     if (ret != 0) {
         flb_plg_debug(ctx->ins, "http client creation error");
 

src/flb_config.c

@@ -54,6 +54,26 @@ 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_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)},
+
+    {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 +341,11 @@ 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_wire_size = 64 * 1024 * 1024;
+    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 +883,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 +941,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);

src/flb_http_client.c

@@ -2385,6 +2385,11 @@ 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;
     client->upstream = upstream;
     client->flags = flags;
@@ -2700,6 +2705,20 @@ 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);
+                }
+                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);
         }
 
@@ -2772,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)
 {
@@ -2871,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));
@@ -2879,12 +2962,17 @@ 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,
                             result);
 
     if (result < 0) {
+        flb_upstream_conn_recycle(session->connection, FLB_FALSE);
         return -2;
     }
 
@@ -2928,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,
@@ -2937,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,
@@ -2960,6 +3056,26 @@ 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;
+
+    /* 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);
+            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,