wip

Author: sparrc

plugins/in_forward/fw_conn.c

@@ -159,6 +159,7 @@ struct fw_conn *fw_conn_add(struct flb_connection *connection, struct flb_in_fw_
         return NULL;
     }
 
+    conn->being_deleted = 0;
     conn->handshake_status = FW_HANDSHAKE_ESTABLISHED;
     /*
      * Always force the secure-forward handshake when:
@@ -243,6 +244,12 @@ struct fw_conn *fw_conn_add(struct flb_connection *connection, struct flb_in_fw_
 
 int fw_conn_del(struct fw_conn *conn)
 {
+    /*
+     * Set being_deleted flag to prevent any in-flight processing
+     * from accessing this connection's resources
+     */
+    conn->being_deleted = 1;
+
     /* The downstream unregisters the file descriptor from the event-loop
      * so there's nothing to be done by the plugin
      */
@@ -254,6 +261,7 @@ int fw_conn_del(struct fw_conn *conn)
     /* Release decompression context if it exists */
     if (conn->d_ctx) {
         flb_decompression_context_destroy(conn->d_ctx);
+        conn->d_ctx = NULL;
     }
 
     if (conn->helo != NULL) {
@@ -264,8 +272,15 @@ int fw_conn_del(struct fw_conn *conn)
             flb_sds_destroy(conn->helo->salt);
         }
         flb_free(conn->helo);
+        conn->helo = NULL;
     }
-    flb_free(conn->buf);
+
+    /* Free buffer and set to NULL to prevent use-after-free */
+    if (conn->buf) {
+        flb_free(conn->buf);
+        conn->buf = NULL;
+    }
+
     flb_free(conn);
 
     return 0;

plugins/in_forward/fw_conn.h

@@ -43,6 +43,7 @@ struct flb_in_fw_helo;
 struct fw_conn {
     int status;                      /* Connection status                 */
     int handshake_status;            /* handshake status                 */
+    int being_deleted;               /* Flag: connection is being deleted */
 
     /* Buffer */
     char *buf;                       /* Buffer data                       */

plugins/in_forward/fw_prot.c

@@ -1076,6 +1076,11 @@ static size_t receiver_recv(struct fw_conn *conn, char *buf, size_t try_size) {
     size_t off;
     size_t actual_size;
 
+    /* Safety check: ensure connection is not being deleted and buffer exists */
+    if (conn->being_deleted || !conn->buf) {
+        return 0;
+    }
+
     off = conn->buf_len - conn->rest;
     actual_size = try_size;
 
@@ -1288,6 +1293,24 @@ int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn)
     conn->rest = conn->buf_len;
 
     while (1) {
+        /* Check if connection is being deleted or plugin is paused */
+        if (conn->being_deleted) {
+            msgpack_unpacker_free(unp);
+            msgpack_unpacked_destroy(&result);
+            flb_sds_destroy(out_tag);
+            return 0;
+        }
+
+        pthread_mutex_lock(&ctx->conn_mutex);
+        if (ctx->is_paused) {
+            pthread_mutex_unlock(&ctx->conn_mutex);
+            msgpack_unpacker_free(unp);
+            msgpack_unpacked_destroy(&result);
+            flb_sds_destroy(out_tag);
+            return 0;
+        }
+        pthread_mutex_unlock(&ctx->conn_mutex);
+
         recv_len = receiver_to_unpacker(conn, EACH_RECV_SIZE, unp);
         if (recv_len == 0) {
             /* No more data */