mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-10-02 08:56:38 +08:00
Revert "fix: 保留用户/规则高级设置并增强节点运行时恢复 (#468)"
This reverts commit 3f374df724.
This commit is contained in:
@@ -90,7 +90,7 @@
|
||||
| 表名 | Model | 特殊处理 |
|
||||
|------|-------|----------|
|
||||
| `user` | `User` | `TableName()` 返回 `"user"` (PG 保留字) |
|
||||
| `forward` | `Forward` | 增加 `proxy_protocol` 字段 |
|
||||
| `forward` | `Forward` | |
|
||||
| `forward_port` | `ForwardPort` | |
|
||||
| `node` | `Node` | |
|
||||
| `speed_limit` | `SpeedLimit` | |
|
||||
|
||||
@@ -1705,12 +1705,6 @@ 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,
|
||||
@@ -1722,10 +1716,7 @@ 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) != "" {
|
||||
if service["metadata"] == nil {
|
||||
service["metadata"] = map[string]interface{}{}
|
||||
}
|
||||
service["metadata"].(map[string]interface{})["interface"] = node.InterfaceName
|
||||
service["metadata"] = map[string]interface{}{"interface": node.InterfaceName}
|
||||
}
|
||||
if limiterID != nil && *limiterID > 0 {
|
||||
service["limiter"] = strconv.FormatInt(*limiterID, 10)
|
||||
|
||||
@@ -1,98 +0,0 @@
|
||||
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, err.Error()))
|
||||
response.WriteJSON(w, response.Err(403, "禁止将目标地址设置为内部网络"))
|
||||
return
|
||||
}
|
||||
if speedIDVal, ok := req["speedId"]; ok && speedIDVal != nil {
|
||||
@@ -1780,9 +1780,7 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
|
||||
userName = "user"
|
||||
}
|
||||
maxConn := asInt(req["maxConn"], 0)
|
||||
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)
|
||||
forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn)
|
||||
if err != nil {
|
||||
response.WriteJSON(w, response.Err(-2, err.Error()))
|
||||
return
|
||||
@@ -1855,7 +1853,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, err.Error()))
|
||||
response.WriteJSON(w, response.Err(403, "禁止将目标地址设置为内部网络"))
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -1934,9 +1932,8 @@ 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, proxyProtocol); err != nil {
|
||||
if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn); err != nil {
|
||||
response.WriteJSON(w, response.Err(-2, err.Error()))
|
||||
return
|
||||
}
|
||||
@@ -4088,7 +4085,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.ProxyProtocol,
|
||||
oldForward.SpeedID, oldForward.MaxConn,
|
||||
time.Now().UnixMilli(),
|
||||
)
|
||||
|
||||
|
||||
@@ -11,37 +11,11 @@ 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") {
|
||||
@@ -53,12 +27,12 @@ func checkSingleRemoteAddr(addr string) error {
|
||||
|
||||
ips, err := net.LookupIP(host)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not resolve address %q: %v", addr, err)
|
||||
return fmt.Errorf("could not resolve address: %v", err)
|
||||
}
|
||||
|
||||
for _, ip := range ips {
|
||||
if ip.IsLoopback() || ip.IsPrivate() {
|
||||
return fmt.Errorf("address %q resolves to internal IP: %s", addr, ip.String())
|
||||
return fmt.Errorf("address resolves to internal IP: %s", ip.String())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -13,13 +13,6 @@ 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"
|
||||
@@ -397,9 +390,6 @@ 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)
|
||||
}
|
||||
|
||||
@@ -415,7 +405,6 @@ 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 {
|
||||
@@ -424,9 +413,6 @@ 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 {
|
||||
@@ -436,86 +422,7 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -30,22 +30,21 @@ func (User) TableName() string { return "user" }
|
||||
|
||||
// Forward maps to the "forward" table.
|
||||
type Forward struct {
|
||||
ID int64 `gorm:"primaryKey;autoIncrement"`
|
||||
UserID int64 `gorm:"column:user_id;not null"`
|
||||
UserName string `gorm:"column:user_name;type:varchar(100);not null"`
|
||||
Name string `gorm:"type:varchar(100);not null"`
|
||||
TunnelID int64 `gorm:"column:tunnel_id;not null"`
|
||||
RemoteAddr string `gorm:"column:remote_addr;type:text;not null"`
|
||||
Strategy string `gorm:"type:varchar(100);not null;default:'fifo'"`
|
||||
InFlow int64 `gorm:"not null;default:0"`
|
||||
OutFlow int64 `gorm:"column:out_flow;not null;default:0"`
|
||||
CreatedTime int64 `gorm:"column:created_time;not null"`
|
||||
UpdatedTime int64 `gorm:"column:updated_time;not null"`
|
||||
Status int `gorm:"not null"`
|
||||
Inx int `gorm:"not null;default:0"`
|
||||
SpeedID sql.NullInt64 `gorm:"column:speed_id"`
|
||||
MaxConn int `gorm:"column:max_conn;not null;default:0"`
|
||||
ProxyProtocol int `gorm:"column:proxy_protocol;not null;default:0"`
|
||||
ID int64 `gorm:"primaryKey;autoIncrement"`
|
||||
UserID int64 `gorm:"column:user_id;not null"`
|
||||
UserName string `gorm:"column:user_name;type:varchar(100);not null"`
|
||||
Name string `gorm:"type:varchar(100);not null"`
|
||||
TunnelID int64 `gorm:"column:tunnel_id;not null"`
|
||||
RemoteAddr string `gorm:"column:remote_addr;type:text;not null"`
|
||||
Strategy string `gorm:"type:varchar(100);not null;default:'fifo'"`
|
||||
InFlow int64 `gorm:"not null;default:0"`
|
||||
OutFlow int64 `gorm:"column:out_flow;not null;default:0"`
|
||||
CreatedTime int64 `gorm:"column:created_time;not null"`
|
||||
UpdatedTime int64 `gorm:"column:updated_time;not null"`
|
||||
Status int `gorm:"not null"`
|
||||
Inx int `gorm:"not null;default:0"`
|
||||
SpeedID sql.NullInt64 `gorm:"column:speed_id"`
|
||||
MaxConn int `gorm:"column:max_conn;not null;default:0"`
|
||||
}
|
||||
|
||||
func (Forward) TableName() string { return "forward" }
|
||||
@@ -440,10 +439,9 @@ type ForwardBackup struct {
|
||||
CreatedTime int64 `json:"createdTime"`
|
||||
UpdatedTime int64 `json:"updatedTime"`
|
||||
Status int `json:"status"`
|
||||
Inx int `json:"inx"`
|
||||
SpeedID *int64 `json:"speedId,omitempty"`
|
||||
ForwardPorts *[]ForwardPortBackup `json:"forwardPorts,omitempty"`
|
||||
ProxyProtocol int `json:"proxyProtocol"`
|
||||
Inx int `json:"inx"`
|
||||
SpeedID *int64 `json:"speedId,omitempty"`
|
||||
ForwardPorts *[]ForwardPortBackup `json:"forwardPorts,omitempty"`
|
||||
}
|
||||
|
||||
type ForwardPortBackup struct {
|
||||
@@ -539,10 +537,9 @@ type ForwardRecord struct {
|
||||
TunnelID int64
|
||||
RemoteAddr string
|
||||
Strategy string
|
||||
Status int
|
||||
SpeedID sql.NullInt64
|
||||
MaxConn int
|
||||
ProxyProtocol int
|
||||
Status int
|
||||
SpeedID sql.NullInt64
|
||||
MaxConn int
|
||||
}
|
||||
|
||||
// TunnelRecord is a minimal tunnel view used by control plane.
|
||||
|
||||
@@ -295,17 +295,6 @@ func prepareSQLiteLegacyColumns(db *gorm.DB) error {
|
||||
}
|
||||
}
|
||||
|
||||
if m.HasTable(&model.Forward{}) {
|
||||
for _, field := range []string{"ProxyProtocol"} {
|
||||
if m.HasColumn(&model.Forward{}, field) {
|
||||
continue
|
||||
}
|
||||
if err := m.AddColumn(&model.Forward{}, field); err != nil {
|
||||
return fmt.Errorf("add forward.%s: %w", field, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -725,7 +714,6 @@ func (r *Repository) ListUsers() ([]map[string]interface{}, error) {
|
||||
"flowResetTime": u.FlowResetTime, "createdTime": u.CreatedTime,
|
||||
"updatedTime": nullableInt64(u.UpdatedTime),
|
||||
"inFlow": u.InFlow, "outFlow": u.OutFlow,
|
||||
"maxConn": u.MaxConn,
|
||||
}
|
||||
if quota := quotaMap[u.ID]; quota != nil {
|
||||
item["dailyQuotaGB"] = quota.DailyLimitGB
|
||||
@@ -781,13 +769,11 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
|
||||
Status int
|
||||
Inx int
|
||||
SpeedID sql.NullInt64
|
||||
MaxConn int
|
||||
ProxyProtocol int
|
||||
}
|
||||
|
||||
var rows []fwdRow
|
||||
err := r.db.Model(&model.Forward{}).
|
||||
Select("forward.id, forward.user_id, forward.user_name, forward.name, forward.tunnel_id, COALESCE(tunnel.name, '') AS tunnel_name, COALESCE(tunnel.traffic_ratio, 1.0) AS traffic_ratio, forward.remote_addr, COALESCE(forward.strategy, 'fifo') AS strategy, forward.in_flow, forward.out_flow, forward.created_time, forward.status, forward.inx, forward.speed_id, forward.max_conn, forward.proxy_protocol").
|
||||
Select("forward.id, forward.user_id, forward.user_name, forward.name, forward.tunnel_id, COALESCE(tunnel.name, '') AS tunnel_name, COALESCE(tunnel.traffic_ratio, 1.0) AS traffic_ratio, forward.remote_addr, COALESCE(forward.strategy, 'fifo') AS strategy, forward.in_flow, forward.out_flow, forward.created_time, forward.status, forward.inx, forward.speed_id").
|
||||
Joins("LEFT JOIN tunnel ON tunnel.id = forward.tunnel_id").
|
||||
Order("forward.inx ASC, forward.id ASC").
|
||||
Find(&rows).Error
|
||||
@@ -809,8 +795,6 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
|
||||
"remoteAddr": row.RemoteAddr, "strategy": row.Strategy,
|
||||
"inFlow": row.InFlow, "outFlow": row.OutFlow,
|
||||
"createdTime": row.CreatedTime, "status": row.Status, "inx": int64(row.Inx),
|
||||
"maxConn": row.MaxConn,
|
||||
"proxyProtocol": row.ProxyProtocol,
|
||||
}
|
||||
if row.SpeedID.Valid {
|
||||
item["speedId"] = row.SpeedID.Int64
|
||||
@@ -2003,7 +1987,6 @@ func (r *Repository) exportForwards() ([]model.ForwardBackup, error) {
|
||||
TunnelID: f.TunnelID, RemoteAddr: f.RemoteAddr, Strategy: f.Strategy,
|
||||
InFlow: f.InFlow, OutFlow: f.OutFlow, CreatedTime: f.CreatedTime,
|
||||
UpdatedTime: f.UpdatedTime, Status: f.Status, Inx: f.Inx,
|
||||
ProxyProtocol: f.ProxyProtocol,
|
||||
}
|
||||
ports, err := r.exportForwardPorts(f.ID)
|
||||
if err != nil {
|
||||
@@ -2401,13 +2384,12 @@ func importForwards(tx *gorm.DB, forwards []model.ForwardBackup, now int64) (int
|
||||
UpdatedTime: now,
|
||||
Status: f.Status,
|
||||
Inx: f.Inx,
|
||||
ProxyProtocol: f.ProxyProtocol,
|
||||
}
|
||||
err := tx.Clauses(clause.OnConflict{
|
||||
Columns: []clause.Column{{Name: "id"}},
|
||||
DoUpdates: clause.AssignmentColumns([]string{
|
||||
"user_id", "user_name", "name", "tunnel_id", "remote_addr", "strategy",
|
||||
"in_flow", "out_flow", "updated_time", "status", "inx", "proxy_protocol",
|
||||
"in_flow", "out_flow", "updated_time", "status", "inx",
|
||||
}),
|
||||
}).Create(&item).Error
|
||||
if err != nil {
|
||||
|
||||
@@ -124,17 +124,16 @@ func (r *Repository) GetForwardRecord(forwardID int64) (*model.ForwardRecord, er
|
||||
return nil, err
|
||||
}
|
||||
fr := model.ForwardRecord{
|
||||
ID: f.ID,
|
||||
UserID: f.UserID,
|
||||
UserName: f.UserName,
|
||||
Name: f.Name,
|
||||
TunnelID: f.TunnelID,
|
||||
RemoteAddr: f.RemoteAddr,
|
||||
Strategy: f.Strategy,
|
||||
Status: f.Status,
|
||||
SpeedID: f.SpeedID,
|
||||
MaxConn: f.MaxConn,
|
||||
ProxyProtocol: f.ProxyProtocol,
|
||||
ID: f.ID,
|
||||
UserID: f.UserID,
|
||||
UserName: f.UserName,
|
||||
Name: f.Name,
|
||||
TunnelID: f.TunnelID,
|
||||
RemoteAddr: f.RemoteAddr,
|
||||
Strategy: f.Strategy,
|
||||
Status: f.Status,
|
||||
SpeedID: f.SpeedID,
|
||||
MaxConn: f.MaxConn,
|
||||
}
|
||||
if strings.TrimSpace(fr.Strategy) == "" {
|
||||
fr.Strategy = "fifo"
|
||||
|
||||
@@ -1,59 +0,0 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go-backend/internal/store/model"
|
||||
)
|
||||
|
||||
func TestGetForwardRecordIncludesProxyProtocol(t *testing.T) {
|
||||
r, err := 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: 1,
|
||||
UserName: "admin",
|
||||
Name: "proxy-forward",
|
||||
TunnelID: 1,
|
||||
RemoteAddr: "1.1.1.1:443",
|
||||
Strategy: "fifo",
|
||||
CreatedTime: now,
|
||||
UpdatedTime: now,
|
||||
Status: 1,
|
||||
ProxyProtocol: 2,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("create forward: %v", err)
|
||||
}
|
||||
|
||||
forwardID := mustRepoLastInsertID(t, r)
|
||||
record, err := r.GetForwardRecord(forwardID)
|
||||
if err != nil {
|
||||
t.Fatalf("GetForwardRecord: %v", err)
|
||||
}
|
||||
if record == nil {
|
||||
t.Fatalf("expected forward record")
|
||||
}
|
||||
if record.ProxyProtocol != 2 {
|
||||
t.Fatalf("expected proxyProtocol 2, got %d", record.ProxyProtocol)
|
||||
}
|
||||
if record.MaxConn != 0 {
|
||||
t.Fatalf("expected default maxConn 0, got %d", record.MaxConn)
|
||||
}
|
||||
}
|
||||
|
||||
func mustRepoLastInsertID(t *testing.T, r *Repository) int64 {
|
||||
t.Helper()
|
||||
var id int64
|
||||
if err := r.DB().Raw("SELECT last_insert_rowid()").Row().Scan(&id); err != nil {
|
||||
t.Fatalf("last_insert_rowid: %v", err)
|
||||
}
|
||||
if id <= 0 {
|
||||
t.Fatalf("invalid last_insert_rowid %d", id)
|
||||
}
|
||||
return id
|
||||
}
|
||||
@@ -695,21 +695,20 @@ func (r *Repository) GetMinForwardPort(forwardID int64) sql.NullInt64 {
|
||||
return p
|
||||
}
|
||||
|
||||
func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}, maxConn int, proxyProtocol int) error {
|
||||
func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}, maxConn int) error {
|
||||
if r == nil || r.db == nil {
|
||||
return errors.New("repository not initialized")
|
||||
}
|
||||
return r.db.Model(&model.Forward{}).
|
||||
Where("id = ?", id).
|
||||
Updates(map[string]interface{}{
|
||||
"name": name,
|
||||
"tunnel_id": tunnelID,
|
||||
"remote_addr": remoteAddr,
|
||||
"strategy": strategy,
|
||||
"speed_id": nullInt64FromInterface(speedID),
|
||||
"max_conn": maxConn,
|
||||
"proxy_protocol": proxyProtocol,
|
||||
"updated_time": now,
|
||||
"name": name,
|
||||
"tunnel_id": tunnelID,
|
||||
"remote_addr": remoteAddr,
|
||||
"strategy": strategy,
|
||||
"speed_id": nullInt64FromInterface(speedID),
|
||||
"max_conn": maxConn,
|
||||
"updated_time": now,
|
||||
}).Error
|
||||
}
|
||||
|
||||
@@ -783,7 +782,7 @@ func (r *Repository) UpdateForwardPortBindIP(forwardID, nodeID int64, port int,
|
||||
Update("in_ip", sql.NullString{String: inIP, Valid: strings.TrimSpace(inIP) != ""}).Error
|
||||
}
|
||||
|
||||
func (r *Repository) RollbackForwardFields(id, userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, status int, speedID interface{}, maxConn int, proxyProtocol int, now int64) {
|
||||
func (r *Repository) RollbackForwardFields(id, userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, status int, speedID interface{}, maxConn int, now int64) {
|
||||
if r == nil || r.db == nil {
|
||||
return
|
||||
}
|
||||
@@ -799,7 +798,6 @@ func (r *Repository) RollbackForwardFields(id, userID int64, userName, name stri
|
||||
"status": status,
|
||||
"speed_id": nullInt64FromInterface(speedID),
|
||||
"max_conn": maxConn,
|
||||
"proxy_protocol": proxyProtocol,
|
||||
"updated_time": now,
|
||||
}).Error
|
||||
}
|
||||
@@ -1260,28 +1258,27 @@ func (r *Repository) EnsureUserTunnelGrant(userID, tunnelID int64) (int64, bool,
|
||||
return ut.ID, true, nil
|
||||
}
|
||||
|
||||
func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}, maxConn int, proxyProtocol int) (int64, error) {
|
||||
func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}, maxConn int) (int64, error) {
|
||||
if r == nil || r.db == nil {
|
||||
return 0, errors.New("repository not initialized")
|
||||
}
|
||||
var forwardID int64
|
||||
err := r.db.Transaction(func(tx *gorm.DB) error {
|
||||
fwd := model.Forward{
|
||||
UserID: userID,
|
||||
UserName: userName,
|
||||
Name: name,
|
||||
TunnelID: tunnelID,
|
||||
RemoteAddr: remoteAddr,
|
||||
Strategy: strategy,
|
||||
InFlow: 0,
|
||||
OutFlow: 0,
|
||||
CreatedTime: now,
|
||||
UpdatedTime: now,
|
||||
Status: 1,
|
||||
Inx: inx,
|
||||
MaxConn: maxConn,
|
||||
SpeedID: nullInt64FromInterface(speedID),
|
||||
ProxyProtocol: proxyProtocol,
|
||||
UserID: userID,
|
||||
UserName: userName,
|
||||
Name: name,
|
||||
TunnelID: tunnelID,
|
||||
RemoteAddr: remoteAddr,
|
||||
Strategy: strategy,
|
||||
InFlow: 0,
|
||||
OutFlow: 0,
|
||||
CreatedTime: now,
|
||||
UpdatedTime: now,
|
||||
Status: 1,
|
||||
Inx: inx,
|
||||
MaxConn: maxConn,
|
||||
SpeedID: nullInt64FromInterface(speedID),
|
||||
}
|
||||
if err := tx.Create(&fwd).Error; err != nil {
|
||||
return err
|
||||
|
||||
@@ -96,7 +96,6 @@ func TestMaxConnLimit(t *testing.T) {
|
||||
"remoteAddr": "1.1.1.1:443",
|
||||
"strategy": "fifo",
|
||||
"maxConn": 42,
|
||||
"proxyProtocol": 2,
|
||||
}
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
@@ -121,47 +120,6 @@ func TestMaxConnLimit(t *testing.T) {
|
||||
t.Fatalf("get forward ID: %v", err)
|
||||
}
|
||||
|
||||
listOut := requestContractEnvelope(t, router, adminToken, "/api/v1/forward/list", nil)
|
||||
if listOut.Code != 0 {
|
||||
t.Fatalf("expected /forward/list success, got code=%d msg=%s", listOut.Code, listOut.Msg)
|
||||
}
|
||||
|
||||
rows := mustContractSlice(t, listOut.Data, "forward list")
|
||||
var target map[string]interface{}
|
||||
for _, row := range rows {
|
||||
item, ok := row.(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("expected forward item to be object, got %T", row)
|
||||
}
|
||||
idVal, ok := item["id"].(float64)
|
||||
if !ok {
|
||||
t.Fatalf("expected forward id to be float64, got %T", item["id"])
|
||||
}
|
||||
if int64(idVal) == forwardID {
|
||||
target = item
|
||||
break
|
||||
}
|
||||
}
|
||||
if target == nil {
|
||||
t.Fatalf("forward %d not found in /forward/list response", forwardID)
|
||||
}
|
||||
|
||||
maxConnVal, ok := target["maxConn"].(float64)
|
||||
if !ok {
|
||||
t.Fatalf("expected maxConn to be float64, got %T (%v)", target["maxConn"], target["maxConn"])
|
||||
}
|
||||
if int(maxConnVal) != 42 {
|
||||
t.Fatalf("expected maxConn 42 in /forward/list, got %v", maxConnVal)
|
||||
}
|
||||
|
||||
proxyProtocolVal, ok := target["proxyProtocol"].(float64)
|
||||
if !ok {
|
||||
t.Fatalf("expected proxyProtocol to be float64, got %T (%v)", target["proxyProtocol"], target["proxyProtocol"])
|
||||
}
|
||||
if int(proxyProtocolVal) != 2 {
|
||||
t.Fatalf("expected proxyProtocol 2 in /forward/list, got %v", proxyProtocolVal)
|
||||
}
|
||||
|
||||
commandMu.Lock()
|
||||
defer commandMu.Unlock()
|
||||
|
||||
|
||||
@@ -349,9 +349,9 @@ func TestBackupExportImportRestoreContracts(t *testing.T) {
|
||||
tunnelID := mustLastInsertID(t, r, "backup-forward-tunnel")
|
||||
|
||||
if err := r.DB().Exec(`
|
||||
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx, proxy_protocol)
|
||||
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`, 1, "admin_user", "backup-forward", tunnelID, "127.0.0.1:9000", "fifo", 0, 0, now, now, 1, 88, 2).Error; err != nil {
|
||||
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
|
||||
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`, 1, "admin_user", "backup-forward", tunnelID, "127.0.0.1:9000", "fifo", 0, 0, now, now, 1, 88).Error; err != nil {
|
||||
t.Fatalf("seed forward for backup: %v", err)
|
||||
}
|
||||
forwardID := mustLastInsertID(t, r, "backup-forward")
|
||||
@@ -412,9 +412,6 @@ func TestBackupExportImportRestoreContracts(t *testing.T) {
|
||||
if !ok {
|
||||
t.Fatalf("expected forwardPorts for forward %d in payload", forwardID)
|
||||
}
|
||||
if proxyProtocol, ok := forwardMap["proxyProtocol"].(float64); !ok || int(proxyProtocol) != 2 {
|
||||
t.Fatalf("expected exported proxyProtocol 2 for forward %d, got %v", forwardID, forwardMap["proxyProtocol"])
|
||||
}
|
||||
for _, p := range portsRaw {
|
||||
portMap, ok := p.(map[string]interface{})
|
||||
if !ok {
|
||||
@@ -478,14 +475,6 @@ func TestBackupExportImportRestoreContracts(t *testing.T) {
|
||||
t.Fatalf("expected forward_port node=%d port=%d after import, got %v", nodeID, port, after)
|
||||
}
|
||||
}
|
||||
|
||||
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 after import: %v", err)
|
||||
}
|
||||
if proxyProtocol != 2 {
|
||||
t.Fatalf("expected proxy_protocol 2 after import, got %d", proxyProtocol)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("backup export tolerates nullable legacy tunnel chain fields", func(t *testing.T) {
|
||||
|
||||
@@ -1,59 +0,0 @@
|
||||
package contract_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go-backend/internal/auth"
|
||||
)
|
||||
|
||||
func TestUserListReturnsMaxConn(t *testing.T) {
|
||||
secret := "contract-jwt-secret"
|
||||
router, repo := setupContractRouter(t, secret)
|
||||
now := time.Now().UnixMilli()
|
||||
|
||||
if err := repo.DB().Exec(`
|
||||
INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, max_conn, created_time, updated_time, status)
|
||||
VALUES(2, 'max_conn_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 10, 37, ?, ?, 1)
|
||||
`, now, now).Error; err != nil {
|
||||
t.Fatalf("insert user: %v", err)
|
||||
}
|
||||
|
||||
adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
|
||||
if err != nil {
|
||||
t.Fatalf("generate admin token: %v", err)
|
||||
}
|
||||
|
||||
out := requestContractEnvelope(t, router, adminToken, "/api/v1/user/list", map[string]interface{}{})
|
||||
if out.Code != 0 {
|
||||
t.Fatalf("expected /user/list success, got code=%d msg=%s", out.Code, out.Msg)
|
||||
}
|
||||
|
||||
rows := mustContractSlice(t, out.Data, "user list")
|
||||
var target map[string]interface{}
|
||||
for _, row := range rows {
|
||||
item, ok := row.(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("expected user item to be object, got %T", row)
|
||||
}
|
||||
idVal, ok := item["id"].(float64)
|
||||
if !ok {
|
||||
t.Fatalf("expected user id to be float64, got %T", item["id"])
|
||||
}
|
||||
if int64(idVal) == 2 {
|
||||
target = item
|
||||
break
|
||||
}
|
||||
}
|
||||
if target == nil {
|
||||
t.Fatalf("user 2 not found in /user/list response")
|
||||
}
|
||||
|
||||
maxConnVal, ok := target["maxConn"].(float64)
|
||||
if !ok {
|
||||
t.Fatalf("expected maxConn to be float64, got %T (%v)", target["maxConn"], target["maxConn"])
|
||||
}
|
||||
if int(maxConnVal) != 37 {
|
||||
t.Fatalf("expected maxConn 37 in /user/list, got %v", maxConnVal)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user