feat(node): per node outbound routing (#5275)

* feat: add per-node outbound routing for panel-to-node connections

* feat(ui): add outbound tag selector to node form with i18n

* fix(xray): avoid potential overflow warning in node egress rule allocation

* chore: run "npm run gen"

* fix

---------

Co-authored-by: Sanaei <ho3ein.sanaei@gmail.com>

internal/web/service/inbound_node_reconcile_test.go

@@ -109,7 +109,7 @@ func TestReconcileNode_SelectedModeLeavesUnselectedRemoteInbounds(t *testing.T)
 	seedInboundConflictNode(t, "keep", "", 443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)
 
 	svc := InboundService{}
-	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node), node); err != nil {
+	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node); err != nil {
 		t.Fatalf("ReconcileNode: %v", err)
 	}
 
@@ -133,7 +133,7 @@ func TestReconcileNode_AllModeDeletesUndesiredRemoteInbounds(t *testing.T) {
 	seedInboundConflictNode(t, "keep", "", 443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id)
 
 	svc := InboundService{}
-	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node), node); err != nil {
+	if err := svc.ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node); err != nil {
 		t.Fatalf("ReconcileNode: %v", err)
 	}
 

internal/web/service/node.go

@@ -17,9 +17,12 @@ import (
 
 	"github.com/mhsanaei/3x-ui/v3/internal/database"
 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"github.com/mhsanaei/3x-ui/v3/internal/logger"
 	"github.com/mhsanaei/3x-ui/v3/internal/util/common"
+	"github.com/mhsanaei/3x-ui/v3/internal/util/json_util"
 	"github.com/mhsanaei/3x-ui/v3/internal/util/netsafe"
 	"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
+	"github.com/mhsanaei/3x-ui/v3/internal/xray"
 )
 
 type HeartbeatPatch struct {
@@ -339,6 +342,7 @@ func (s *NodeService) Update(id int, in *model.Node) error {
 		"pinned_cert_sha256":    in.PinnedCertSha256,
 		"inbound_sync_mode":     in.InboundSyncMode,
 		"inbound_tags":          string(inboundTagsJSON),
+		"outbound_tag":          in.OutboundTag,
 	}
 	if err := db.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
 		return err
@@ -353,7 +357,7 @@ func (s *NodeService) GetRemoteInboundOptions(ctx context.Context, n *model.Node
 	if err := s.normalize(n); err != nil {
 		return nil, err
 	}
-	return runtime.NewRemote(n).ListInboundOptions(ctx)
+	return runtime.NewRemote(n, nil).ListInboundOptions(ctx)
 }
 
 // EnsureInboundTagAllowed adds a panel-managed inbound's tag to the node's
@@ -427,7 +431,13 @@ func (s *NodeService) Delete(id int) error {
 
 func (s *NodeService) SetEnable(id int, enable bool) error {
 	db := database.GetDB()
-	return db.Model(model.Node{}).Where("id = ?", id).Update("enable", enable).Error
+	if err := db.Model(model.Node{}).Where("id = ?", id).Update("enable", enable).Error; err != nil {
+		return err
+	}
+	if mgr := runtime.GetManager(); mgr != nil {
+		mgr.InvalidateNode(id)
+	}
+	return nil
 }
 
 // GetWebCertFiles asks a node for its own web TLS certificate/key file paths,
@@ -588,6 +598,115 @@ func (s *NodeService) AggregateNodeMetric(id int, metric string, bucketSeconds i
 }
 
 func (s *NodeService) Probe(ctx context.Context, n *model.Node) (HeartbeatPatch, error) {
+	proxyURL := ""
+	if n.OutboundTag != "" {
+		if mgr := runtime.GetManager(); mgr != nil {
+			proxyURL = mgr.NodeEgressProxyURL(n.Id)
+		}
+	}
+	return s.probe(ctx, n, proxyURL)
+}
+
+func (s *NodeService) ProbeWithOutbound(ctx context.Context, n *model.Node, outboundTag string) (HeartbeatPatch, error) {
+	if outboundTag == "" {
+		return s.Probe(ctx, n)
+	}
+	proc := XrayProcess()
+	if proc == nil || !proc.IsRunning() {
+		return s.Probe(ctx, n)
+	}
+	apiPort := proc.GetAPIPort()
+	if apiPort <= 0 {
+		return s.Probe(ctx, n)
+	}
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return s.Probe(ctx, n)
+	}
+	port := listener.Addr().(*net.TCPAddr).Port
+	listener.Close()
+
+	tag := fmt.Sprintf("node-test-%d-%d", n.Id, time.Now().UnixNano())
+	proxyURL := fmt.Sprintf("socks5://127.0.0.1:%d", port)
+
+	inboundJSON, err := json.Marshal(xray.InboundConfig{
+		Listen:   json_util.RawMessage(`"127.0.0.1"`),
+		Port:     port,
+		Protocol: "socks",
+		Settings: json_util.RawMessage(`{"auth":"noauth","udp":false}`),
+		Tag:      tag,
+	})
+	if err != nil {
+		return s.Probe(ctx, n)
+	}
+
+	cfg := proc.GetConfig()
+	routing := map[string]any{}
+	if len(cfg.RouterConfig) > 0 {
+		_ = json.Unmarshal(cfg.RouterConfig, &routing)
+	}
+	rules, _ := routing["rules"].([]any)
+	rule := map[string]any{
+		"type":       "field",
+		"inboundTag": []any{tag},
+	}
+	if routingTagIsBalancer(routing, outboundTag) {
+		rule["balancerTag"] = outboundTag
+	} else {
+		rule["outboundTag"] = outboundTag
+	}
+	routing["rules"] = append([]any{rule}, rules...)
+	routingJSON, err := json.Marshal(routing)
+	if err != nil {
+		return s.Probe(ctx, n)
+	}
+	originalRoutingJSON := cfg.RouterConfig
+
+	api := xray.XrayAPI{}
+	if err := api.Init(apiPort); err != nil {
+		return s.Probe(ctx, n)
+	}
+	defer api.Close()
+
+	if err := api.AddInbound(inboundJSON); err != nil {
+		return s.Probe(ctx, n)
+	}
+	removed := false
+	defer func() {
+		if removed {
+			return
+		}
+		if err := api.DelInbound(tag); err != nil {
+			logger.Warning("remove temp node test inbound failed:", err)
+		}
+	}()
+
+	if err := api.ApplyRoutingConfig(routingJSON); err != nil {
+		return s.Probe(ctx, n)
+	}
+	defer func() {
+		restore := originalRoutingJSON
+		if len(restore) == 0 {
+			restore = []byte("{}")
+		}
+		if err := api.ApplyRoutingConfig(restore); err != nil {
+			logger.Warning("restore routing after node test failed:", err)
+		}
+	}()
+
+	patch, err := s.probe(ctx, n, proxyURL)
+	removed = true
+	if delErr := api.DelInbound(tag); delErr != nil {
+		logger.Warning("remove temp node test inbound failed:", delErr)
+	}
+	if err != nil {
+		return patch, err
+	}
+	return patch, nil
+}
+
+func (s *NodeService) probe(ctx context.Context, n *model.Node, proxyURL string) (HeartbeatPatch, error) {
 	patch := HeartbeatPatch{LastHeartbeat: time.Now().Unix()}
 
 	addr, err := netsafe.NormalizeHost(n.Address)
@@ -621,7 +740,7 @@ func (s *NodeService) Probe(ctx context.Context, n *model.Node) (HeartbeatPatch,
 	}
 	req.Header.Set("Accept", "application/json")
 
-	client, err := runtime.HTTPClientForNode(n)
+	client, err := runtime.HTTPClientForNode(n, proxyURL)
 	if err != nil {
 		patch.LastError = err.Error()
 		return patch, err

internal/web/service/setting.go

@@ -434,6 +434,26 @@ func (s *SettingService) PanelEgressProxyURL() string {
 	return ""
 }
 
+func (s *SettingService) NodeEgressProxyURL(nodeID int) string {
+	tag := NodeEgressInboundTag(nodeID)
+	proc := XrayProcess()
+	if proc == nil || !proc.IsRunning() {
+		logger.Warning("node outbound [", tag, "] is set but Xray is not running, using a direct connection")
+		return ""
+	}
+	cfg := proc.GetConfig()
+	if cfg == nil {
+		return ""
+	}
+	for i := range cfg.InboundConfigs {
+		if cfg.InboundConfigs[i].Tag == tag {
+			return fmt.Sprintf("socks5://127.0.0.1:%d", cfg.InboundConfigs[i].Port)
+		}
+	}
+	logger.Warning("node outbound [", tag, "] is set but the egress bridge is not in the running config, using a direct connection")
+	return ""
+}
+
 // NewProxiedHTTPClient returns an HTTP client that routes the panel's own
 // outbound requests through the configured panel outbound (via the loopback
 // SOCKS bridge in the running Xray). When the feature is off or the bridge

internal/web/service/xray.go

@@ -3,6 +3,7 @@ package service
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"path"
 	"path/filepath"
 	"runtime"
@@ -31,6 +32,7 @@ var (
 type XrayService struct {
 	inboundService InboundService
 	settingService SettingService
+	nodeService    NodeService
 	xrayAPI        xray.XrayAPI
 }
 
@@ -296,6 +298,13 @@ func (s *XrayService) GetXrayConfig() (*xray.Config, error) {
 		injectPanelEgress(xrayConfig, egressTag)
 	}
 
+	nodes, err := s.nodeService.GetAll()
+	if err != nil {
+		logger.Warning("read nodes for egress injection failed:", err)
+	} else {
+		injectNodeEgresses(xrayConfig, nodes)
+	}
+
 	return xrayConfig, nil
 }
 
@@ -372,6 +381,88 @@ func injectPanelEgress(cfg *xray.Config, outboundTag string) {
 	})
 }
 
+// NodeEgressInboundTag returns the loopback SOCKS inbound tag for a given node.
+func NodeEgressInboundTag(nodeID int) string {
+	return fmt.Sprintf("node-egress-%d", nodeID)
+}
+
+// nodeEgressBasePort is the first port tried for node egress bridges.
+const nodeEgressBasePort = 62800
+
+// injectNodeEgresses appends a loopback SOCKS inbound per enabled node that has
+// an OutboundTag, and prepends a routing rule sending that inbound's traffic to
+// the selected outbound tag. These bridges are hot-appliable.
+func injectNodeEgresses(cfg *xray.Config, nodes []*model.Node) {
+	routing := map[string]any{}
+	if len(cfg.RouterConfig) > 0 {
+		if err := json.Unmarshal(cfg.RouterConfig, &routing); err != nil {
+			logger.Warning("node egress: routing section is unparsable, skipping injection:", err)
+			return
+		}
+	}
+
+	used := make(map[int]struct{}, len(cfg.InboundConfigs))
+	usedTags := make(map[string]struct{}, len(cfg.InboundConfigs))
+	for i := range cfg.InboundConfigs {
+		used[cfg.InboundConfigs[i].Port] = struct{}{}
+		usedTags[cfg.InboundConfigs[i].Tag] = struct{}{}
+	}
+
+	rules, _ := routing["rules"].([]any)
+	newRules := make([]any, 0)
+
+	for _, n := range nodes {
+		if !n.Enable || n.OutboundTag == "" {
+			continue
+		}
+		tag := NodeEgressInboundTag(n.Id)
+		if _, exists := usedTags[tag]; exists {
+			logger.Warning("node egress: inbound tag [", tag, "] already exists, skipping")
+			continue
+		}
+		usedTags[tag] = struct{}{}
+
+		rule := map[string]any{
+			"type":       "field",
+			"inboundTag": []any{tag},
+		}
+		if routingTagIsBalancer(routing, n.OutboundTag) {
+			rule["balancerTag"] = n.OutboundTag
+		} else {
+			rule["outboundTag"] = n.OutboundTag
+		}
+		newRules = append(newRules, rule)
+
+		port := nodeEgressBasePort + n.Id
+		for {
+			if _, taken := used[port]; !taken {
+				break
+			}
+			port++
+		}
+		used[port] = struct{}{}
+
+		cfg.InboundConfigs = append(cfg.InboundConfigs, xray.InboundConfig{
+			Listen:   json_util.RawMessage(`"127.0.0.1"`),
+			Port:     port,
+			Protocol: "socks",
+			Settings: json_util.RawMessage(`{"auth":"noauth","udp":false}`),
+			Tag:      tag,
+		})
+	}
+
+	if len(newRules) == 0 {
+		return
+	}
+	routing["rules"] = append(newRules, rules...)
+	newRouting, err := json.Marshal(routing)
+	if err != nil {
+		logger.Warning("node egress: failed to rebuild routing section, skipping injection:", err)
+		return
+	}
+	cfg.RouterConfig = json_util.RawMessage(newRouting)
+}
+
 // routingTagIsBalancer reports whether tag names a balancer in the parsed
 // routing section. The panel-egress rule targets a balancer via balancerTag and
 // a concrete outbound via outboundTag, so the caller picks the key from this.