From df63b0113a3bfc3ac3a2d95ff8f2fd631a2bccff Mon Sep 17 00:00:00 2001 From: ryan Date: Mon, 1 Jun 2026 16:41:46 +0800 Subject: [PATCH] =?UTF-8?q?[=E4=BC=98=E5=8C=96]=20=E6=B7=BB=E5=8A=A0=20Ope?= =?UTF-8?q?nFlared=20API=20=E6=94=AF=E6=8C=81=EF=BC=8C=E5=A2=9E=E5=BC=BA?= =?UTF-8?q?=E5=BF=83=E8=B7=B3=E5=92=8C=E9=85=8D=E7=BD=AE=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/reference/api.md | 34 ++++ openflare_server/controller/flared.go | 173 ++++++++++++++++++ openflare_server/middleware/tunnel-auth.go | 36 ++++ openflare_server/router/api-router.go | 8 + openflare_server/router/api_flared_test.go | 192 ++++++++++++++++++++ openflare_server/service/agent.go | 3 + openflare_server/service/config_version.go | 7 +- openflare_server/service/flared_ws.go | 54 ++++++ openflare_server/service/node.go | 4 + openflare_server/service/relay.go | 110 +++++++++++ openflare_server/service/relay_test.go | 202 +++++++++++++++++++++ 11 files changed, 821 insertions(+), 2 deletions(-) create mode 100644 openflare_server/controller/flared.go create mode 100644 openflare_server/middleware/tunnel-auth.go create mode 100644 openflare_server/router/api_flared_test.go create mode 100644 openflare_server/service/flared_ws.go diff --git a/docs/reference/api.md b/docs/reference/api.md index 0c1d7a11..577d15d2 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -22,6 +22,8 @@ OpenFlare 的管理端 API 与 Agent API 都使用 JSON。 | --- | --- | | 管理端 API | 由管理端 Session 鉴权 | | Agent API | 固定放在 `/api/agent/*` | +| Relay API | 固定放在 `/api/relay/*`，使用 `X-Agent-Token` 鉴权（与 Agent 复用同一 token） | +| OpenFlared API | 固定放在 `/api/flared/*`，使用 `X-Tunnel-Token` 鉴权（独立的 tunnel_token） | | 只读接口 | 使用 `GET` | | 变更类接口 | 使用 `POST` | @@ -106,6 +108,38 @@ Agent 也可以在应用新版本后主动请求差异同步： 当 Server 侧 IP 组更新时，已连接的 Agent WebSocket 会收到 `type = "waf_ip_groups"` 的消息，payload 为发生变化的 IP 组数组。Agent 应只更新收到的组，不要求 Server 每次下发全部 IP 组。 +## OpenFlared API + +OpenFlared 客户端用于内网穿透场景，通过 `tunnel_token` 与 Server 通信，独立于 Agent 认证体系。所有接口都使用 `X-Tunnel-Token` 鉴权，Server 会校验节点 `node_type = tunnel_client`，否则返回 `403`。 + +| 方法 | 路径 | 说明 | +| --- | --- | --- | +| `POST` | `/api/flared/heartbeat` | 客户端心跳，刷新在线状态并返回 tunnel 配置版本摘要 | +| `GET` | `/api/flared/config/active` | 拉取完整的 tunnel 路由配置（relay 列表 + frpc 代理定义） | +| `POST` | `/api/flared/apply-log` | 上报配置应用结果（success / warning / failed） | +| `GET` | `/api/flared/ws` | 升级为 WebSocket，用于实时接收 `active_config` 推送 | + +心跳请求示例： + +```http +POST /api/flared/heartbeat +X-Tunnel-Token: +Content-Type: application/json + +{ + "client_version": "v0.2.0", + "frp_version": "0.61.0", + "tunnel_status": "running", + "connected_relays": [ + { "relay_node_id": "node-relay-1", "status": "healthy", "proxy_count": 3 } + ], + "current_version": "v1", + "current_checksum": "sha256..." +} +``` + +心跳响应包含 `active_config` 摘要与 `tunnel_settings`（包含心跳间隔、WebSocket 升级开关等运行时参数）。当 Server 发布新版本时，已连接的 OpenFlared WebSocket 会收到 `type = "active_config"` 消息，payload 为版本摘要，客户端应立即拉取完整配置并应用。 + 日志中不得打印完整 Token。 ## Swagger diff --git a/openflare_server/controller/flared.go b/openflare_server/controller/flared.go new file mode 100644 index 00000000..c48f7dbd --- /dev/null +++ b/openflare_server/controller/flared.go @@ -0,0 +1,173 @@ +package controller + +import ( + "log/slog" + "net" + "openflare/common" + "openflare/model" + "openflare/service" + "time" + + "github.com/gin-gonic/gin" + "golang.org/x/net/websocket" +) + +// FlaredHeartbeat godoc +// @Summary Report OpenFlared heartbeat +// @Tags Flared +// @Accept json +// @Produce json +// @Security TunnelTokenAuth +// @Param payload body service.FlaredHeartbeatPayload true "Flared heartbeat payload" +// @Success 200 {object} map[string]interface{} +// @Failure 400 {object} map[string]interface{} +// @Router /api/flared/heartbeat [post] +func FlaredHeartbeat(c *gin.Context) { + var payload service.FlaredHeartbeatPayload + if !bindJSON(c, &payload) { + return + } + authNode, ok := c.Get("flared_node") + if !ok { + respondUnauthorized(c, "无权进行此操作，Tunnel Token 无效") + return + } + node := authNode.(*model.Node) + response, err := service.HeartbeatFlared(node, payload) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, response) +} + +// FlaredGetActiveConfig godoc +// @Summary Get active tunnel config for OpenFlared +// @Tags Flared +// @Produce json +// @Security TunnelTokenAuth +// @Success 200 {object} map[string]interface{} +// @Router /api/flared/config/active [get] +func FlaredGetActiveConfig(c *gin.Context) { + authNode, ok := c.Get("flared_node") + if !ok { + respondUnauthorized(c, "无权进行此操作，Tunnel Token 无效") + return + } + node := authNode.(*model.Node) + config, err := service.GetFlaredTunnelConfig(node) + if err != nil { + respondFailure(c, "无法生成隧道配置: "+err.Error()) + return + } + respondSuccess(c, config) +} + +// FlaredReportApplyLog godoc +// @Summary Report OpenFlared apply result +// @Tags Flared +// @Accept json +// @Produce json +// @Security TunnelTokenAuth +// @Param payload body service.ApplyLogPayload true "Apply log payload" +// @Success 200 {object} map[string]interface{} +// @Router /api/flared/apply-log [post] +func FlaredReportApplyLog(c *gin.Context) { + var payload service.ApplyLogPayload + if !bindJSON(c, &payload) { + return + } + if authNode, ok := c.Get("flared_node"); ok { + payload.NodeID = authNode.(*model.Node).NodeID + } + log, err := service.ReportApplyLog(payload) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, log) +} + +// FlaredWebSocket godoc +// @Summary Upgrade OpenFlared connection to websocket +// @Tags Flared +// @Security TunnelTokenAuth +// @Router /api/flared/ws [get] +func FlaredWebSocket(c *gin.Context) { + authNode, ok := c.Get("flared_node") + if !ok { + respondUnauthorized(c, "无权进行此操作，Tunnel Token 无效") + return + } + node := authNode.(*model.Node) + slog.Debug("flared ws upgrade requested", "node_id", node.NodeID, "remote", c.Request.RemoteAddr) + websocket.Handler(func(conn *websocket.Conn) { + client := service.RegisterFlaredWSClient(node.NodeID) + defer service.UnregisterFlaredWSClient(client) + defer func() { + _ = conn.Close() + slog.Debug("flared ws connection closed", "node_id", node.NodeID) + }() + + slog.Debug("flared ws upgrade succeeded", "node_id", node.NodeID, "remote", c.Request.RemoteAddr) + + go func() { + <-client.Done() + _ = conn.Close() + }() + + go streamFlaredWSMessages(c, conn, client) + + for { + var message service.WSMessage + _ = conn.SetReadDeadline(time.Now().Add(flaredWSReadTimeout())) + if err := websocket.JSON.Receive(conn, &message); err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + slog.Debug("flared ws receive timeout", "node_id", node.NodeID) + return + } + slog.Debug("flared ws receive failed", "node_id", node.NodeID, "error", err) + return + } + slog.Debug("flared ws message received", "node_id", node.NodeID, "type", message.Type) + switch message.Type { + case "ping": + if !service.SendFlaredWSPong(node.NodeID) { + slog.Debug("flared ws pong enqueue failed", "node_id", node.NodeID) + } + case "pong": + slog.Debug("flared ws pong received", "node_id", node.NodeID) + default: + slog.Debug("flared ws unsupported message type", "node_id", node.NodeID, "type", message.Type) + } + } + }).ServeHTTP(c.Writer, c.Request) +} + +func streamFlaredWSMessages(c *gin.Context, conn *websocket.Conn, client *service.WSClient) { + for { + select { + case <-c.Request.Context().Done(): + return + case <-client.Done(): + return + case message, ok := <-client.Messages(): + if !ok { + return + } + _ = conn.SetWriteDeadline(time.Now().Add(agentWSWriteTimeout())) + if err := websocket.JSON.Send(conn, message); err != nil { + slog.Debug("flared ws send failed", "node_id", client.ID(), "error", err) + return + } + } + } +} + +func flaredWSReadTimeout() time.Duration { + timeout := time.Duration(common.AgentHeartbeatInterval) * time.Millisecond * 3 + if timeout < 30*time.Second { + return 30 * time.Second + } + return timeout +} diff --git a/openflare_server/middleware/tunnel-auth.go b/openflare_server/middleware/tunnel-auth.go new file mode 100644 index 00000000..af169438 --- /dev/null +++ b/openflare_server/middleware/tunnel-auth.go @@ -0,0 +1,36 @@ +package middleware + +import ( + "net/http" + "openflare/service" + + "github.com/gin-gonic/gin" +) + +// TunnelAuth authenticates OpenFlared client requests using the per-node +// tunnel_token carried in the X-Tunnel-Token header, and verifies the node is +// of the tunnel_client type. +func TunnelAuth() func(c *gin.Context) { + return func(c *gin.Context) { + token := c.GetHeader("X-Tunnel-Token") + node, err := service.AuthenticateAccessToken(token) + if err != nil { + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": "无权进行此操作，Tunnel Token 无效", + }) + c.Abort() + return + } + if node.NodeType != "tunnel_client" { + c.JSON(http.StatusForbidden, gin.H{ + "success": false, + "message": "此节点不是 TunnelClient 类型", + }) + c.Abort() + return + } + c.Set("flared_node", node) + c.Next() + } +} diff --git a/openflare_server/router/api-router.go b/openflare_server/router/api-router.go index 7eb04ca1..fc8ad27e 100644 --- a/openflare_server/router/api-router.go +++ b/openflare_server/router/api-router.go @@ -232,6 +232,14 @@ func SetApiRouter(router *gin.Engine) { relayRoute.POST("/heartbeat", controller.RelayHeartbeat) relayRoute.GET("/ws", controller.RelayWebSocket) } + flaredRoute := apiRouter.Group("/flared") + flaredRoute.Use(middleware.TunnelAuth()) + { + flaredRoute.POST("/heartbeat", controller.FlaredHeartbeat) + flaredRoute.GET("/config/active", controller.FlaredGetActiveConfig) + flaredRoute.POST("/apply-log", controller.FlaredReportApplyLog) + flaredRoute.GET("/ws", controller.FlaredWebSocket) + } } } diff --git a/openflare_server/router/api_flared_test.go b/openflare_server/router/api_flared_test.go new file mode 100644 index 00000000..3694a58a --- /dev/null +++ b/openflare_server/router/api_flared_test.go @@ -0,0 +1,192 @@ +package router_test + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "openflare/common" + "openflare/model" + "openflare/router" + "openflare/service" + "testing" + + "github.com/gin-contrib/sessions" + "github.com/gin-contrib/sessions/cookie" + "github.com/gin-gonic/gin" +) + +func TestPhaseFlaredRoutesUnauthorized(t *testing.T) { + gin.SetMode(gin.TestMode) + common.RedisEnabled = false + setupTestDB(t) + + engine := gin.New() + engine.Use(sessions.Sessions("session", cookie.NewStore([]byte("test-secret")))) + router.SetApiRouter(engine) + + heartbeatReq := httptest.NewRequest(http.MethodPost, "/api/flared/heartbeat", bytes.NewReader([]byte(`{}`))) + heartbeatReq.Header.Set("Content-Type", "application/json") + heartbeatRec := httptest.NewRecorder() + engine.ServeHTTP(heartbeatRec, heartbeatReq) + if heartbeatRec.Code != http.StatusUnauthorized { + t.Fatalf("expected unauthorized status for missing token, got %d body=%s", heartbeatRec.Code, heartbeatRec.Body.String()) + } + + activeReq := httptest.NewRequest(http.MethodGet, "/api/flared/config/active", nil) + activeRec := httptest.NewRecorder() + engine.ServeHTTP(activeRec, activeReq) + if activeRec.Code != http.StatusUnauthorized { + t.Fatalf("expected unauthorized status for missing token on active config, got %d", activeRec.Code) + } + + applyReq := httptest.NewRequest(http.MethodPost, "/api/flared/apply-log", bytes.NewReader([]byte(`{}`))) + applyReq.Header.Set("Content-Type", "application/json") + applyRec := httptest.NewRecorder() + engine.ServeHTTP(applyRec, applyReq) + if applyRec.Code != http.StatusUnauthorized { + t.Fatalf("expected unauthorized status for missing token on apply log, got %d", applyRec.Code) + } +} + +func TestPhaseFlaredRoutesRejectWrongNodeType(t *testing.T) { + gin.SetMode(gin.TestMode) + common.RedisEnabled = false + setupTestDB(t) + + engine := gin.New() + engine.Use(sessions.Sessions("session", cookie.NewStore([]byte("test-secret")))) + router.SetApiRouter(engine) + + adminToken := prepareRootToken(t) + createNodeResp := performJSONRequest(t, engine, adminToken, http.MethodPost, "/api/nodes/", map[string]any{ + "name": "edge-for-flared-test", + "ip": "10.0.0.20", + }) + var createdNode service.NodeView + decodeResponseData(t, createNodeResp, &createdNode) + + heartbeatReq := httptest.NewRequest(http.MethodPost, "/api/flared/heartbeat", bytes.NewReader([]byte(`{}`))) + heartbeatReq.Header.Set("Content-Type", "application/json") + heartbeatReq.Header.Set("X-Tunnel-Token", createdNode.AccessToken) + heartbeatRec := httptest.NewRecorder() + engine.ServeHTTP(heartbeatRec, heartbeatReq) + if heartbeatRec.Code != http.StatusForbidden { + t.Fatalf("expected forbidden status for edge_node token, got %d body=%s", heartbeatRec.Code, heartbeatRec.Body.String()) + } +} + +func TestPhaseFlaredLifecycle(t *testing.T) { + gin.SetMode(gin.TestMode) + common.RedisEnabled = false + setupTestDB(t) + + engine := gin.New() + engine.Use(sessions.Sessions("session", cookie.NewStore([]byte("test-secret")))) + router.SetApiRouter(engine) + + adminToken := prepareRootToken(t) + + // Create an enabled proxy route that will be served to the flared client + // through the tunnel upstream flow. + createRouteAndPublishVersion(t, engine, adminToken) + + // Seed a tunnel_client node directly so we can use its access token as the + // tunnel_token when calling the flared endpoints. + tunnelNode := &model.Node{ + NodeID: "tun-flared-1", + Name: "office-flared-1", + IP: "192.168.10.20", + AccessToken: "tunnel-token-phase", + Status: service.NodeStatusPending, + NodeType: "tunnel_client", + Version: "", + } + if err := tunnelNode.Insert(); err != nil { + t.Fatalf("failed to seed tunnel client node: %v", err) + } + + heartbeatResp := performFlaredJSONRequest(t, engine, tunnelNode.AccessToken, http.MethodPost, "/api/flared/heartbeat", map[string]any{ + "client_version": "v0.2.0", + "frp_version": "0.61.0", + "tunnel_status": "running", + "current_version": "", + }) + if !heartbeatResp.Success { + t.Fatalf("flared heartbeat failed: %s", heartbeatResp.Message) + } + var heartbeatData service.FlaredHeartbeatResponse + if err := json.Unmarshal(heartbeatResp.Data, &heartbeatData); err != nil { + t.Fatalf("failed to decode flared heartbeat response: %v", err) + } + if heartbeatData.ActiveConfig == nil { + t.Fatal("expected heartbeat to return active config summary") + } + if heartbeatData.TunnelSettings == nil { + t.Fatal("expected heartbeat to return tunnel_settings") + } + + // Re-fetch node and assert status flipped to online. + updated, err := model.GetNodeByNodeID(tunnelNode.NodeID) + if err != nil { + t.Fatalf("failed to reload flared node: %v", err) + } + if updated.Status != service.NodeStatusOnline { + t.Fatalf("expected flared node status to be online, got %q", updated.Status) + } + if updated.Version != "v0.2.0" { + t.Fatalf("expected flared client_version to be stored, got %q", updated.Version) + } + + activeResp := performFlaredJSONRequest(t, engine, tunnelNode.AccessToken, http.MethodGet, "/api/flared/config/active", nil) + if !activeResp.Success { + t.Fatalf("flared get active config failed: %s", activeResp.Message) + } + var activeConfig service.FlaredTunnelConfigResponse + if err := json.Unmarshal(activeResp.Data, &activeConfig); err != nil { + t.Fatalf("failed to decode flared active config: %v", err) + } + if activeConfig.Version == "" || activeConfig.Checksum == "" { + t.Fatalf("expected flared active config to return version summary, got %+v", activeConfig) + } + + applyResp := performFlaredJSONRequest(t, engine, tunnelNode.AccessToken, http.MethodPost, "/api/flared/apply-log", map[string]any{ + "version": activeConfig.Version, + "result": service.ApplyResultOK, + "message": "apply ok", + "checksum": activeConfig.Checksum, + }) + if !applyResp.Success { + t.Fatalf("flared apply log failed: %s", applyResp.Message) + } +} + +func performFlaredJSONRequest(t *testing.T, engine http.Handler, token string, method string, path string, body any) apiResponse { + t.Helper() + var payload []byte + if body != nil { + var err error + payload, err = json.Marshal(body) + if err != nil { + t.Fatalf("failed to marshal request body: %v", err) + } + } + req := httptest.NewRequest(method, path, bytes.NewReader(payload)) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + req.Header.Set("X-Tunnel-Token", token) + recorder := httptest.NewRecorder() + engine.ServeHTTP(recorder, req) + if recorder.Code != http.StatusOK { + t.Fatalf("unexpected status %d for %s %s: %s", recorder.Code, method, path, recorder.Body.String()) + } + var resp apiResponse + if err := json.Unmarshal(recorder.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to unmarshal response: %v", err) + } + if !resp.Success { + t.Fatalf("request %s %s failed: %s", method, path, resp.Message) + } + return resp +} diff --git a/openflare_server/service/agent.go b/openflare_server/service/agent.go index a3827cfb..37b0716e 100644 --- a/openflare_server/service/agent.go +++ b/openflare_server/service/agent.go @@ -468,6 +468,9 @@ func computeNodeStatus(node *model.Node) string { if node.NodeType == "tunnel_relay" && IsRelayWSConnected(node.NodeID) { return NodeStatusOnline } + if node.NodeType == "tunnel_client" && IsFlaredWSConnected(node.NodeID) { + return NodeStatusOnline + } if IsAgentWSConnected(node.NodeID) { return NodeStatusOnline } diff --git a/openflare_server/service/config_version.go b/openflare_server/service/config_version.go index 2b8707ed..022b4589 100644 --- a/openflare_server/service/config_version.go +++ b/openflare_server/service/config_version.go @@ -356,6 +356,7 @@ func PublishConfigVersion(createdBy string, force bool) (*ReleaseResult, error) Checksum: record.Checksum, } BroadcastAgentWSActiveConfig(activeConfig) + BroadcastFlaredWSActiveConfig(activeConfig) return &ReleaseResult{ Version: record, Routes: bundle.Routes, @@ -403,10 +404,12 @@ func ActivateConfigVersion(id uint) (*model.ConfigVersion, error) { return nil, err } version.IsActive = true - BroadcastAgentWSActiveConfig(&ActiveConfigMeta{ + activeConfig := &ActiveConfigMeta{ Version: version.Version, Checksum: version.Checksum, - }) + } + BroadcastAgentWSActiveConfig(activeConfig) + BroadcastFlaredWSActiveConfig(activeConfig) return version, nil } diff --git a/openflare_server/service/flared_ws.go b/openflare_server/service/flared_ws.go new file mode 100644 index 00000000..3a8aef8f --- /dev/null +++ b/openflare_server/service/flared_ws.go @@ -0,0 +1,54 @@ +package service + +const ( + FlaredWSConnectedLastSeenValue = "__OPENFLARE_FLARED_WS_CONNECTED__" + + FlaredWSMessageTypeActiveConfig = "active_config" + FlaredWSMessageTypeForceSync = "force_sync" + FlaredWSMessageTypePong = "pong" +) + +var DefaultFlaredWSHub = NewWSHub("flared") + +func RegisterFlaredWSClient(nodeID string) *WSClient { + return DefaultFlaredWSHub.Register(nodeID) +} + +func UnregisterFlaredWSClient(client *WSClient) { + DefaultFlaredWSHub.Unregister(client) +} + +func DisconnectFlaredWSClient(nodeID string) { + DefaultFlaredWSHub.Disconnect(nodeID) +} + +func IsFlaredWSConnected(nodeID string) bool { + return DefaultFlaredWSHub.IsConnected(nodeID) +} + +func SendFlaredWSPong(nodeID string) bool { + return DefaultFlaredWSHub.SendMessage(nodeID, WSMessage{ + Type: FlaredWSMessageTypePong, + }) +} + +func SendFlaredWSActiveConfig(nodeID string, activeConfig *ActiveConfigMeta) bool { + if activeConfig == nil { + return false + } + return DefaultFlaredWSHub.SendMessage(nodeID, WSMessage{ + Type: FlaredWSMessageTypeActiveConfig, + Payload: activeConfig, + }) +} + +func BroadcastFlaredWSActiveConfig(activeConfig *ActiveConfigMeta) WSBroadcastResult { + if activeConfig == nil { + return WSBroadcastResult{} + } + result := DefaultFlaredWSHub.Broadcast(WSMessage{ + Type: FlaredWSMessageTypeActiveConfig, + Payload: activeConfig, + }) + return result +} diff --git a/openflare_server/service/node.go b/openflare_server/service/node.go index d00d1c67..6d14299a 100644 --- a/openflare_server/service/node.go +++ b/openflare_server/service/node.go @@ -166,6 +166,7 @@ func DeleteNode(id uint) error { } invalidateAccessTokenCache(node.AccessToken) DisconnectAgentWSClient(node.NodeID) + DisconnectFlaredWSClient(node.NodeID) return nil } @@ -365,6 +366,9 @@ func nodeViewLastSeenAt(node *model.Node) any { if node.NodeType == "tunnel_relay" && IsRelayWSConnected(node.NodeID) { return RelayWSConnectedLastSeenValue } + if node.NodeType == "tunnel_client" && IsFlaredWSConnected(node.NodeID) { + return FlaredWSConnectedLastSeenValue + } if IsAgentWSConnected(node.NodeID) { return AgentWSConnectedLastSeenValue } diff --git a/openflare_server/service/relay.go b/openflare_server/service/relay.go index 18f343ed..7e3b9a6a 100644 --- a/openflare_server/service/relay.go +++ b/openflare_server/service/relay.go @@ -1,6 +1,7 @@ package service import ( + "errors" "fmt" "log/slog" "openflare/common" @@ -194,6 +195,115 @@ type FlaredHeartbeatPayload struct { CurrentChecksum string `json:"current_checksum"` } +func normalizeFlaredHeartbeatPayload(payload FlaredHeartbeatPayload) FlaredHeartbeatPayload { + payload.ClientVersion = strings.TrimSpace(payload.ClientVersion) + payload.FrpVersion = strings.TrimSpace(payload.FrpVersion) + payload.TunnelStatus = strings.ToLower(strings.TrimSpace(payload.TunnelStatus)) + payload.CurrentVersion = strings.TrimSpace(payload.CurrentVersion) + payload.CurrentChecksum = strings.TrimSpace(payload.CurrentChecksum) + cleaned := make([]FlaredConnectedRelay, 0, len(payload.ConnectedRelays)) + for _, relay := range payload.ConnectedRelays { + relay.RelayNodeID = strings.TrimSpace(relay.RelayNodeID) + relay.Status = strings.ToLower(strings.TrimSpace(relay.Status)) + if relay.RelayNodeID == "" { + continue + } + if relay.Status == "" { + relay.Status = "unknown" + } + cleaned = append(cleaned, relay) + } + payload.ConnectedRelays = cleaned + return payload +} + +// HeartbeatFlared processes an OpenFlared heartbeat, refreshes node status, +// persists the connected relay snapshot, and returns the active tunnel +// config summary plus runtime settings. +func HeartbeatFlared(node *model.Node, payload FlaredHeartbeatPayload) (*FlaredHeartbeatResponse, error) { + if node == nil { + return nil, fmt.Errorf("tunnel client node is nil") + } + if node.NodeType != "tunnel_client" { + return nil, fmt.Errorf("node %s is not a tunnel_client", node.NodeID) + } + slog.Debug("flared heartbeat received", "node_id", node.NodeID, "client_version", payload.ClientVersion) + payload = normalizeFlaredHeartbeatPayload(payload) + + now := time.Now() + previous := *node + + changes := make(map[string]any) + if previous.Version != payload.ClientVersion { + changes["version"] = payload.ClientVersion + } + if previous.ExtVersion != payload.FrpVersion { + changes["ext_version"] = payload.FrpVersion + } + if previous.CurrentVersion != payload.CurrentVersion { + changes["current_version"] = payload.CurrentVersion + } + if !previous.LastSeenAt.Equal(now) { + changes["last_seen_at"] = now + } + changes["status"] = NodeStatusOnline + + node.Version = payload.ClientVersion + node.ExtVersion = payload.FrpVersion + node.CurrentVersion = payload.CurrentVersion + node.LastSeenAt = now + node.Status = NodeStatusOnline + if !node.GeoManualOverride { + applyGeoInfoFromIP(node, node.IP) + } + + if len(changes) > 0 { + if err := model.DB.Model(node).Updates(changes).Error; err != nil { + return nil, fmt.Errorf("update flared heartbeat: %w", err) + } + } + refreshAccessTokenCache(node) + persistFlaredObservability(node.NodeID, payload, now) + + activeConfig, err := GetActiveConfigMetaForAgent() + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, err + } + return &FlaredHeartbeatResponse{ + ActiveConfig: activeConfig, + TunnelSettings: buildRelaySettings(), + }, nil +} + +// persistFlaredObservability records the latest connection snapshot and +// health event for the OpenFlared client. +func persistFlaredObservability(nodeID string, payload FlaredHeartbeatPayload, reportedAt time.Time) { + connected := make([]string, 0, len(payload.ConnectedRelays)) + for _, relay := range payload.ConnectedRelays { + connected = append(connected, fmt.Sprintf("%s:%s", relay.RelayNodeID, relay.Status)) + } + managedTypes := map[string]struct{}{ + "flared_runtime_unhealthy": {}, + } + var events []AgentNodeHealthEvent + if payload.TunnelStatus == "unhealthy" { + events = append(events, AgentNodeHealthEvent{ + EventType: "flared_runtime_unhealthy", + Severity: NodeHealthSeverityCritical, + Message: "openflared runtime is not healthy", + TriggeredAtUnix: reportedAt.Unix(), + Metadata: map[string]string{ + "tunnel_status": payload.TunnelStatus, + "client_version": payload.ClientVersion, + "current_version": payload.CurrentVersion, + "current_checksum": payload.CurrentChecksum, + "connected_relays": strings.Join(connected, ","), + }, + }) + } + _ = reconcileScopedNodeHealthEvents(model.DB, nodeID, events, reportedAt, managedTypes) +} + // FlaredConnectedRelay describes the status of a relay connection from a client. type FlaredConnectedRelay struct { RelayNodeID string `json:"relay_node_id"` diff --git a/openflare_server/service/relay_test.go b/openflare_server/service/relay_test.go index af2144be..8fe7ed20 100644 --- a/openflare_server/service/relay_test.go +++ b/openflare_server/service/relay_test.go @@ -1,9 +1,12 @@ package service import ( + "errors" "openflare/model" "testing" "time" + + "gorm.io/gorm" ) func TestHeartbeatRelayPersistsRuntimeAndObservability(t *testing.T) { @@ -93,3 +96,202 @@ func TestHeartbeatRelayPersistsRuntimeAndObservability(t *testing.T) { t.Fatalf("unexpected relay dashboard: %+v", observability.RelayDashboard) } } + +func TestHeartbeatFlaredRejectsWrongNodeType(t *testing.T) { + setupServiceTestDB(t) + + node := &model.Node{ + NodeID: "node-not-tunnel-client", + Name: "edge", + IP: "10.0.0.1", + AccessToken: "edge-token", + Status: NodeStatusPending, + NodeType: "edge_node", + Version: "v0.0.0", + } + if err := node.Insert(); err != nil { + t.Fatalf("failed to seed edge node: %v", err) + } + + _, err := HeartbeatFlared(node, FlaredHeartbeatPayload{ + ClientVersion: "v0.1.0", + FrpVersion: "0.61.0", + TunnelStatus: "running", + CurrentVersion: "v1", + }) + if err == nil { + t.Fatal("expected error for non-tunnel_client node type") + } +} + +func TestHeartbeatFlaredRejectsNilNode(t *testing.T) { + if _, err := HeartbeatFlared(nil, FlaredHeartbeatPayload{}); err == nil { + t.Fatal("expected error when node is nil") + } +} + +func TestHeartbeatFlaredPersistsRuntime(t *testing.T) { + setupServiceTestDB(t) + + node := &model.Node{ + NodeID: "node-flared-1", + Name: "flared-1", + IP: "", + AccessToken: "tunnel-token-abc", + Status: NodeStatusPending, + NodeType: "tunnel_client", + Version: "", + } + if err := node.Insert(); err != nil { + t.Fatalf("failed to seed flared node: %v", err) + } + + resp, err := HeartbeatFlared(node, FlaredHeartbeatPayload{ + ClientVersion: " v0.2.0 ", + FrpVersion: " 0.61.1 ", + TunnelStatus: " RUNNING ", + ConnectedRelays: []FlaredConnectedRelay{ + {RelayNodeID: " node-relay-1 ", Status: " HEALTHY ", ProxyCount: 3}, + {RelayNodeID: "", Status: "running"}, + {RelayNodeID: "node-relay-2", Status: ""}, + }, + CurrentVersion: "v1", + CurrentChecksum: "checksum-1", + }) + if err != nil { + t.Fatalf("HeartbeatFlared failed: %v", err) + } + if resp == nil { + t.Fatal("expected non-nil response") + } + if resp.TunnelSettings == nil { + t.Fatal("expected tunnel_settings in response") + } + if resp.TunnelSettings.HeartbeatInterval == 0 { + t.Fatal("expected heartbeat interval to be set in tunnel_settings") + } + + updated, err := model.GetNodeByNodeID(node.NodeID) + if err != nil { + t.Fatalf("failed to reload flared node: %v", err) + } + if updated.Status != NodeStatusOnline { + t.Fatalf("expected flared node to be online, got %q", updated.Status) + } + if updated.Version != "v0.2.0" { + t.Fatalf("expected client_version to be trimmed and stored, got %q", updated.Version) + } + if updated.ExtVersion != "0.61.1" { + t.Fatalf("expected frp_version to be trimmed and stored, got %q", updated.ExtVersion) + } + if updated.CurrentVersion != "v1" { + t.Fatalf("expected current_version to be stored, got %q", updated.CurrentVersion) + } + if updated.LastSeenAt.IsZero() { + t.Fatal("expected last_seen_at to be updated") + } +} + +func TestHeartbeatFlaredTrimsAndFiltersRelays(t *testing.T) { + normalized := normalizeFlaredHeartbeatPayload(FlaredHeartbeatPayload{ + TunnelStatus: " UNHEALTHY ", + ConnectedRelays: []FlaredConnectedRelay{ + {RelayNodeID: " node-a ", Status: " OK "}, + {RelayNodeID: "", Status: "running"}, + }, + }) + if normalized.TunnelStatus != "unhealthy" { + t.Fatalf("expected tunnel_status to be lower-cased, got %q", normalized.TunnelStatus) + } + if len(normalized.ConnectedRelays) != 1 { + t.Fatalf("expected empty relay_node_id to be dropped, got %+v", normalized.ConnectedRelays) + } + relay := normalized.ConnectedRelays[0] + if relay.RelayNodeID != "node-a" { + t.Fatalf("expected relay_node_id to be trimmed, got %q", relay.RelayNodeID) + } + if relay.Status != "ok" { + t.Fatalf("expected status to be lower-cased, got %q", relay.Status) + } +} + +func TestHeartbeatFlaredEmitsHealthEventOnUnhealthy(t *testing.T) { + setupServiceTestDB(t) + + node := &model.Node{ + NodeID: "node-flared-unhealthy", + Name: "flared-unhealthy", + IP: "", + AccessToken: "tunnel-token-unhealthy", + Status: NodeStatusPending, + NodeType: "tunnel_client", + Version: "", + } + if err := node.Insert(); err != nil { + t.Fatalf("failed to seed flared node: %v", err) + } + + if _, err := HeartbeatFlared(node, FlaredHeartbeatPayload{ + ClientVersion: "v0.2.0", + FrpVersion: "0.61.0", + TunnelStatus: "unhealthy", + CurrentVersion: "v1", + CurrentChecksum: "checksum-1", + }); err != nil { + t.Fatalf("HeartbeatFlared failed: %v", err) + } + + events, err := model.ListNodeHealthEvents(node.NodeID, false, 20) + if err != nil { + t.Fatalf("ListNodeHealthEvents failed: %v", err) + } + if len(events) == 0 { + t.Fatal("expected unhealthy heartbeat to emit a node health event") + } + foundUnhealthy := false + for _, event := range events { + if event.EventType == "flared_runtime_unhealthy" { + foundUnhealthy = true + } + } + if !foundUnhealthy { + t.Fatalf("expected flared_runtime_unhealthy event in %+v", events) + } +} + +func TestHeartbeatFlaredEmitsEmptyConnectedRelays(t *testing.T) { + normalized := normalizeFlaredHeartbeatPayload(FlaredHeartbeatPayload{}) + if normalized.ConnectedRelays == nil { + t.Fatal("expected ConnectedRelays to be non-nil empty slice for nil input") + } + if len(normalized.ConnectedRelays) != 0 { + t.Fatalf("expected empty ConnectedRelays, got %+v", normalized.ConnectedRelays) + } +} + +func TestGetFlaredTunnelConfigRequiresActiveVersion(t *testing.T) { + setupServiceTestDB(t) + + node := &model.Node{ + NodeID: "node-flared-noactive", + Name: "flared-noactive", + IP: "", + AccessToken: "tunnel-token-na", + Status: NodeStatusPending, + NodeType: "tunnel_client", + Version: "", + } + if err := node.Insert(); err != nil { + t.Fatalf("failed to seed flared node: %v", err) + } + + _, err := GetFlaredTunnelConfig(node) + if err == nil { + t.Fatal("expected error when no active config version exists") + } + if !errors.Is(err, gorm.ErrRecordNotFound) { + // We accept either wrapping the underlying error or surfacing a friendly message. + // Just ensure we surface a clear failure instead of a nil result. + t.Logf("GetFlaredTunnelConfig returned wrapped error: %v", err) + } +}