tcjlj/flvx
1
0
Fork 0
mirror of https://github.com/Sagit-chu/flvx.git synced 2026-10-08 10:46:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into opencode/happy-island

Browse Source
This commit is contained in:
sagit
2026-02-14 04:32:50 +00:00
parent d05c8a2ea4 e00e41bb64
commit 47c05c3d02
6 changed files with 327 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go-backend/internal/http/handler/control_plane.go
+31 -3
View File
@@ -200,6 +200,26 @@ func (h *Handler) listForwardPorts(forwardID int64) ([]forwardPortRecord, error)
return result, nil
}
func (h *Handler) isTunnelSelectedTLSProtocol(tunnelID int64) (bool, error) {
row := h.repo.DB().QueryRow(`
SELECT protocol
FROM chain_tunnel
WHERE tunnel_id = ? AND chain_type = '3'
ORDER BY id ASC
LIMIT 1
`, tunnelID)
var protocol sql.NullString
if err := row.Scan(&protocol); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return false, nil
}
return false, err
}
return isTLSTunnelProtocol(protocol.String), nil
}
func (h *Handler) getNodeRecord(nodeID int64) (*nodeRecord, error) {
row := h.repo.DB().QueryRow(`
SELECT id, name, server_ip, server_ip_v4, server_ip_v6, status, port, tcp_listen_addr, udp_listen_addr, interface_name, is_remote, remote_url, remote_token, remote_config
@@ -346,6 +366,10 @@ func (h *Handler) syncForwardServices(forward *forwardRecord, method string, all
return err
}
serviceBase := buildForwardServiceBase(forward.ID, forward.UserID, userTunnelID)
tunnelTLSProtocol, err := h.isTunnelSelectedTLSProtocol(forward.TunnelID)
if err != nil {
return err
}
for _, fp := range ports {
if limiterID != nil && speed != nil {
@@ -356,7 +380,7 @@ func (h *Handler) syncForwardServices(forward *forwardRecord, method string, all
if err != nil {
return err
}
services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, limiterID)
services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, limiterID, tunnelTLSProtocol)
_, err = h.sendNodeCommand(node.ID, method, services, true, false)
if err != nil && allowFallbackAdd && method == "UpdateService" {
_, err = h.sendNodeCommand(node.ID, "AddService", services, true, false)
@@ -1095,7 +1119,7 @@ func isNotFoundError(err error) bool {
return strings.Contains(msg, "not found") || strings.Contains(msg, "不存在")
}
func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, limiterID *int64) []map[string]interface{} {
func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, limiterID *int64, tunnelTLSProtocol bool) []map[string]interface{} {
protocols := []string{"tcp", "udp"}
services := make([]map[string]interface{}, 0, 2)
targets := splitRemoteTargets(forward.RemoteAddr)
@@ -1128,7 +1152,11 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
},
}
if protocol == "udp" {
service["listener"].(map[string]interface{})["metadata"] = map[string]interface{}{"keepAlive": true}
listenerMetadata := map[string]interface{}{"keepAlive": true}
if tunnelTLSProtocol {
listenerMetadata["ttl"] = "10s"
}
service["listener"].(map[string]interface{})["metadata"] = listenerMetadata
}
if tunnel != nil && tunnel.Type == 2 {
service["handler"].(map[string]interface{})["chain"] = fmt.Sprintf("chains_%d", forward.TunnelID)
go-backend/internal/http/handler/mutations.go
+4 -3
View File
@@ -2613,9 +2613,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64
}
for _, inNode := range state.InNodes {
if node := state.Nodes[inNode.NodeID]; node != nil && node.IsRemote == 1 {
continue
}
node := state.Nodes[inNode.NodeID]
targets := state.OutNodes
if len(state.ChainHops) > 0 {
targets = state.ChainHops[0]
@@ -2625,6 +2623,9 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64
return createdChains, createdServices, err
}
if _, err := h.sendNodeCommand(inNode.NodeID, "AddChains", chainData, true, false); err != nil {
if node != nil && node.IsRemote == 1 && shouldDeferTunnelRuntimeApplyError(err) {
continue
}
return createdChains, createdServices, fmt.Errorf("入口节点 %s 下发转发链失败: %w", nodeDisplayName(state.Nodes[inNode.NodeID]), err)
}
createdChains = append(createdChains, inNode.NodeID)
go-backend/internal/store/sqlite/repository.go
+38 -3
View File
@@ -2330,13 +2330,25 @@ func (r *Repository) exportTunnels() ([]TunnelBackup, error) {
var tunnels []TunnelBackup
for rows.Next() {
var t TunnelBackup
var protocol sql.NullString
var updatedTime sql.NullInt64
var inIP sql.NullString
if err := rows.Scan(&t.ID, &t.Name, &t.TrafficRatio, &t.Type, &t.Protocol, &t.Flow, &t.CreatedTime, &t.UpdatedTime, &t.Status, &inIP, &t.Inx); err != nil {
var inx sql.NullInt64
if err := rows.Scan(&t.ID, &t.Name, &t.TrafficRatio, &t.Type, &protocol, &t.Flow, &t.CreatedTime, &updatedTime, &t.Status, &inIP, &inx); err != nil {
return nil, err
}
if protocol.Valid {
t.Protocol = protocol.String
}
if updatedTime.Valid {
t.UpdatedTime = updatedTime.Int64
}
if inIP.Valid {
t.InIP = inIP.String
}
if inx.Valid {
t.Inx = int(inx.Int64)
}
// Export chain tunnels
chainTunnels, err := r.exportChainTunnels(t.ID)
if err != nil {
@@ -2362,12 +2374,23 @@ func (r *Repository) exportChainTunnels(tunnelID int64) ([]ChainTunnelBackup, er
for rows.Next() {
var ct ChainTunnelBackup
var port sql.NullInt64
if err := rows.Scan(&ct.ID, &ct.TunnelID, &ct.ChainType, &ct.NodeID, &port, &ct.Strategy, &ct.Inx, &ct.Protocol); err != nil {
var strategy, protocol sql.NullString
var inx sql.NullInt64
if err := rows.Scan(&ct.ID, &ct.TunnelID, &ct.ChainType, &ct.NodeID, &port, &strategy, &inx, &protocol); err != nil {
return nil, err
}
if port.Valid {
ct.Port = int(port.Int64)
}
if strategy.Valid {
ct.Strategy = strategy.String
}
if inx.Valid {
ct.Inx = int(inx.Int64)
}
if protocol.Valid {
ct.Protocol = protocol.String
}
chainTunnels = append(chainTunnels, ct)
}
return chainTunnels, rows.Err()
@@ -2386,9 +2409,21 @@ func (r *Repository) exportForwards() ([]ForwardBackup, error) {
var forwards []ForwardBackup
for rows.Next() {
var f ForwardBackup
if err := rows.Scan(&f.ID, &f.UserID, &f.UserName, &f.Name, &f.TunnelID, &f.RemoteAddr, &f.Strategy, &f.InFlow, &f.OutFlow, &f.CreatedTime, &f.UpdatedTime, &f.Status, &f.Inx); err != nil {
var strategy sql.NullString
var updatedTime sql.NullInt64
var inx sql.NullInt64
if err := rows.Scan(&f.ID, &f.UserID, &f.UserName, &f.Name, &f.TunnelID, &f.RemoteAddr, &strategy, &f.InFlow, &f.OutFlow, &f.CreatedTime, &updatedTime, &f.Status, &inx); err != nil {
return nil, err
}
if strategy.Valid {
f.Strategy = strategy.String
}
if updatedTime.Valid {
f.UpdatedTime = updatedTime.Int64
}
if inx.Valid {
f.Inx = int(inx.Int64)
}
forwards = append(forwards, f)
}
return forwards, rows.Err()