feat: SSRF protection for HTTP clients

- Extract SSRF protection logic into reusable `httpx.GetSSRFDialContext`
- Apply SSRF protection to `Global HTTP Client`, `Service HTTP Client`, and `IO HTTP Client`
- Introduce `EnableAppPrivateNet` configuration to allow internal network access (default: false)

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

# Conflicts:
#	internal/pkg/httpx/http.go

Author: ngjaying

docs/en_US/configuration/global_configurations.md

@@ -48,8 +48,19 @@ basic:
   cfgStorageType: file
   # If it is enabled, each REST API call will print logs
   enableRestAuditLog: false
+  # If it is enabled, the rule functions can access the private network.
+  enablePrivateNet: false
 ```
 
+The configuration item **enablePrivateNet** is used to specify whether the rule functions (e.g. valid func, sinks)
+can access the private network (e.g. localhost, 127.0.0.1). If it is true, the private network can be accessed. Default
+is false for security.
+
+> [!WARNING]
+> Since version v2.4.0, the default value of `enablePrivateNet` is `false`, which means accessing private network
+> addresses is blocked by default. If your rules rely on accessing local resources (e.g., local REST services, local
+> files), you MUST set this configuration to `true`.
+
 for debug option in basic following env is valid `KUIPER__BASIC__DEBUG=true` and if used debug value will be set to true.
 
 The configuration item **ignoreCase** is used to specify whether case is ignored in SQL processing. If it is true, the column name case of the input data can be different from that defined in SQL. If the column name case in SQL statements, stream definitions, and input data can be guaranteed to be exactly the same, it is recommended to set this value to "false" to obtain better performance. Before version 1.10, its default value was true to be compatible with standard SQL; after version 1.10, its default value was changed to false for better performance.

docs/zh_CN/configuration/global_configurations.md

@@ -44,8 +44,17 @@ basic:
   cfgStorageType: file
   # If it is enabled, each REST API call will print logs
   enableRestAuditLog: false
+  # If it is enabled, the rule functions can access the private network.
+  enablePrivateNet: false
 ```
 
+配置项 **enablePrivateNet** 用于指定规则函数（例如 valid func、sinks）是否可以访问私有网络（例如 localhost、127.0.0.1）。如果为
+true，则可以访问私有网络。出于安全考虑，默认为 false。
+
+> [!WARNING]
+> 自版本 v2.4.0 起，`enablePrivateNet` 的默认值为 `false`，这意味着默认情况下会阻止访问私有网络地址。如果您的规则依赖于访问本地资源（例如本地
+> REST 服务、本地文件），您必须将此配置设置为 `true`。
+
 将basic项目下debug的值设置为true是有效的 `KUIPER__BASIC__DEBUG=true`。
 
 配置项 **ignoreCase** 用于指定 SQL 处理中是否大小写无关。若为 true，则输入数据的列名大小写可以与 SQL 中的定义不同。如果 SQL 语句中，流定义以及输入数据中可以保证列名大小写完全一致，则建议设置该值为 false 以获得更优的性能。在 1.10 版本之前，其默认值为 true ， 以兼容标准 SQL ；在 1.10 及之后版本中，默认值改为 false ，以获得更优的性能。

etc/kuiper.yaml

@@ -77,6 +77,8 @@ basic:
     retainedDuration: 6h
   # If it is enabled, each REST API call will print logs
   enableRestAuditLog: false
+  # If it is enabled, the rule functions can access the private network.
+  enablePrivateNet: false
 
 # The default options for all rules. Each rule can override this setting by defining its own option
 rule:

internal/io/http/client.go

