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

feat: implement consumer side panel peering logic

Browse Source
This commit is contained in:
sagit
2026-02-08 09:03:56 +00:00
parent 1c72fb233a
commit 11dc21e46f
11 changed files with 954 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go-backend/internal/http/handler/federation.go
+249
View File
@@ -0,0 +1,249 @@
package handler
import (
"encoding/json"
"fmt"
"net/http"
"strings"
"time"
"go-backend/internal/http/client"
"go-backend/internal/http/response"
)
type federationTunnelRequest struct {
Protocol string `json:"protocol"`
RemotePort int `json:"remotePort"`
Target string `json:"target"`
}
type nodeImportRequest struct {
RemoteURL string `json:"remoteUrl"`
Token string `json:"token"`
}
func (h *Handler) nodeImport(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
response.WriteJSON(w, response.ErrDefault("Invalid method"))
return
}
var req nodeImportRequest
if err := decodeJSON(r.Body, &req); err != nil {
response.WriteJSON(w, response.ErrDefault("Invalid JSON"))
return
}
if req.RemoteURL == "" || req.Token == "" {
response.WriteJSON(w, response.ErrDefault("Remote URL and Token are required"))
return
}
fc := client.NewFederationClient()
info, err := fc.Connect(req.RemoteURL, req.Token)
if err != nil {
response.WriteJSON(w, response.Err(-2, "Failed to connect: "+err.Error()))
return
}
// Prepare config json for local storage (metadata about limits)
configData := map[string]interface{}{
"shareId": info.ShareID,
"maxBandwidth": info.MaxBandwidth,
"expiryTime": info.ExpiryTime,
"portRangeStart": info.PortRangeStart,
"portRangeEnd": info.PortRangeEnd,
}
configBytes, _ := json.Marshal(configData)
db := h.repo.DB()
inx := nextIndex(db, "node")
now := time.Now().UnixMilli()
_, err = db.Exec(`
INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx, is_remote, remote_url, remote_token, remote_config)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, 0, 0, 0, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)
`,
fmt.Sprintf("%s (Remote)", info.NodeName),
randomToken(16), // Dummy secret
info.ServerIP,
"", "", // v4/v6 unknown, use server_ip
"0", // port range not applicable for remote
"",
"",
now, now,
info.Status,
"[::]", "[::]",
inx,
req.RemoteURL,
req.Token,
string(configBytes),
)
if err != nil {
response.WriteJSON(w, response.Err(-2, "Database error: "+err.Error()))
return
}
response.WriteJSON(w, response.OKEmpty())
}
func (h *Handler) authPeer(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
authHeader := r.Header.Get("Authorization")
if authHeader == "" {
response.WriteJSON(w, response.Err(401, "Missing Authorization header"))
return
}
parts := strings.Split(authHeader, " ")
if len(parts) != 2 || parts[0] != "Bearer" {
response.WriteJSON(w, response.Err(401, "Invalid Authorization format"))
return
}
token := parts[1]
share, err := h.repo.GetPeerShareByToken(token)
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
if share == nil {
response.WriteJSON(w, response.Err(401, "Invalid token"))
return
}
if share.IsActive == 0 {
response.WriteJSON(w, response.Err(403, "Share is disabled"))
return
}
if share.ExpiryTime > 0 && share.ExpiryTime < time.Now().UnixMilli() {
response.WriteJSON(w, response.Err(403, "Share expired"))
return
}
next(w, r)
}
}
func (h *Handler) federationConnect(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
response.WriteJSON(w, response.ErrDefault("Invalid method"))
return
}
token := extractBearerToken(r)
share, err := h.repo.GetPeerShareByToken(token)
if err != nil || share == nil {
response.WriteJSON(w, response.Err(401, "Unauthorized"))
return
}
var nodeName string
var serverIP string
var status int
err = h.repo.DB().QueryRow("SELECT name, server_ip, status FROM node WHERE id = ?", share.NodeID).Scan(&nodeName, &serverIP, &status)
if err != nil {
response.WriteJSON(w, response.Err(-2, "Node not found"))
return
}
response.WriteJSON(w, response.OK(map[string]interface{}{
"shareId": share.ID,
"shareName": share.Name,
"nodeId": share.NodeID,
"nodeName": nodeName,
"serverIp": serverIP,
"status": status,
"maxBandwidth": share.MaxBandwidth,
"expiryTime": share.ExpiryTime,
"portRangeStart": share.PortRangeStart,
"portRangeEnd": share.PortRangeEnd,
}))
}
func (h *Handler) federationTunnelCreate(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
response.WriteJSON(w, response.ErrDefault("Invalid method"))
return
}
token := extractBearerToken(r)
share, err := h.repo.GetPeerShareByToken(token)
if err != nil || share == nil {
response.WriteJSON(w, response.Err(401, "Unauthorized"))
return
}
var req federationTunnelRequest
if err := decodeJSON(r.Body, &req); err != nil {
response.WriteJSON(w, response.ErrDefault("Invalid JSON"))
return
}
if req.RemotePort < share.PortRangeStart || req.RemotePort > share.PortRangeEnd {
response.WriteJSON(w, response.Err(403, "Port out of range"))
return
}
tunnelType := 1
if strings.ToLower(req.Protocol) == "udp" {
tunnelType = 2
}
tx, err := h.repo.DB().Begin()
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
defer tx.Rollback()
now := time.Now().UnixMilli()
res, err := tx.Exec(`INSERT INTO tunnel (name, type, protocol, flow, created_time, updated_time, status, in_ip) VALUES (?, ?, ?, 0, ?, ?, 1, ?)`,
fmt.Sprintf("Share-%d-Port-%d", share.ID, req.RemotePort),
tunnelType,
req.Protocol,
now,
now,
"",
)
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
tunnelID, _ := res.LastInsertId()
_, err = tx.Exec(`INSERT INTO chain_tunnel (tunnel_id, chain_type, node_id, port, strategy, inx, protocol) VALUES (?, 1, ?, ?, 'fifo', 0, ?)`,
tunnelID,
share.NodeID,
req.RemotePort,
req.Protocol,
)
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
if err := tx.Commit(); err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
h.wsServer.SendCommand(share.NodeID, "reload", nil, time.Second*5)
response.WriteJSON(w, response.OK(map[string]interface{}{
"tunnelId": tunnelID,
}))
}
func extractBearerToken(r *http.Request) string {
authHeader := r.Header.Get("Authorization")
parts := strings.Split(authHeader, " ")
if len(parts) == 2 && parts[0] == "Bearer" {
return parts[1]
}
return ""
}
go-backend/internal/http/handler/handler.go
+3
View File
@@ -143,6 +143,9 @@ func (h *Handler) Register(mux *http.ServeMux) {
mux.HandleFunc("/api/v1/group/permission/assign", h.groupPermissionAssign)
mux.HandleFunc("/api/v1/group/permission/remove", h.groupPermissionRemove)
mux.HandleFunc("/api/v1/open_api/sub_store", h.openAPISubStore)
mux.HandleFunc("/api/v1/federation/connect", h.authPeer(h.federationConnect))
mux.HandleFunc("/api/v1/federation/tunnel/create", h.authPeer(h.federationTunnelCreate))
mux.HandleFunc("/api/v1/federation/node/import", h.nodeImport)
mux.HandleFunc("/flow/test", h.flowTest)
mux.HandleFunc("/flow/config", h.flowConfig)
go-backend/internal/http/handler/mutations.go
+46 -2
View File
@@ -15,6 +15,7 @@ import (
"strings"
"time"
"go-backend/internal/http/client"
"go-backend/internal/http/response"
"go-backend/internal/security"
)
@@ -273,8 +274,8 @@ func (h *Handler) nodeCreate(w http.ResponseWriter, r *http.Request) {
now := time.Now().UnixMilli()
inx := nextIndex(db, "node")
_, err := db.Exec(`
INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx, is_remote, remote_url, remote_token, remote_config)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`,
name,
randomToken(16),
@@ -293,6 +294,10 @@ func (h *Handler) nodeCreate(w http.ResponseWriter, r *http.Request) {
defaultString(asString(req["tcpListenAddr"]), "[::]"),
defaultString(asString(req["udpListenAddr"]), "[::]"),
inx,
asInt(req["isRemote"], 0),
nullableText(asString(req["remoteUrl"])),
nullableText(asString(req["remoteToken"])),
nullableText(asString(req["remoteConfig"])),
)
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
@@ -509,6 +514,45 @@ func (h *Handler) tunnelCreate(w http.ResponseWriter, r *http.Request) {
inIP = buildTunnelInIP(runtimeState.InNodes, runtimeState.Nodes)
}
if len(runtimeState.InNodes) > 0 {
firstNodeID := runtimeState.InNodes[0].NodeID
var isRemote int
var rUrl, rToken sql.NullString
if err := h.repo.DB().QueryRow("SELECT is_remote, remote_url, remote_token FROM node WHERE id = ?", firstNodeID).Scan(&isRemote, &rUrl, &rToken); err == nil && isRemote == 1 {
fc := client.NewFederationClient()
targetProto := "tcp"
targetPort := 0
targetAddr := ""
if typeVal == 1 {
if len(runtimeState.OutNodes) > 0 {
outNode := runtimeState.OutNodes[0]
outNodeRec := runtimeState.Nodes[outNode.NodeID]
targetAddr = processServerAddress(outNodeRec.ServerIP)
if outNode.Port > 0 {
targetAddr = fmt.Sprintf("%s:%d", targetAddr, outNode.Port)
}
}
if len(runtimeState.InNodes) > 0 {
inNodesRaw := asMapSlice(req["inNodeId"])
if len(inNodesRaw) > 0 {
targetPort = asInt(inNodesRaw[0]["port"], 0)
targetProto = defaultString(asString(inNodesRaw[0]["protocol"]), "tcp")
}
}
if targetPort > 0 && targetAddr != "" {
_, err := fc.CreateTunnel(rUrl.String, rToken.String, targetProto, targetPort, targetAddr)
if err != nil {
response.WriteJSON(w, response.ErrDefault("Remote tunnel creation failed: "+err.Error()))
return
}
}
}
}
}
res, err := tx.Exec(`INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
name, trafficRatio, typeVal, "tls", flow, now, now, status, nullableText(inIP), inx)
if err != nil {