fix: 保留用户/规则高级设置并增强节点运行时恢复 (#468)

* fix: 用户编辑时 maxConn 字段未正确回填

- User 接口添加 maxConn 类型定义
- handleEdit 中回填 maxConn 值
- normalizeUserItem 添加 maxConn 字段处理

解决编辑用户时最大连接数显示为 0 的问题

* fix: 保留转发设置并增强节点运行时恢复
This commit is contained in:
sagit
2026-04-25 18:12:13 +08:00
committed by GitHub
parent 9b923a2d0b
commit 3f374df724
30 changed files with 744 additions and 123 deletions
@@ -1705,6 +1705,12 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
if cLimiterName != "" {
service["climiter"] = cLimiterName
}
if forward.ProxyProtocol > 0 {
if service["metadata"] == nil {
service["metadata"] = map[string]interface{}{}
}
service["metadata"].(map[string]interface{})["proxyProtocol"] = forward.ProxyProtocol
}
if protocol == "udp" {
listenerMetadata := map[string]interface{}{
"keepAlive": true,
@@ -1716,7 +1722,10 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
service["handler"].(map[string]interface{})["chain"] = fmt.Sprintf("chains_%d", forward.TunnelID)
}
if tunnel != nil && tunnel.Type == 1 && strings.TrimSpace(node.InterfaceName) != "" {
service["metadata"] = map[string]interface{}{"interface": node.InterfaceName}
if service["metadata"] == nil {
service["metadata"] = map[string]interface{}{}
}
service["metadata"].(map[string]interface{})["interface"] = node.InterfaceName
}
if limiterID != nil && *limiterID > 0 {
service["limiter"] = strconv.FormatInt(*limiterID, 10)
@@ -0,0 +1,98 @@
package handler
import (
"testing"
"time"
"go-backend/internal/store/model"
"go-backend/internal/store/repo"
)
func TestBuildForwardServiceConfigsPreservesProxyProtocolWithInterfaceMetadata(t *testing.T) {
forward := &forwardRecord{
ID: 1,
UserID: 2,
TunnelID: 3,
RemoteAddr: "1.1.1.1:443",
Strategy: "fifo",
ProxyProtocol: 2,
}
tunnel := &tunnelRecord{Type: 1}
node := &nodeRecord{
InterfaceName: "eth0",
TCPListenAddr: "0.0.0.0",
UDPListenAddr: "0.0.0.0",
}
services := buildForwardServiceConfigs("1_2_3", forward, tunnel, node, 4001, "", nil, "")
if len(services) != 2 {
t.Fatalf("expected 2 services, got %d", len(services))
}
for _, service := range services {
metadata, ok := service["metadata"].(map[string]interface{})
if !ok {
t.Fatalf("expected metadata map, got %T", service["metadata"])
}
if metadata["interface"] != "eth0" {
t.Fatalf("expected interface metadata eth0, got %v", metadata["interface"])
}
if metadata["proxyProtocol"] != 2 {
t.Fatalf("expected proxyProtocol 2, got %v", metadata["proxyProtocol"])
}
}
}
func TestRollbackForwardMutationRestoresProxyProtocol(t *testing.T) {
r, err := repo.Open(":memory:")
if err != nil {
t.Fatalf("open repo: %v", err)
}
defer r.Close()
now := time.Now().UnixMilli()
if err := r.DB().Create(&model.Forward{
UserID: 2,
UserName: "rollback-user",
Name: "rollback-forward",
TunnelID: 3,
RemoteAddr: "9.9.9.9:443",
Strategy: "fifo",
CreatedTime: now,
UpdatedTime: now,
Status: 1,
ProxyProtocol: 2,
}).Error; err != nil {
t.Fatalf("create forward: %v", err)
}
forwardID := mustLastInsertID(t, r, "rollback-forward")
if err := r.DB().Model(&model.Forward{}).Where("id = ?", forwardID).Updates(map[string]interface{}{
"name": "changed-forward",
"proxy_protocol": 0,
"updated_time": now + 1,
}).Error; err != nil {
t.Fatalf("mutate forward: %v", err)
}
h := &Handler{repo: r}
h.rollbackForwardMutation(&forwardRecord{
ID: forwardID,
UserID: 2,
UserName: "rollback-user",
Name: "rollback-forward",
TunnelID: 3,
RemoteAddr: "9.9.9.9:443",
Strategy: "fifo",
Status: 1,
ProxyProtocol: 2,
}, nil)
var proxyProtocol int
if err := r.DB().Raw("SELECT proxy_protocol FROM forward WHERE id = ?", forwardID).Row().Scan(&proxyProtocol); err != nil {
t.Fatalf("query proxy_protocol: %v", err)
}
if proxyProtocol != 2 {
t.Fatalf("expected proxyProtocol restored to 2, got %d", proxyProtocol)
}
}
@@ -1728,7 +1728,7 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
}
if roleID != 0 {
if err := IsSafeRemoteAddr(remoteAddr); err != nil {
response.WriteJSON(w, response.Err(403, "禁止将目标地址设置为内部网络"))
response.WriteJSON(w, response.Err(403, err.Error()))
return
}
if speedIDVal, ok := req["speedId"]; ok && speedIDVal != nil {
@@ -1780,7 +1780,9 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
userName = "user"
}
maxConn := asInt(req["maxConn"], 0)
forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn)
proxyProtocol := asInt(req["proxyProtocol"], 0)
forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, proxyProtocol)
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
@@ -1853,7 +1855,7 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
}
if actorRole != 0 {
if err := IsSafeRemoteAddr(remoteAddr); err != nil {
response.WriteJSON(w, response.Err(403, "禁止将目标地址设置为内部网络"))
response.WriteJSON(w, response.Err(403, err.Error()))
return
}
}
@@ -1932,8 +1934,9 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
}
now := time.Now().UnixMilli()
maxConn := asInt(req["maxConn"], forward.MaxConn)
proxyProtocol := asInt(req["proxyProtocol"], forward.ProxyProtocol)
if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn); err != nil {
if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, proxyProtocol); err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
@@ -4085,7 +4088,7 @@ func (h *Handler) rollbackForwardMutation(oldForward *forwardRecord, oldPorts []
h.repo.RollbackForwardFields(
oldForward.ID, oldForward.UserID, oldForward.UserName, oldForward.Name,
oldForward.TunnelID, oldForward.RemoteAddr, oldForward.Strategy, oldForward.Status,
oldForward.SpeedID, oldForward.MaxConn,
oldForward.SpeedID, oldForward.MaxConn, oldForward.ProxyProtocol,
time.Now().UnixMilli(),
)
@@ -11,11 +11,37 @@ var DisableSafeRemoteAddrCheckForTesting = false
// IsSafeRemoteAddr checks if a given address is safe to connect to (prevents SSRF/Open Proxy).
// It resolves domains to IPs to prevent DNS rebinding attacks pointing to internal networks.
// Supports multiple addresses separated by commas or newlines (one per line).
func IsSafeRemoteAddr(addr string) error {
if DisableSafeRemoteAddrCheckForTesting {
return nil
}
for _, part := range splitRemoteParts(addr) {
if err := checkSingleRemoteAddr(part); err != nil {
return err
}
}
return nil
}
// splitRemoteParts splits a multi-address string by commas and newlines.
func splitRemoteParts(addr string) []string {
addr = strings.ReplaceAll(addr, "\n", ",")
addr = strings.ReplaceAll(addr, "\r", ",")
parts := strings.Split(addr, ",")
out := make([]string, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part != "" {
out = append(out, part)
}
}
return out
}
// checkSingleRemoteAddr validates a single address.
func checkSingleRemoteAddr(addr string) error {
host, _, err := net.SplitHostPort(addr)
if err != nil {
if strings.Contains(err.Error(), "missing port in address") {
@@ -27,12 +53,12 @@ func IsSafeRemoteAddr(addr string) error {
ips, err := net.LookupIP(host)
if err != nil {
return fmt.Errorf("could not resolve address: %v", err)
return fmt.Errorf("could not resolve address %q: %v", addr, err)
}
for _, ip := range ips {
if ip.IsLoopback() || ip.IsPrivate() {
return fmt.Errorf("address resolves to internal IP: %s", ip.String())
return fmt.Errorf("address %q resolves to internal IP: %s", addr, ip.String())
}
}
@@ -13,6 +13,13 @@ import (
"go-backend/internal/http/response"
)
// failedForward tracks a forward that failed redeployment, for retry.
type failedForward struct {
id int64
forward *forwardRecord
err error
}
const (
githubRepo = "Sagit-chu/flvx"
githubAPIBase = "https://api.github.com"
@@ -390,6 +397,9 @@ func (h *Handler) consumeNodePendingUpgradeRedeploy(nodeID int64) bool {
func (h *Handler) onNodeOnline(nodeID int64) {
h.consumeNodePendingUpgradeRedeploy(nodeID)
// Always redeploy rules on reconnection, not just for pending upgrade nodes.
// This handles cases where the node restarted and lost its in-memory config
// before persistence had time to flush, or if the panel also restarted.
h.redeployNodeRuntimeAfterUpgrade(nodeID)
}
@@ -405,6 +415,7 @@ func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) {
return
}
// First pass: deploy everything
tunnelFailed := make(map[int64]struct{})
for _, tunnelID := range tunnelIDs {
if err := h.redeployTunnelAndForwards(tunnelID); err != nil {
@@ -413,6 +424,9 @@ func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) {
}
}
// Collect forwards that failed independently (not skipped due to tunnel failure)
var failedForwards []failedForward
for _, forwardID := range forwardIDs {
forward, getErr := h.getForwardRecord(forwardID)
if getErr != nil || forward == nil {
@@ -422,7 +436,86 @@ func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) {
continue
}
if err := h.syncForwardServices(forward, "UpdateService", true); err != nil {
failedForwards = append(failedForwards, failedForward{id: forwardID, forward: forward, err: err})
fmt.Printf("post-upgrade redeploy: forward %d failed on node %d: %v\n", forwardID, nodeID, err)
}
}
// Retry failed items with exponential backoff (max 3 attempts)
h.retryFailedRedeploys(nodeID, tunnelFailed, failedForwards)
}
// isRetryableError returns true if the error looks transient and worth retrying.
func isRetryableError(err error) bool {
if err == nil {
return false
}
msg := strings.ToLower(err.Error())
// Skip non-retryable errors: not-found, already-exists, validation errors
if strings.Contains(msg, "not found") || strings.Contains(msg, "不存在") {
return false
}
if strings.Contains(msg, "already exists") || strings.Contains(msg, "已存在") {
return false
}
// Everything else (timeout, connection lost, port in use, etc.) is retryable
return true
}
// retryFailedRedeploys retries failed tunnels and forwards with exponential backoff.
func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]struct{}, failedForwards []failedForward) {
if len(tunnelFailed) == 0 && len(failedForwards) == 0 {
return
}
const maxRetries = 3
baseDelay := time.Second
for attempt := 1; attempt <= maxRetries; attempt++ {
delay := baseDelay * time.Duration(1<<uint(attempt-1)) // 1s, 2s, 4s
time.Sleep(delay)
// Retry failed tunnels
for tunnelID := range tunnelFailed {
if err := h.redeployTunnelAndForwards(tunnelID); err == nil {
delete(tunnelFailed, tunnelID)
fmt.Printf("post-upgrade redeploy retry: tunnel %d succeeded on node %d (attempt %d)\n", tunnelID, nodeID, attempt)
} else if !isRetryableError(err) {
delete(tunnelFailed, tunnelID) // Non-retryable, don't retry again
} else {
fmt.Printf("post-upgrade redeploy retry: tunnel %d still failing on node %d (attempt %d): %v\n", tunnelID, nodeID, attempt, err)
}
}
// Retry failed forwards
var stillFailed []failedForward
for _, ff := range failedForwards {
if _, skipped := tunnelFailed[ff.forward.TunnelID]; skipped {
stillFailed = append(stillFailed, ff) // Tunnel still failed, skip forward
continue
}
if err := h.syncForwardServices(ff.forward, "UpdateService", true); err == nil {
fmt.Printf("post-upgrade redeploy retry: forward %d succeeded on node %d (attempt %d)\n", ff.id, nodeID, attempt)
} else if !isRetryableError(err) {
// Non-retryable, drop it
} else {
stillFailed = append(stillFailed, ff)
fmt.Printf("post-upgrade redeploy retry: forward %d still failing on node %d (attempt %d): %v\n", ff.id, nodeID, attempt, err)
}
}
failedForwards = stillFailed
if len(tunnelFailed) == 0 && len(failedForwards) == 0 {
fmt.Printf("post-upgrade redeploy retry: all items recovered on node %d\n", nodeID)
return
}
}
// Final summary
for tunnelID := range tunnelFailed {
fmt.Printf("post-upgrade redeploy: tunnel %d permanently failed on node %d after retries\n", tunnelID, nodeID)
}
for _, ff := range failedForwards {
fmt.Printf("post-upgrade redeploy: forward %d permanently failed on node %d after retries\n", ff.id, nodeID)
}
}