@@ -204,6 +204,7 @@ func (cc *ClientConf) InitConf(ctx api.StreamContext, device string, props map[s
 		return err
 	}
 	tr := newTransport(tlscfg, conf.Log)
+	tr.DialContext = httpx.GetSSRFDialContext(time.Duration(c.Timeout))
 	cc.client = &http.Client{
 		Transport: tr,
 		Timeout:   time.Duration(c.Timeout),

internal/io/http/client_test.go

@@ -21,9 +21,16 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
+func init() {
+	testx.InitEnv("http")
+	conf.Config.Basic.EnablePrivateNet = true
+}
+
 func TestInitConf(t *testing.T) {
 	m := map[string]interface{}{}
 	ctx := mockContext.NewMockContext("1", "2")

internal/pkg/httpx/http.go

@@ -16,15 +16,18 @@ package httpx
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"fmt"
 	"io"
 	"mime/multipart"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
 	"strconv"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
@@ -179,6 +182,7 @@ func ReadFile(uri string) (io.ReadCloser, error) {
 			Timeout: timeout,
 			Transport: &http.Transport{
 				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+				DialContext:     GetSSRFDialContext(timeout),
 			},
 		}
 		resp, err := client.Get(uri)
@@ -195,6 +199,29 @@ func ReadFile(uri string) (io.ReadCloser, error) {
 	return src, nil
 }
 
+func GetSSRFDialContext(timeout time.Duration) func(ctx context.Context, network, addr string) (net.Conn, error) {
+	return func(ctx context.Context, network, addr string) (net.Conn, error) {
+		d := net.Dialer{
+			Timeout: timeout,
+			Control: func(network, address string, c syscall.RawConn) error {
+				if conf.Config != nil && conf.Config.Basic.EnablePrivateNet {
+					return nil
+				}
+				host, _, err := net.SplitHostPort(address)
+				if err != nil {
+					return err
+				}
+				ip := net.ParseIP(host)
+				if ip != nil && (ip.IsLoopback() || ip.IsPrivate() || ip.IsUnspecified() || ip.IsMulticast() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast()) {
+					return fmt.Errorf("ip %s is in internal network", ip.String())
+				}
+				return nil
+			},
+		}
+		return d.DialContext(ctx, network, addr)
+	}
+}
+
 func DownloadFile(folder string, name string, uri string) (err error) {
 	src, err := ReadFile(uri)
 	if err != nil {

internal/pkg/httpx/ssrf_test.go

@@ -0,0 +1,47 @@
+package httpx
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+)
+
+func TestReadFileSSRF(t *testing.T) {
+	// Start a local test server
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintln(w, "secret data")
+	}))
+	defer server.Close()
+
+	// 1. Defaut behavior: should BLOCK private access
+	// Current behavior: ReadFile checks internal/conf.Config.Basic.EnablePrivateNet
+	// Since conf.Config is nil or default, it blocks (we should ensure it blocks if nil too, or we mock it)
+
+	// Ensure config is nil or clean to start
+	conf.Config = nil
+
+	_, err := ReadFile(server.URL)
+	assert.Error(t, err, "ReadFile should block access to local server by default")
+	if err != nil {
+		assert.Contains(t, err.Error(), "internal network")
+	}
+
+	// 2. Enable private access: should ALLOW
+	conf.Config = &model.KuiperConf{}
+	conf.Config.Basic.EnablePrivateNet = true
+
+	// Reset config after test
+	defer func() { conf.Config = nil }()
+
+	rc, err := ReadFile(server.URL)
+	assert.NoError(t, err, "ReadFile should allow access when EnablePrivateNet is true")
+	if rc != nil {
+		rc.Close()
+	}
+}

internal/plugin/native/manager_test.go

@@ -31,13 +31,15 @@ import (
 
 	"github.com/lf-edge/ekuiper/v2/internal/binder"
 	"github.com/lf-edge/ekuiper/v2/internal/binder/function"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/plugin"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 )
 
 func init() {
 	testx.InitEnv("native")
+	conf.Config.Basic.EnablePrivateNet = true
 	meta.InitYamlConfigManager()
 	var (
 		nativeManager *Manager

internal/plugin/native/security_test.go

@@ -4,37 +4,12 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/lf-edge/ekuiper/v2/internal/binder"
-	"github.com/lf-edge/ekuiper/v2/internal/binder/function"
-	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/plugin"
-	"github.com/lf-edge/ekuiper/v2/internal/testx"
 )
 
-func init() {
-	testx.InitEnv("native")
-	meta.InitYamlConfigManager()
-	var (
-		nativeManager *Manager
-		err           error
-	)
-	for i := 0; i < 10; i++ {
-		if nativeManager, err = InitManager(); err != nil {
-			time.Sleep(10 * time.Millisecond)
-		} else {
-			break
-		}
-	}
-	err = function.Initialize([]binder.FactoryEntry{{Name: "native plugin", Factory: nativeManager}})
-	if err != nil {
-		panic(err)
-	}
-}
-
 func TestManager_Register_PathTraversal(t *testing.T) {
 	s := httptest.NewServer(
 		http.FileServer(http.Dir("../testzips")),

internal/plugin/portable/manager_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/plugin"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
@@ -37,6 +38,7 @@ import (
 
 func init() {
 	testx.InitEnv("portable")
+	conf.Config.Basic.EnablePrivateNet = true
 	// Wait for other db tests to finish to avoid db lock
 	for i := 0; i < 10; i++ {
 		if _, err := InitManager(); err != nil {