[branch-4.1][fix](security) Add auth and config gate for _stream_load_forward endpoint

be/src/common/config.cpp

@@ -272,6 +272,8 @@ DEFINE_mInt32(download_low_speed_limit_kbps, "50");
 DEFINE_mInt32(download_low_speed_time, "300");
 // whether to download small files in batch
 DEFINE_mBool(enable_batch_download, "true");
+// whether to enable stream load forward endpoint for cloud group commit
+DEFINE_mBool(enable_group_commit_streamload_be_forward, "false");
 // whether to check md5sum when download
 DEFINE_mBool(enable_download_md5sum_check, "false");
 // download binlog meta timeout, default 30s

be/src/common/config.h

@@ -330,6 +330,8 @@ DECLARE_mInt32(download_low_speed_limit_kbps);
 DECLARE_mInt32(download_low_speed_time);
 // whether to download small files in batch.
 DECLARE_mBool(enable_batch_download);
+// whether to enable stream load forward endpoint for cloud group commit
+DECLARE_mBool(enable_group_commit_streamload_be_forward);
 // whether to check md5sum when download
 DECLARE_mBool(enable_download_md5sum_check);
 // download binlog meta timeout

be/src/service/http/action/stream_load_forward_handler.cpp

@@ -32,6 +32,18 @@ namespace doris {
 #include "common/compile_check_begin.h"
 
 int StreamLoadForwardHandler::on_header(HttpRequest* req) {
+    if (!config::enable_group_commit_streamload_be_forward) {
+        HttpChannel::send_reply(req, HttpStatus::FORBIDDEN,
+                                "Stream load forward is disabled. "
+                                "Set enable_group_commit_streamload_be_forward=true to enable.");
+        return -1;
+    }
+
+    int auth_ret = HttpHandlerWithAuth::on_header(req);
+    if (auth_ret != 0) {
+        return auth_ret;
+    }
+
     std::ostringstream headers_info;
     const auto& headers = req->headers();
     for (const auto& header : headers) {

be/src/service/http/action/stream_load_forward_handler.h

@@ -25,7 +25,8 @@
 #include <string>
 
 #include "common/status.h"
-#include "service/http/http_handler.h"
+#include "runtime/exec_env.h"
+#include "service/http/http_handler_with_auth.h"
 #include "service/http/http_request.h"
 #include "util/byte_buffer.h"
 
@@ -98,9 +99,10 @@ class StreamLoadForwardContext {
 // Stream Load request forward handler
 // Forwards Stream Load requests to other BE nodes
 // Supports streaming forward, maintains original request path format: /api/{db}/{table}/_stream_load_forward
-class StreamLoadForwardHandler : public HttpHandler {
+class StreamLoadForwardHandler : public HttpHandlerWithAuth {
 public:
-    StreamLoadForwardHandler() = default;
+    explicit StreamLoadForwardHandler(ExecEnv* exec_env)
+            : HttpHandlerWithAuth(exec_env, TPrivilegeHier::GLOBAL, TPrivilegeType::LOAD) {}
     ~StreamLoadForwardHandler() override = default;
 
     void handle(HttpRequest* req) override;

be/src/service/http_service.cpp

@@ -141,7 +141,7 @@ Status HttpService::start() {
                                       streamload_2pc_action);
 
     // register stream load forward handler
-    auto* forward_handler = _pool.add(new StreamLoadForwardHandler());
+    auto* forward_handler = _pool.add(new StreamLoadForwardHandler(_env));
     _ev_http_server->register_handler(HttpMethod::PUT, "/api/{db}/{table}/_stream_load_forward",
                                       forward_handler);
 

regression-test/suites/load_p0/stream_load/test_group_commit_redirect.groovy

@@ -93,6 +93,7 @@ suite('test_group_commit_redirect', 'docker') {
     options.beNum = 3
     options.cloudMode = true
     options.beConfigs.add('enable_java_support=false')
+    options.beConfigs.add('enable_group_commit_streamload_be_forward=true')
     options.feConfigs.add('enable_group_commit_streamload_be_forward=true')
     docker(options) {
         // get fe and be info