fix(forward): split proxy protocol directions

Closes #520
This commit is contained in:
sagit
2026-06-21 21:03:47 +08:00
committed by GitHub
parent 3ce320da5a
commit 559a793b09
13 changed files with 505 additions and 259 deletions
@@ -1757,6 +1757,7 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
services := make([]map[string]interface{}, 0, 2) services := make([]map[string]interface{}, 0, 2)
targets := splitRemoteTargets(forward.RemoteAddr) targets := splitRemoteTargets(forward.RemoteAddr)
strategy := strings.TrimSpace(forward.Strategy) strategy := strings.TrimSpace(forward.Strategy)
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(forward.ProxyProtocol, forward.ProxyProtocolReceive, forward.ProxyProtocolSend)
if strategy == "" { if strategy == "" {
strategy = "fifo" strategy = "fifo"
} }
@@ -1801,12 +1802,16 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
if runtimeLimiters.TrafficLimiter != "" { if runtimeLimiters.TrafficLimiter != "" {
service["limiter"] = runtimeLimiters.TrafficLimiter service["limiter"] = runtimeLimiters.TrafficLimiter
} }
if forward.ProxyProtocol > 0 { if proxyProtocolReceive > 0 {
serviceMetadata := ensureServiceMetadata(service)
serviceMetadata["proxyProtocol"] = proxyProtocolReceive
}
if proxyProtocolSend > 0 {
handlerConfig := service["handler"].(map[string]interface{}) handlerConfig := service["handler"].(map[string]interface{})
if handlerConfig["metadata"] == nil { if handlerConfig["metadata"] == nil {
handlerConfig["metadata"] = map[string]interface{}{} handlerConfig["metadata"] = map[string]interface{}{}
} }
handlerConfig["metadata"].(map[string]interface{})["proxyProtocol"] = forward.ProxyProtocol handlerConfig["metadata"].(map[string]interface{})["proxyProtocol"] = proxyProtocolSend
} }
if protocol == "udp" { if protocol == "udp" {
listenerMetadata := map[string]interface{}{ listenerMetadata := map[string]interface{}{
@@ -1819,10 +1824,8 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
service["handler"].(map[string]interface{})["chain"] = fmt.Sprintf("chains_%d", forward.TunnelID) service["handler"].(map[string]interface{})["chain"] = fmt.Sprintf("chains_%d", forward.TunnelID)
} }
if tunnel != nil && tunnel.Type == 1 && strings.TrimSpace(node.InterfaceName) != "" { if tunnel != nil && tunnel.Type == 1 && strings.TrimSpace(node.InterfaceName) != "" {
if service["metadata"] == nil { serviceMetadata := ensureServiceMetadata(service)
service["metadata"] = map[string]interface{}{} serviceMetadata["interface"] = node.InterfaceName
}
service["metadata"].(map[string]interface{})["interface"] = node.InterfaceName
} }
services = append(services, service) services = append(services, service)
} }
@@ -1841,6 +1844,25 @@ func buildForwarderNodes(targets []string) []map[string]interface{} {
return nodes return nodes
} }
func ensureServiceMetadata(service map[string]interface{}) map[string]interface{} {
if service["metadata"] == nil {
service["metadata"] = map[string]interface{}{}
}
metadata, ok := service["metadata"].(map[string]interface{})
if !ok {
metadata = map[string]interface{}{}
service["metadata"] = metadata
}
return metadata
}
func normalizeForwardProxyProtocol(legacy, receive, send int) (int, int) {
if send == 0 && legacy > 0 {
send = legacy
}
return receive, send
}
func processServerAddress(serverAddr string) string { func processServerAddress(serverAddr string) string {
serverAddr = normalizeServerAddressInput(serverAddr) serverAddr = normalizeServerAddressInput(serverAddr)
if serverAddr == "" { if serverAddr == "" {
@@ -9,14 +9,15 @@ import (
"go-backend/internal/store/repo" "go-backend/internal/store/repo"
) )
func TestBuildForwardServiceConfigsSendsProxyProtocolToForwardHandler(t *testing.T) { func TestBuildForwardServiceConfigsAppliesProxyProtocolReceiveAndSendIndependently(t *testing.T) {
forward := &forwardRecord{ forward := &forwardRecord{
ID: 1, ID: 1,
UserID: 2, UserID: 2,
TunnelID: 3, TunnelID: 3,
RemoteAddr: "1.1.1.1:443", RemoteAddr: "1.1.1.1:443",
Strategy: "fifo", Strategy: "fifo",
ProxyProtocol: 2, ProxyProtocolReceive: 1,
ProxyProtocolSend: 2,
} }
tunnel := &tunnelRecord{Type: 1} tunnel := &tunnelRecord{Type: 1}
node := &nodeRecord{ node := &nodeRecord{
@@ -38,8 +39,8 @@ func TestBuildForwardServiceConfigsSendsProxyProtocolToForwardHandler(t *testing
if serviceMetadata["interface"] != "eth0" { if serviceMetadata["interface"] != "eth0" {
t.Fatalf("expected interface metadata eth0, got %v", serviceMetadata["interface"]) t.Fatalf("expected interface metadata eth0, got %v", serviceMetadata["interface"])
} }
if _, ok := serviceMetadata["proxyProtocol"]; ok { if serviceMetadata["proxyProtocol"] != 1 {
t.Fatalf("proxyProtocol should not be listener metadata: %v", serviceMetadata) t.Fatalf("expected service proxyProtocol 1 for receive mode, got %v", serviceMetadata["proxyProtocol"])
} }
handlerConfig, ok := service["handler"].(map[string]interface{}) handlerConfig, ok := service["handler"].(map[string]interface{})
@@ -56,6 +57,46 @@ func TestBuildForwardServiceConfigsSendsProxyProtocolToForwardHandler(t *testing
} }
} }
func TestBuildForwardServiceConfigsKeepsLegacyProxyProtocolAsSend(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{
TCPListenAddr: "0.0.0.0",
UDPListenAddr: "0.0.0.0",
}
services := buildForwardServiceConfigs("1_2_3", forward, tunnel, node, 4001, "", forwardRuntimeLimiters{})
if len(services) != 2 {
t.Fatalf("expected 2 services, got %d", len(services))
}
for _, service := range services {
serviceMetadata, _ := service["metadata"].(map[string]interface{})
if _, ok := serviceMetadata["proxyProtocol"]; ok {
t.Fatalf("legacy proxyProtocol should not enable receive mode: %v", serviceMetadata)
}
handlerConfig, ok := service["handler"].(map[string]interface{})
if !ok {
t.Fatalf("expected handler config map, got %T", service["handler"])
}
handlerMetadata, ok := handlerConfig["metadata"].(map[string]interface{})
if !ok {
t.Fatalf("expected handler metadata map, got %T", handlerConfig["metadata"])
}
if handlerMetadata["proxyProtocol"] != 2 {
t.Fatalf("expected legacy proxyProtocol to send version 2, got %v", handlerMetadata["proxyProtocol"])
}
}
}
func TestRollbackForwardMutationRestoresProxyProtocol(t *testing.T) { func TestRollbackForwardMutationRestoresProxyProtocol(t *testing.T) {
r, err := repo.Open(":memory:") r, err := repo.Open(":memory:")
if err != nil { if err != nil {
@@ -2143,8 +2143,10 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
ipMaxConn = 0 ipMaxConn = 0
} }
proxyProtocol := asInt(req["proxyProtocol"], 0) proxyProtocol := asInt(req["proxyProtocol"], 0)
proxyProtocolReceive := asInt(req["proxyProtocolReceive"], 0)
proxyProtocolSend := asInt(req["proxyProtocolSend"], proxyProtocol)
forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, ipMaxConn, nullableInt(ipSpeedID), proxyProtocol) forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, ipMaxConn, nullableInt(ipSpeedID), proxyProtocol, proxyProtocolReceive, proxyProtocolSend)
if err != nil { if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error())) response.WriteJSON(w, response.Err(-2, err.Error()))
return return
@@ -2333,8 +2335,10 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
ipMaxConn = 0 ipMaxConn = 0
} }
proxyProtocol := asInt(req["proxyProtocol"], forward.ProxyProtocol) proxyProtocol := asInt(req["proxyProtocol"], forward.ProxyProtocol)
proxyProtocolReceive := asInt(req["proxyProtocolReceive"], forward.ProxyProtocolReceive)
proxyProtocolSend := asInt(req["proxyProtocolSend"], forward.ProxyProtocolSend)
if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, ipMaxConn, newIPSpeedID, proxyProtocol); err != nil { if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, ipMaxConn, newIPSpeedID, proxyProtocol, proxyProtocolReceive, proxyProtocolSend); err != nil {
response.WriteJSON(w, response.Err(-2, err.Error())) response.WriteJSON(w, response.Err(-2, err.Error()))
return return
} }
@@ -4722,7 +4726,7 @@ func (h *Handler) rollbackForwardMutation(oldForward *forwardRecord, oldPorts []
h.repo.RollbackForwardFields( h.repo.RollbackForwardFields(
oldForward.ID, oldForward.UserID, oldForward.UserName, oldForward.Name, oldForward.ID, oldForward.UserID, oldForward.UserName, oldForward.Name,
oldForward.TunnelID, oldForward.RemoteAddr, oldForward.Strategy, oldForward.Status, oldForward.TunnelID, oldForward.RemoteAddr, oldForward.Strategy, oldForward.Status,
oldForward.SpeedID, oldForward.MaxConn, oldForward.IPMaxConn, oldForward.IPSpeedID, oldForward.ProxyProtocol, oldForward.SpeedID, oldForward.MaxConn, oldForward.IPMaxConn, oldForward.IPSpeedID, oldForward.ProxyProtocol, oldForward.ProxyProtocolReceive, oldForward.ProxyProtocolSend,
time.Now().UnixMilli(), time.Now().UnixMilli(),
) )
@@ -490,7 +490,7 @@ func seedForwardForNftables(t *testing.T, h *Handler, tunnelID, nodeID int64, re
now := time.Now().UnixMilli() now := time.Now().UnixMilli()
forwardID, err := h.repo.CreateForwardTx( forwardID, err := h.repo.CreateForwardTx(
1, "admin", "nft-forward", tunnelID, remoteAddr, "fifo", now, 1, 1, "admin", "nft-forward", tunnelID, remoteAddr, "fifo", now, 1,
[]int64{nodeID}, 20000, "", nil, 0, 0, nil, 0, []int64{nodeID}, 20000, "", nil, 0, 0, nil, 0, 0, 0,
) )
if err != nil { if err != nil {
t.Fatalf("create forward: %v", err) t.Fatalf("create forward: %v", err)
+6
View File
@@ -49,6 +49,8 @@ type Forward struct {
IPMaxConn int `gorm:"column:ip_max_conn;not null;default:0"` IPMaxConn int `gorm:"column:ip_max_conn;not null;default:0"`
IPSpeedID sql.NullInt64 `gorm:"column:ip_speed_id"` IPSpeedID sql.NullInt64 `gorm:"column:ip_speed_id"`
ProxyProtocol int `gorm:"column:proxy_protocol;not null;default:0"` ProxyProtocol int `gorm:"column:proxy_protocol;not null;default:0"`
ProxyProtocolReceive int `gorm:"column:proxy_protocol_receive;not null;default:0"`
ProxyProtocolSend int `gorm:"column:proxy_protocol_send;not null;default:0"`
} }
func (Forward) TableName() string { return "forward" } func (Forward) TableName() string { return "forward" }
@@ -505,6 +507,8 @@ type ForwardBackup struct {
IPSpeedID *int64 `json:"ipSpeedId,omitempty"` IPSpeedID *int64 `json:"ipSpeedId,omitempty"`
ForwardPorts *[]ForwardPortBackup `json:"forwardPorts,omitempty"` ForwardPorts *[]ForwardPortBackup `json:"forwardPorts,omitempty"`
ProxyProtocol int `json:"proxyProtocol"` ProxyProtocol int `json:"proxyProtocol"`
ProxyProtocolReceive int `json:"proxyProtocolReceive,omitempty"`
ProxyProtocolSend int `json:"proxyProtocolSend,omitempty"`
} }
type ForwardPortBackup struct { type ForwardPortBackup struct {
@@ -606,6 +610,8 @@ type ForwardRecord struct {
IPMaxConn int IPMaxConn int
IPSpeedID sql.NullInt64 IPSpeedID sql.NullInt64
ProxyProtocol int ProxyProtocol int
ProxyProtocolReceive int
ProxyProtocolSend int
} }
// TunnelRecord is a minimal tunnel view used by control plane. // TunnelRecord is a minimal tunnel view used by control plane.
+12 -3
View File
@@ -424,7 +424,7 @@ func prepareSQLiteLegacyColumns(db *gorm.DB) error {
} }
if m.HasTable(&model.Forward{}) { if m.HasTable(&model.Forward{}) {
for _, field := range []string{"MaxConn", "IPMaxConn", "IPSpeedID", "ProxyProtocol"} { for _, field := range []string{"MaxConn", "IPMaxConn", "IPSpeedID", "ProxyProtocol", "ProxyProtocolReceive", "ProxyProtocolSend"} {
if m.HasColumn(&model.Forward{}, field) { if m.HasColumn(&model.Forward{}, field) {
continue continue
} }
@@ -964,11 +964,13 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
IPSpeedID sql.NullInt64 IPSpeedID sql.NullInt64
IPSpeedLimitName string IPSpeedLimitName string
ProxyProtocol int ProxyProtocol int
ProxyProtocolReceive int
ProxyProtocolSend int
} }
var rows []fwdRow var rows []fwdRow
err := r.db.Model(&model.Forward{}). 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.ip_max_conn, forward.ip_speed_id, COALESCE(ip_speed_limit.name, '') AS ip_speed_limit_name, 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, forward.max_conn, forward.ip_max_conn, forward.ip_speed_id, COALESCE(ip_speed_limit.name, '') AS ip_speed_limit_name, forward.proxy_protocol, forward.proxy_protocol_receive, forward.proxy_protocol_send").
Joins("LEFT JOIN tunnel ON tunnel.id = forward.tunnel_id"). Joins("LEFT JOIN tunnel ON tunnel.id = forward.tunnel_id").
Joins("LEFT JOIN speed_limit AS ip_speed_limit ON ip_speed_limit.id = forward.ip_speed_id"). Joins("LEFT JOIN speed_limit AS ip_speed_limit ON ip_speed_limit.id = forward.ip_speed_id").
Order("forward.inx ASC, forward.id ASC"). Order("forward.inx ASC, forward.id ASC").
@@ -979,6 +981,7 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
items := make([]map[string]interface{}, 0, len(rows)) items := make([]map[string]interface{}, 0, len(rows))
for _, row := range rows { for _, row := range rows {
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(row.ProxyProtocol, row.ProxyProtocolReceive, row.ProxyProtocolSend)
inIP, inPort, err := resolveForwardIngress(r.db, row.ID, row.TunnelID) inIP, inPort, err := resolveForwardIngress(r.db, row.ID, row.TunnelID)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -994,6 +997,8 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
"maxConn": row.MaxConn, "maxConn": row.MaxConn,
"ipMaxConn": row.IPMaxConn, "ipMaxConn": row.IPMaxConn,
"proxyProtocol": row.ProxyProtocol, "proxyProtocol": row.ProxyProtocol,
"proxyProtocolReceive": proxyProtocolReceive,
"proxyProtocolSend": proxyProtocolSend,
} }
if row.SpeedID.Valid { if row.SpeedID.Valid {
item["speedId"] = row.SpeedID.Int64 item["speedId"] = row.SpeedID.Int64
@@ -2197,6 +2202,8 @@ func (r *Repository) exportForwards() ([]model.ForwardBackup, error) {
UpdatedTime: f.UpdatedTime, Status: f.Status, Inx: f.Inx, UpdatedTime: f.UpdatedTime, Status: f.Status, Inx: f.Inx,
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: f.ProxyProtocolReceive,
ProxyProtocolSend: f.ProxyProtocolSend,
} }
if f.SpeedID.Valid { if f.SpeedID.Valid {
v := f.SpeedID.Int64 v := f.SpeedID.Int64
@@ -2647,12 +2654,14 @@ func importForwards(tx *gorm.DB, forwards []model.ForwardBackup, now int64) (int
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
IPSpeedID: sql.NullInt64{Int64: nullableBackupInt64(f.IPSpeedID), Valid: f.IPSpeedID != nil && *f.IPSpeedID > 0}, IPSpeedID: sql.NullInt64{Int64: nullableBackupInt64(f.IPSpeedID), Valid: f.IPSpeedID != nil && *f.IPSpeedID > 0},
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: f.ProxyProtocolReceive,
ProxyProtocolSend: f.ProxyProtocolSend,
} }
err := tx.Clauses(clause.OnConflict{ err := tx.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "id"}}, Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns([]string{ DoUpdates: clause.AssignmentColumns([]string{
"user_id", "user_name", "name", "tunnel_id", "remote_addr", "strategy", "user_id", "user_name", "name", "tunnel_id", "remote_addr", "strategy",
"in_flow", "out_flow", "updated_time", "status", "inx", "speed_id", "ip_max_conn", "ip_speed_id", "proxy_protocol", "in_flow", "out_flow", "updated_time", "status", "inx", "speed_id", "ip_max_conn", "ip_speed_id", "proxy_protocol", "proxy_protocol_receive", "proxy_protocol_send",
}), }),
}).Create(&item).Error }).Create(&item).Error
if err != nil { if err != nil {
@@ -44,6 +44,7 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode
} }
rows := make([]model.ForwardRecord, 0, len(forwards)) rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards { for _, f := range forwards {
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(f.ProxyProtocol, f.ProxyProtocolReceive, f.ProxyProtocolSend)
rows = append(rows, model.ForwardRecord{ rows = append(rows, model.ForwardRecord{
ID: f.ID, ID: f.ID,
UserID: f.UserID, UserID: f.UserID,
@@ -58,6 +59,8 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
IPSpeedID: f.IPSpeedID, IPSpeedID: f.IPSpeedID,
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: proxyProtocolReceive,
ProxyProtocolSend: proxyProtocolSend,
}) })
} }
for i := range rows { for i := range rows {
@@ -120,6 +120,7 @@ func (r *Repository) ListActiveForwardsByUser(userID int64) ([]model.ForwardReco
} }
rows := make([]model.ForwardRecord, 0, len(forwards)) rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards { for _, f := range forwards {
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(f.ProxyProtocol, f.ProxyProtocolReceive, f.ProxyProtocolSend)
rows = append(rows, model.ForwardRecord{ rows = append(rows, model.ForwardRecord{
ID: f.ID, ID: f.ID,
UserID: f.UserID, UserID: f.UserID,
@@ -134,6 +135,8 @@ func (r *Repository) ListActiveForwardsByUser(userID int64) ([]model.ForwardReco
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
IPSpeedID: f.IPSpeedID, IPSpeedID: f.IPSpeedID,
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: proxyProtocolReceive,
ProxyProtocolSend: proxyProtocolSend,
}) })
} }
for i := range rows { for i := range rows {
@@ -155,6 +158,7 @@ func (r *Repository) ListActiveForwardsByUserTunnel(userID, tunnelID int64) ([]m
} }
rows := make([]model.ForwardRecord, 0, len(forwards)) rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards { for _, f := range forwards {
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(f.ProxyProtocol, f.ProxyProtocolReceive, f.ProxyProtocolSend)
rows = append(rows, model.ForwardRecord{ rows = append(rows, model.ForwardRecord{
ID: f.ID, ID: f.ID,
UserID: f.UserID, UserID: f.UserID,
@@ -169,6 +173,8 @@ func (r *Repository) ListActiveForwardsByUserTunnel(userID, tunnelID int64) ([]m
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
IPSpeedID: f.IPSpeedID, IPSpeedID: f.IPSpeedID,
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: proxyProtocolReceive,
ProxyProtocolSend: proxyProtocolSend,
}) })
} }
for i := range rows { for i := range rows {
@@ -190,6 +196,7 @@ func (r *Repository) ListForwardsByUserAndTunnel(userID, tunnelID int64) ([]mode
} }
rows := make([]model.ForwardRecord, 0, len(forwards)) rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards { for _, f := range forwards {
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(f.ProxyProtocol, f.ProxyProtocolReceive, f.ProxyProtocolSend)
rows = append(rows, model.ForwardRecord{ rows = append(rows, model.ForwardRecord{
ID: f.ID, ID: f.ID,
UserID: f.UserID, UserID: f.UserID,
@@ -204,6 +211,8 @@ func (r *Repository) ListForwardsByUserAndTunnel(userID, tunnelID int64) ([]mode
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
IPSpeedID: f.IPSpeedID, IPSpeedID: f.IPSpeedID,
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: proxyProtocolReceive,
ProxyProtocolSend: proxyProtocolSend,
}) })
} }
for i := range rows { for i := range rows {
@@ -226,6 +235,7 @@ func (r *Repository) GetForwardRecord(forwardID int64) (*model.ForwardRecord, er
} }
return nil, err return nil, err
} }
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(f.ProxyProtocol, f.ProxyProtocolReceive, f.ProxyProtocolSend)
fr := model.ForwardRecord{ fr := model.ForwardRecord{
ID: f.ID, ID: f.ID,
UserID: f.UserID, UserID: f.UserID,
@@ -240,6 +250,8 @@ func (r *Repository) GetForwardRecord(forwardID int64) (*model.ForwardRecord, er
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
IPSpeedID: f.IPSpeedID, IPSpeedID: f.IPSpeedID,
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: proxyProtocolReceive,
ProxyProtocolSend: proxyProtocolSend,
} }
if strings.TrimSpace(fr.Strategy) == "" { if strings.TrimSpace(fr.Strategy) == "" {
fr.Strategy = "fifo" fr.Strategy = "fifo"
@@ -160,7 +160,7 @@ func TestForwardRepositoryPersistsPerIPLimits(t *testing.T) {
defer r.Close() defer r.Close()
now := time.Now().UnixMilli() now := time.Now().UnixMilli()
forwardID, err := r.CreateForwardTx(1, "admin", "per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, []int64{3}, 24000, "", nil, 0, 5, int64(21), 0) forwardID, err := r.CreateForwardTx(1, "admin", "per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, []int64{3}, 24000, "", nil, 0, 5, int64(21), 0, 0, 0)
if err != nil { if err != nil {
t.Fatalf("CreateForwardTx: %v", err) t.Fatalf("CreateForwardTx: %v", err)
} }
@@ -175,7 +175,7 @@ func TestForwardRepositoryPersistsPerIPLimits(t *testing.T) {
t.Fatalf("expected created ipSpeedId 21, got %+v", record.IPSpeedID) t.Fatalf("expected created ipSpeedId 21, got %+v", record.IPSpeedID)
} }
if err := r.UpdateForward(forwardID, "per-ip-forward", 2, "2.2.2.2:443", "fifo", now+1, nil, 0, 9, int64(22), 0); err != nil { if err := r.UpdateForward(forwardID, "per-ip-forward", 2, "2.2.2.2:443", "fifo", now+1, nil, 0, 9, int64(22), 0, 0, 0); err != nil {
t.Fatalf("UpdateForward: %v", err) t.Fatalf("UpdateForward: %v", err)
} }
record, err = r.GetForwardRecord(forwardID) record, err = r.GetForwardRecord(forwardID)
@@ -216,6 +216,83 @@ func TestForwardRepositoryPersistsPerIPLimits(t *testing.T) {
} }
} }
func TestForwardRepositoryPersistsProxyProtocolReceiveAndSend(t *testing.T) {
r, err := Open(":memory:")
if err != nil {
t.Fatalf("open repo: %v", err)
}
defer r.Close()
now := time.Now().UnixMilli()
forwardID, err := r.CreateForwardTx(1, "admin", "proxy-protocol-forward", 2, "1.1.1.1:443", "fifo", now, 1, []int64{3}, 24000, "", nil, 0, 0, nil, 0, 1, 2)
if err != nil {
t.Fatalf("CreateForwardTx: %v", err)
}
record, err := r.GetForwardRecord(forwardID)
if err != nil {
t.Fatalf("GetForwardRecord after create: %v", err)
}
if record.ProxyProtocolReceive != 1 || record.ProxyProtocolSend != 2 {
t.Fatalf("expected proxyProtocol receive/send 1/2 after create, got %d/%d", record.ProxyProtocolReceive, record.ProxyProtocolSend)
}
if err := r.UpdateForward(forwardID, "proxy-protocol-forward", 2, "2.2.2.2:443", "fifo", now+1, nil, 0, 0, nil, 0, 2, 1); err != nil {
t.Fatalf("UpdateForward: %v", err)
}
record, err = r.GetForwardRecord(forwardID)
if err != nil {
t.Fatalf("GetForwardRecord after update: %v", err)
}
if record.ProxyProtocolReceive != 2 || record.ProxyProtocolSend != 1 {
t.Fatalf("expected proxyProtocol receive/send 2/1 after update, got %d/%d", record.ProxyProtocolReceive, record.ProxyProtocolSend)
}
records, err := r.ListForwardsByTunnel(2)
if err != nil {
t.Fatalf("ListForwardsByTunnel: %v", err)
}
if len(records) != 1 {
t.Fatalf("expected 1 listed record, got %d", len(records))
}
if records[0].ProxyProtocolReceive != 2 || records[0].ProxyProtocolSend != 1 {
t.Fatalf("expected listed proxyProtocol receive/send 2/1, got %d/%d", records[0].ProxyProtocolReceive, records[0].ProxyProtocolSend)
}
}
func TestForwardRepositoryMapsLegacyProxyProtocolToSend(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: 2,
UserName: "user",
Name: "legacy-proxy-protocol-forward",
TunnelID: 9,
RemoteAddr: "1.1.1.1:443",
Strategy: "fifo",
CreatedTime: now,
UpdatedTime: now,
Status: 1,
ProxyProtocol: 2,
}).Error; err != nil {
t.Fatalf("create legacy forward: %v", err)
}
forwardID := mustRepoLastInsertID(t, r)
record, err := r.GetForwardRecord(forwardID)
if err != nil {
t.Fatalf("GetForwardRecord: %v", err)
}
if record.ProxyProtocolReceive != 0 || record.ProxyProtocolSend != 2 {
t.Fatalf("expected legacy proxyProtocol to map to receive/send 0/2, got %d/%d", record.ProxyProtocolReceive, record.ProxyProtocolSend)
}
}
func TestRollbackForwardFieldsRestoresPerIPLimits(t *testing.T) { func TestRollbackForwardFieldsRestoresPerIPLimits(t *testing.T) {
r, err := Open(":memory:") r, err := Open(":memory:")
if err != nil { if err != nil {
@@ -224,15 +301,15 @@ func TestRollbackForwardFieldsRestoresPerIPLimits(t *testing.T) {
defer r.Close() defer r.Close()
now := time.Now().UnixMilli() now := time.Now().UnixMilli()
forwardID, err := r.CreateForwardTx(1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, nil, 0, "", nil, 7, 5, int64(21), 2) forwardID, err := r.CreateForwardTx(1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, nil, 0, "", nil, 7, 5, int64(21), 2, 0, 2)
if err != nil { if err != nil {
t.Fatalf("CreateForwardTx: %v", err) t.Fatalf("CreateForwardTx: %v", err)
} }
if err := r.UpdateForward(forwardID, "rollback-per-ip-forward", 2, "2.2.2.2:443", "fifo", now+1, nil, 0, 0, nil, 0); err != nil { if err := r.UpdateForward(forwardID, "rollback-per-ip-forward", 2, "2.2.2.2:443", "fifo", now+1, nil, 0, 0, nil, 0, 0, 0); err != nil {
t.Fatalf("UpdateForward: %v", err) t.Fatalf("UpdateForward: %v", err)
} }
r.RollbackForwardFields(forwardID, 1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", 1, nil, 7, 5, int64(21), 2, now+2) r.RollbackForwardFields(forwardID, 1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", 1, nil, 7, 5, int64(21), 2, 0, 2, now+2)
record, err := r.GetForwardRecord(forwardID) record, err := r.GetForwardRecord(forwardID)
if err != nil { if err != nil {
@@ -726,10 +726,11 @@ func (r *Repository) GetMinForwardPort(forwardID int64) sql.NullInt64 {
return p return p
} }
func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}, maxConn int, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int) error { func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}, maxConn int, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int, proxyProtocolReceive int, proxyProtocolSend int) error {
if r == nil || r.db == nil { if r == nil || r.db == nil {
return errors.New("repository not initialized") return errors.New("repository not initialized")
} }
_, proxyProtocolSend = normalizeForwardProxyProtocol(proxyProtocol, proxyProtocolReceive, proxyProtocolSend)
return r.db.Model(&model.Forward{}). return r.db.Model(&model.Forward{}).
Where("id = ?", id). Where("id = ?", id).
Updates(map[string]interface{}{ Updates(map[string]interface{}{
@@ -742,6 +743,8 @@ func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remote
"ip_max_conn": ipMaxConn, "ip_max_conn": ipMaxConn,
"ip_speed_id": nullInt64FromInterface(ipSpeedID), "ip_speed_id": nullInt64FromInterface(ipSpeedID),
"proxy_protocol": proxyProtocol, "proxy_protocol": proxyProtocol,
"proxy_protocol_receive": proxyProtocolReceive,
"proxy_protocol_send": proxyProtocolSend,
"updated_time": now, "updated_time": now,
}).Error }).Error
} }
@@ -819,10 +822,11 @@ func (r *Repository) UpdateForwardPortBindIP(forwardID, nodeID int64, port int,
Update("in_ip", sql.NullString{String: inIP, Valid: strings.TrimSpace(inIP) != ""}).Error 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, ipMaxConn int, ipSpeedID interface{}, 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, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int, proxyProtocolReceive int, proxyProtocolSend int, now int64) {
if r == nil || r.db == nil { if r == nil || r.db == nil {
return return
} }
_, proxyProtocolSend = normalizeForwardProxyProtocol(proxyProtocol, proxyProtocolReceive, proxyProtocolSend)
_ = r.db.Model(&model.Forward{}). _ = r.db.Model(&model.Forward{}).
Where("id = ?", id). Where("id = ?", id).
Updates(map[string]interface{}{ Updates(map[string]interface{}{
@@ -838,6 +842,8 @@ func (r *Repository) RollbackForwardFields(id, userID int64, userName, name stri
"ip_max_conn": ipMaxConn, "ip_max_conn": ipMaxConn,
"ip_speed_id": nullInt64FromInterface(ipSpeedID), "ip_speed_id": nullInt64FromInterface(ipSpeedID),
"proxy_protocol": proxyProtocol, "proxy_protocol": proxyProtocol,
"proxy_protocol_receive": proxyProtocolReceive,
"proxy_protocol_send": proxyProtocolSend,
"updated_time": now, "updated_time": now,
}).Error }).Error
} }
@@ -1298,10 +1304,11 @@ func (r *Repository) EnsureUserTunnelGrant(userID, tunnelID int64) (int64, bool,
return ut.ID, true, nil 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, ipMaxConn int, ipSpeedID interface{}, 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, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int, proxyProtocolReceive int, proxyProtocolSend int) (int64, error) {
if r == nil || r.db == nil { if r == nil || r.db == nil {
return 0, errors.New("repository not initialized") return 0, errors.New("repository not initialized")
} }
_, proxyProtocolSend = normalizeForwardProxyProtocol(proxyProtocol, proxyProtocolReceive, proxyProtocolSend)
var forwardID int64 var forwardID int64
err := r.db.Transaction(func(tx *gorm.DB) error { err := r.db.Transaction(func(tx *gorm.DB) error {
fwd := model.Forward{ fwd := model.Forward{
@@ -1322,6 +1329,8 @@ func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnel
IPMaxConn: ipMaxConn, IPMaxConn: ipMaxConn,
IPSpeedID: nullInt64FromInterface(ipSpeedID), IPSpeedID: nullInt64FromInterface(ipSpeedID),
ProxyProtocol: proxyProtocol, ProxyProtocol: proxyProtocol,
ProxyProtocolReceive: proxyProtocolReceive,
ProxyProtocolSend: proxyProtocolSend,
} }
if err := tx.Create(&fwd).Error; err != nil { if err := tx.Create(&fwd).Error; err != nil {
return err return err
@@ -207,6 +207,7 @@ func (r *Repository) ListActiveForwardsByNode(nodeID int64) ([]model.ForwardReco
} }
rows := make([]model.ForwardRecord, 0, len(forwards)) rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards { for _, f := range forwards {
proxyProtocolReceive, proxyProtocolSend := normalizeForwardProxyProtocol(f.ProxyProtocol, f.ProxyProtocolReceive, f.ProxyProtocolSend)
rows = append(rows, model.ForwardRecord{ rows = append(rows, model.ForwardRecord{
ID: f.ID, ID: f.ID,
UserID: f.UserID, UserID: f.UserID,
@@ -221,6 +222,8 @@ func (r *Repository) ListActiveForwardsByNode(nodeID int64) ([]model.ForwardReco
IPMaxConn: f.IPMaxConn, IPMaxConn: f.IPMaxConn,
IPSpeedID: f.IPSpeedID, IPSpeedID: f.IPSpeedID,
ProxyProtocol: f.ProxyProtocol, ProxyProtocol: f.ProxyProtocol,
ProxyProtocolReceive: proxyProtocolReceive,
ProxyProtocolSend: proxyProtocolSend,
}) })
} }
for i := range rows { for i := range rows {
@@ -237,3 +240,10 @@ func defaultString(value, fallback string) string {
} }
return value return value
} }
func normalizeForwardProxyProtocol(legacy, receive, send int) (int, int) {
if send == 0 && legacy > 0 {
send = legacy
}
return receive, send
}
+4
View File
@@ -82,6 +82,8 @@ export interface ForwardApiItem {
ipSpeedLimitName?: string; ipSpeedLimitName?: string;
maxConn?: number; maxConn?: number;
proxyProtocol?: number; proxyProtocol?: number;
proxyProtocolReceive?: number;
proxyProtocolSend?: number;
inx?: number; inx?: number;
[key: string]: unknown; [key: string]: unknown;
} }
@@ -422,6 +424,8 @@ export interface ForwardMutationPayload {
ipSpeedId?: number | null; ipSpeedId?: number | null;
maxConn?: number; maxConn?: number;
proxyProtocol?: number; proxyProtocol?: number;
proxyProtocolReceive?: number;
proxyProtocolSend?: number;
} }
export interface SpeedLimitMutationPayload { export interface SpeedLimitMutationPayload {
+55 -6
View File
@@ -130,6 +130,8 @@ interface Forward {
ipSpeedId?: number | null; ipSpeedId?: number | null;
ipSpeedLimitName?: string; ipSpeedLimitName?: string;
proxyProtocol?: number; proxyProtocol?: number;
proxyProtocolReceive?: number;
proxyProtocolSend?: number;
} }
interface Tunnel { interface Tunnel {
@@ -169,6 +171,8 @@ interface ForwardForm {
ipSpeedId: number | null; ipSpeedId: number | null;
maxConn?: number; maxConn?: number;
proxyProtocol?: number; proxyProtocol?: number;
proxyProtocolReceive?: number;
proxyProtocolSend?: number;
} }
interface ForwardUserGroup { interface ForwardUserGroup {
@@ -600,6 +604,16 @@ const mapForwardApiItems = (items: ForwardApiItem[]): Forward[] => {
typeof forward.proxyProtocol === "number" typeof forward.proxyProtocol === "number"
? forward.proxyProtocol ? forward.proxyProtocol
: undefined, : undefined,
proxyProtocolReceive:
typeof forward.proxyProtocolReceive === "number"
? forward.proxyProtocolReceive
: 0,
proxyProtocolSend:
typeof forward.proxyProtocolSend === "number"
? forward.proxyProtocolSend
: typeof forward.proxyProtocol === "number"
? forward.proxyProtocol
: 0,
serviceRunning: forward.status === 1, serviceRunning: forward.status === 1,
})); }));
}; };
@@ -1337,6 +1351,8 @@ export default function ForwardPage() {
ipSpeedId: null, ipSpeedId: null,
maxConn: 0, maxConn: 0,
proxyProtocol: 0, proxyProtocol: 0,
proxyProtocolReceive: 0,
proxyProtocolSend: 0,
}); });
const [inIpTouched, setInIpTouched] = useState(false); const [inIpTouched, setInIpTouched] = useState(false);
@@ -2128,6 +2144,8 @@ export default function ForwardPage() {
ipMaxConn: 0, ipMaxConn: 0,
ipSpeedId: null, ipSpeedId: null,
proxyProtocol: 0, proxyProtocol: 0,
proxyProtocolReceive: 0,
proxyProtocolSend: 0,
}); });
setErrors({}); setErrors({});
setModalOpen(true); setModalOpen(true);
@@ -2152,6 +2170,8 @@ export default function ForwardPage() {
ipSpeedId: normalizeSpeedId(forward.ipSpeedId), ipSpeedId: normalizeSpeedId(forward.ipSpeedId),
maxConn: forward.maxConn ?? 0, maxConn: forward.maxConn ?? 0,
proxyProtocol: forward.proxyProtocol ?? 0, proxyProtocol: forward.proxyProtocol ?? 0,
proxyProtocolReceive: forward.proxyProtocolReceive ?? 0,
proxyProtocolSend: forward.proxyProtocolSend ?? forward.proxyProtocol ?? 0,
}); });
setErrors({}); setErrors({});
setModalOpen(true); setModalOpen(true);
@@ -2285,7 +2305,9 @@ export default function ForwardPage() {
ipMaxConn: form.ipMaxConn, ipMaxConn: form.ipMaxConn,
...(isAdmin ? { ipSpeedId: normalizedIPSpeedId } : {}), ...(isAdmin ? { ipSpeedId: normalizedIPSpeedId } : {}),
maxConn: form.maxConn, maxConn: form.maxConn,
proxyProtocol: form.proxyProtocol, proxyProtocol: form.proxyProtocolSend,
proxyProtocolReceive: form.proxyProtocolReceive,
proxyProtocolSend: form.proxyProtocolSend,
}; };
res = await updateForward(updateData); res = await updateForward(updateData);
@@ -2301,7 +2323,9 @@ export default function ForwardPage() {
ipMaxConn: form.ipMaxConn, ipMaxConn: form.ipMaxConn,
...(isAdmin ? { ipSpeedId: normalizedIPSpeedId } : {}), ...(isAdmin ? { ipSpeedId: normalizedIPSpeedId } : {}),
maxConn: form.maxConn, maxConn: form.maxConn,
proxyProtocol: form.proxyProtocol, proxyProtocol: form.proxyProtocolSend,
proxyProtocolReceive: form.proxyProtocolReceive,
proxyProtocolSend: form.proxyProtocolSend,
}; };
res = await createForward(createData); res = await createForward(createData);
@@ -4968,18 +4992,21 @@ export default function ForwardPage() {
setForm((prev) => ({ ...prev, ipMaxConn: value })); setForm((prev) => ({ ...prev, ipMaxConn: value }));
}} }}
/> />
<div className="grid grid-cols-1 gap-4 md:grid-cols-2">
<Select <Select
description="启用 PROXY protocol,用于透传客户端真实 IP" description="入口监听接收 PROXY protocol,用于读取上游传入的真实客户端 IP。"
label="Proxy Protocol" label="Proxy Protocol 接收"
placeholder="禁用" placeholder="禁用"
selectedKeys={[String(form.proxyProtocol || 0)]} selectedKeys={[
String(form.proxyProtocolReceive || 0),
]}
variant="bordered" variant="bordered"
onSelectionChange={(keys) => { onSelectionChange={(keys) => {
const selectedKey = Array.from(keys)[0] as string; const selectedKey = Array.from(keys)[0] as string;
setForm((prev) => ({ setForm((prev) => ({
...prev, ...prev,
proxyProtocol: Number(selectedKey), proxyProtocolReceive: Number(selectedKey),
})); }));
}} }}
> >
@@ -4987,6 +5014,28 @@ export default function ForwardPage() {
<SelectItem key="1">Version 1</SelectItem> <SelectItem key="1">Version 1</SelectItem>
<SelectItem key="2">Version 2</SelectItem> <SelectItem key="2">Version 2</SelectItem>
</Select> </Select>
<Select
description="连接目标地址时发送 PROXY protocol,用于向下游透传客户端真实 IP。"
label="Proxy Protocol 发送"
placeholder="禁用"
selectedKeys={[String(form.proxyProtocolSend || 0)]}
variant="bordered"
onSelectionChange={(keys) => {
const selectedKey = Array.from(keys)[0] as string;
const proxyProtocolSend = Number(selectedKey);
setForm((prev) => ({
...prev,
proxyProtocol: proxyProtocolSend,
proxyProtocolSend,
}));
}}
>
<SelectItem key="0">禁用</SelectItem>
<SelectItem key="1">Version 1</SelectItem>
<SelectItem key="2">Version 2</SelectItem>
</Select>
</div>
{isAdmin && ( {isAdmin && (
<Select <Select
label="规则限速" label="规则限速"