diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 21182511..087a576b 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -2,7 +2,9 @@ 你会学到：OpenFlare 的整体架构、Server、Agent、OpenResty 与管理端前端的职责边界，以及一次配置发布从管理端到节点生效的请求流。 -OpenFlare 由 Server、Agent、节点本地 OpenResty 和管理端前端组成。Server 是控制面，Agent 是节点侧唯一受控落地入口，OpenResty 是实际数据面。 +OpenFlare 由 Server、Agent、节点本地 OpenResty 和管理端前端组成。Server 是控制面，Agent 是节点侧唯一受控落地入口，OpenResty 是实际数据面。内网穿透场景中，Relay（frps 管理器）和 OpenFlared（frpc 管理器）扩展了数据面流量路径。 + +### 标准反代流量路径 ```text Browser @@ -24,14 +26,38 @@ OpenResty binary Origin ``` +### 内网穿透流量路径 + +```text +Browser + | + | HTTPS request + v +OpenResty (Agent, TLS/WAF) <-- TunnelRelay 节点 + | + | proxy_pass http://localhost:vhost_port (Host header preserved) + v +OpenFlareRelay (frps) <-- TunnelRelay 节点，与 Agent 同机部署 + | + | frp tunnel protocol (HTTP Vhost routing by Host header) + v +OpenFlared (frpc) <-- 内网服务器 + | + | HTTP/HTTPS forward + v +Internal Service (192.168.x.x) +``` + ## 组件职责 -| 组件 | 职责 | -| --------- | ---------------------------------------------------------------------- | -| Server | 管理端 UI、管理 API、Agent API、配置渲染、版本发布、数据存储与聚合查询 | -| Agent | 注册、心跳、同步、写入文件、校验、reload、失败回滚、自更新与轻量采集 | -| OpenResty | 接收真实流量，按 OpenFlare 渲染的配置执行 WAF、PoW、认证与反向代理 | -| Frontend | 管理网站配置、WAF、源站、证书、节点、版本、用户、设置与观测页面 | +| 组件 | 职责 | +| --------------- | ---------------------------------------------------------------------- | +| Server | 管理端 UI、管理 API、Agent/Relay/Client API、配置渲染、版本发布、数据存储与聚合查询 | +| Agent | 注册、心跳、同步、写入文件、校验、reload、失败回滚、自更新与轻量采集 | +| OpenResty | 接收真实流量，按 OpenFlare 渲染的配置执行 WAF、PoW、认证与反向代理 | +| OpenFlareRelay | 管理 frps 进程生命周期，提供隧道中继服务，通过心跳接收 frps 配置 | +| OpenFlared | 管理 frpc 进程（可多个），连接 Relay 中继，将流量转发到内网服务 | +| Frontend | 管理网站配置、WAF、源站、证书、节点、Tunnel、版本、用户、设置与观测页面 | ## Server @@ -97,6 +123,27 @@ Agent 执行 OpenResty 校验与 reload Agent 上报应用结果 ``` +### Relay 同步流 + +```text +Relay HTTP heartbeat -> Server 返回 frps 配置 +Relay 生成 frps.toml 并启动/重启 frps +Relay 定期上报 frps 状态 +``` + +Relay 使用与 Agent 相同的 `agent_token` 认证（同一节点），通过 `/api/relay/*` 端点通信。frps 配置相对静态（端口、认证 Token），通过心跳下发，不纳入版本化发布流。 + +### OpenFlared 同步流 + +```text +Client HTTP heartbeat -> Server 返回 tunnel 配置版本摘要 +Client 发现新版本 -> 拉取 tunnel 路由配置（relay 列表 + proxy 定义） +Client 为每个 Relay 生成 frpc.toml 并启动/重载 frpc 进程 +Client 上报应用结果 +``` + +OpenFlared 使用独立的 `tunnel_token` 认证，通过 `/api/flared/*` 端点通信。Tunnel 路由配置随发布流程版本化同步，配置变更时优先使用 `frpc reload` 热重载。 + **WebSocket 升级流程**（可选，通过 `AgentWebsocketUpgradeEnabled` 选项控制）： 当启用 WebSocket 升级时： @@ -126,6 +173,7 @@ WAF 在 OpenResty `access_by_lua_file` 阶段执行。规则来自当前激活 * `origins` * `config_versions` * `nodes` +* `tunnels` * `auth_sources` * `external_accounts` * `node_system_profiles` @@ -152,6 +200,9 @@ WAF 在 OpenResty `access_by_lua_file` 阶段执行。规则来自当前激活 | 全局单激活版本 | 降低 MVP 复杂度，保证所有节点默认一致；支持版本预览、历史查询与一键回滚 | | 网站配置聚合多域名 | 支持一个业务站点共享站点级策略，同时允许按域名绑定证书 | | 观测数据服务端聚合 | 避免前端临时统计造成口径不一致 | +| 内网穿透基于 frp 整合 | 复用成熟隧道协议，避免自研隧道的稳定性风险；frps HTTP Vhost 路由天然适配 | +| Relay/Client 独立二进制 | 职责分离，Relay 管理 frps，Client 管理 frpc，各自独立升级和部署 | +| Tunnel 与 Node 体系分离 | Tunnel 客户端在内网运行，与公网节点概念不同，使用独立的注册和认证体系 | ## 贡献者阅读建议 diff --git a/docs/design/index.md b/docs/design/index.md index 8a361c97..972841c7 100644 --- a/docs/design/index.md +++ b/docs/design/index.md @@ -31,12 +31,15 @@ OpenFlare 当前不定位为通用日志平台、服务网格、Kubernetes Ingre | 节点管理 | 节点状态、令牌体系、部署与更新链路 | | 管理端前端 | 基于 Next.js 的正式管理端 | | 认证源登录 | 支持以认证源形式配置 GitHub 与标准 OIDC 登录入口，并允许第三方账号绑定已有本地用户 | +| 内网穿透 | 通过 TunnelRelay 节点与 OpenFlared 客户端，将内网 HTTP 服务安全暴露到公网，复用 Agent 的 HTTPS/WAF 能力 | 默认工作方式： * 所有节点消费同一份全局激活版本。 * Server 保存配置与状态，不直接 SSH 管理节点。 * Agent 是节点侧唯一受控落地入口。 +* TunnelRelay 节点同时运行 Agent（OpenResty）和 Relay（frps），提供内网穿透中继。 +* OpenFlared 客户端在内网运行，管理 frpc 进程连接 Relay，将流量转发到内网服务。 ## 典型使用场景 @@ -48,6 +51,7 @@ OpenFlare 当前不定位为通用日志平台、服务网格、Kubernetes Ingre | 快速回滚 | 重新激活旧版本，让 Agent 拉取并应用 | | 证书托管 | 为不同域名绑定 TLS 证书 | | 基础观测 | 查看节点状态、请求聚合、访问分析和健康事件 | +| 内网穿透 | 通过 Tunnel 将无法直接公网访问的内网 HTTP 服务暴露到互联网，享有 HTTPS、WAF 等全部防护能力 | ## 网站配置约束 @@ -71,13 +75,45 @@ OpenFlare 当前不定位为通用日志平台、服务网格、Kubernetes Ingre 上游约束： -* `proxy_routes` 至少包含一个上游地址。 +* `proxy_routes` 至少包含一个上游地址（直连类型），或关联一个 Tunnel（内网穿透类型）。 +* `proxy_routes.upstream_type` 区分上游类型：`direct`（默认，直连）或 `tunnel`（内网穿透）。 * 为兼容历史数据保留 `origin_url` 主上游字段，也允许在同一规则内补充多个上游做负载均衡。 * 上游统一渲染为带 keepalive 的 named `upstream`。 * 单上游可附带 base path 或 query 并在 `proxy_pass` 中追加。 * 多上游限定为纯 `scheme://host[:port]`。 * `proxy_routes.origin_host` 为可选字段，用于回源时覆盖 `Host` 请求头。 -* 所有上游地址都必须为合法 `http://` 或 `https://`。 +* 所有直连类型上游地址都必须为合法 `http://` 或 `https://`。 +* 内网穿透类型上游必须关联 `tunnel_id`，并指定内网目标地址与协议。 + +## 内网穿透约束 + +OpenFlare 通过 TunnelRelay 节点与 OpenFlared 客户端实现内网穿透，底层基于 frp 构建。 + +节点类型： + +* `nodes.node_type` 区分节点类型：`edge_node`（边缘节点，默认）和 `tunnel_relay`（隧道中继）。 +* TunnelRelay 节点同时运行 Agent（管理 OpenResty）和 Relay（管理 frps），共享同一个 `agent_token`。 +* Agent 负责 HTTPS 终结、WAF 防护等，Relay 负责隧道流量中继。 + +Tunnel 实体： + +* `tunnels` 表存储内网穿透客户端注册信息，与 `nodes` 体系独立。 +* 每个 Tunnel 拥有唯一的 `tunnel_id`（格式 `tun-<32hex>`）和 `tunnel_token`。 +* OpenFlared 客户端使用 `tunnel_token` 认证，通过 `/api/flared/*` 端点通信。 + +流量路径： + +* 数据面：浏览器 → Agent（OpenResty，TLS/WAF）→ Relay（frps，HTTP Vhost 路由）→ 隧道 → Client（frpc）→ 内网服务。 +* frps 使用 HTTP Vhost 单端口复用，通过 Host 头将请求路由到对应 frpc，无需为每个隧道分配端口。 +* Relay 配置（frps 端口、认证 Token）通过心跳下发，相对静态。 +* Tunnel 路由配置（frpc 代理定义）随发布流程版本化同步。 + +当前阶段约束： + +* 仅支持 HTTP 协议隧道流量，保留未来 TCP 隧道扩展性。 +* Tunnel 类型上游的域名 DNS 应仅解析到 TunnelRelay 节点，EdgeNode 上对应请求会因 frps 不可达返回 502。 +* 一个 OpenFlared 客户端可连接多个 Relay（每个 Relay 对应一个 frpc 进程）。 + ## HTTPS 约束 diff --git a/openflare_server/controller/agent.go b/openflare_server/controller/agent.go index 70f74fea..2f8a3b2e 100644 --- a/openflare_server/controller/agent.go +++ b/openflare_server/controller/agent.go @@ -191,31 +191,26 @@ func agentWSReadTimeout() time.Duration { return timeout } -func streamAgentWSMessages(c *gin.Context, conn *websocket.Conn, client *service.AgentWSClient) { - pingTicker := time.NewTicker(30 * time.Second) - defer pingTicker.Stop() +func agentWSWriteTimeout() time.Duration { + return 10 * time.Second +} + +func streamAgentWSMessages(c *gin.Context, conn *websocket.Conn, client *service.WSClient) { for { select { - case message := <-client.Messages(): - slog.Debug("agent ws sending message", "node_id", client.NodeID(), "type", message.Type) - if err := websocket.JSON.Send(conn, message); err != nil { - slog.Debug("agent ws send failed", "node_id", client.NodeID(), "type", message.Type, "error", err) - client.Close() - return - } - case <-pingTicker.C: - message := service.AgentWSOutboundMessage{Type: service.AgentWSMessageTypePing} - slog.Debug("agent ws sending ping", "node_id", client.NodeID()) - if err := websocket.JSON.Send(conn, message); err != nil { - slog.Debug("agent ws ping failed", "node_id", client.NodeID(), "error", err) - client.Close() - return - } + case <-c.Request.Context().Done(): + return case <-client.Done(): return - case <-c.Request.Context().Done(): - client.Close() - 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("agent ws send failed", "node_id", client.ID(), "error", err) + return + } } } } diff --git a/openflare_server/controller/flared.go b/openflare_server/controller/flared.go new file mode 100644 index 00000000..b0d08a5a --- /dev/null +++ b/openflare_server/controller/flared.go @@ -0,0 +1,171 @@ +package controller + +import ( + "log/slog" + "net" + "openflare/model" + "openflare/service" + "time" + + "github.com/gin-gonic/gin" + "golang.org/x/net/websocket" +) + +// FlaredHeartbeat godoc +// @Summary Report OpenFlared client 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 + } + authTunnel, ok := c.Get("tunnel") + if !ok { + respondUnauthorized(c, "无权进行此操作") + return + } + tunnel := authTunnel.(*model.Tunnel) + result, err := service.HeartbeatFlared(tunnel, payload) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, result) +} + +// FlaredGetActiveConfig godoc +// @Summary Get active tunnel config for OpenFlared client +// @Tags Flared +// @Produce json +// @Security TunnelTokenAuth +// @Success 200 {object} map[string]interface{} +// @Router /api/flared/config/active [get] +func FlaredGetActiveConfig(c *gin.Context) { + authTunnel, ok := c.Get("tunnel") + if !ok { + respondUnauthorized(c, "无权进行此操作") + return + } + tunnel := authTunnel.(*model.Tunnel) + config, err := service.GetFlaredTunnelConfig(tunnel) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, config) +} + +// FlaredReportApplyLog godoc +// @Summary Report apply log for OpenFlared client +// @Tags Flared +// @Accept json +// @Produce json +// @Security TunnelTokenAuth +// @Param payload body service.ApplyLogPayload true "Apply log payload" +// @Success 200 {object} map[string]interface{} +// @Failure 400 {object} map[string]interface{} +// @Router /api/flared/apply-log [post] +func FlaredReportApplyLog(c *gin.Context) { + var payload service.ApplyLogPayload + if !bindJSON(c, &payload) { + return + } + authTunnel, ok := c.Get("tunnel") + if !ok { + respondUnauthorized(c, "无权进行此操作") + return + } + tunnel := authTunnel.(*model.Tunnel) + payload.NodeID = tunnel.TunnelID + 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) { + authTunnel, ok := c.Get("tunnel") + if !ok { + respondUnauthorized(c, "无权进行此操作") + return + } + tunnel := authTunnel.(*model.Tunnel) + slog.Debug("flared ws upgrade requested", "tunnel_id", tunnel.TunnelID, "remote", c.Request.RemoteAddr) + websocket.Handler(func(conn *websocket.Conn) { + client := service.RegisterFlaredWSClient(tunnel.TunnelID) + defer service.UnregisterFlaredWSClient(client) + defer func() { + _ = conn.Close() + slog.Debug("flared ws connection closed", "tunnel_id", tunnel.TunnelID) + }() + + slog.Debug("flared ws upgrade succeeded", "tunnel_id", tunnel.TunnelID, "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(agentWSReadTimeout())) + if err := websocket.JSON.Receive(conn, &message); err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + slog.Debug("flared ws receive timeout", "tunnel_id", tunnel.TunnelID) + return + } + slog.Debug("flared ws receive failed", "tunnel_id", tunnel.TunnelID, "error", err) + return + } + slog.Debug("flared ws message received", "tunnel_id", tunnel.TunnelID, "type", message.Type) + switch message.Type { + case "status": + // Handle status if needed for flared + case "ping": + if !service.SendFlaredWSPong(tunnel.TunnelID) { + slog.Debug("flared ws pong enqueue failed", "tunnel_id", tunnel.TunnelID) + } + case "pong": + slog.Debug("flared ws pong received", "tunnel_id", tunnel.TunnelID) + default: + slog.Debug("flared ws unsupported message type", "tunnel_id", tunnel.TunnelID, "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", "tunnel_id", client.ID(), "error", err) + return + } + } + } +} diff --git a/openflare_server/controller/relay.go b/openflare_server/controller/relay.go new file mode 100644 index 00000000..ac622b19 --- /dev/null +++ b/openflare_server/controller/relay.go @@ -0,0 +1,117 @@ +package controller + +import ( + "log/slog" + "net" + "openflare/model" + "openflare/service" + "time" + + "github.com/gin-gonic/gin" + "golang.org/x/net/websocket" +) + +// RelayHeartbeat godoc +// @Summary Report relay heartbeat +// @Tags Relay +// @Accept json +// @Produce json +// @Security AgentTokenAuth +// @Param payload body service.RelayHeartbeatPayload true "Relay heartbeat payload" +// @Success 200 {object} map[string]interface{} +// @Failure 400 {object} map[string]interface{} +// @Router /api/relay/heartbeat [post] +func RelayHeartbeat(c *gin.Context) { + var payload service.RelayHeartbeatPayload + if !bindJSON(c, &payload) { + return + } + authNode, ok := c.Get("relay_node") + if !ok { + respondUnauthorized(c, "无权进行此操作") + return + } + node := authNode.(*model.Node) + result, err := service.HeartbeatRelay(node, payload) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, result) +} + +// RelayWebSocket godoc +// @Summary Upgrade relay connection to websocket +// @Tags Relay +// @Security AgentTokenAuth +// @Router /api/relay/ws [get] +func RelayWebSocket(c *gin.Context) { + authNode, ok := c.Get("relay_node") + if !ok { + respondUnauthorized(c, "无权进行此操作") + return + } + node := authNode.(*model.Node) + slog.Debug("relay ws upgrade requested", "node_id", node.NodeID, "remote", c.Request.RemoteAddr) + websocket.Handler(func(conn *websocket.Conn) { + client := service.RegisterRelayWSClient(node.NodeID) + defer service.UnregisterRelayWSClient(client) + defer func() { + _ = conn.Close() + slog.Debug("relay ws connection closed", "node_id", node.NodeID) + }() + + slog.Debug("relay ws upgrade succeeded", "node_id", node.NodeID, "remote", c.Request.RemoteAddr) + + go func() { + <-client.Done() + _ = conn.Close() + }() + + go streamRelayWSMessages(c, conn, client) + + for { + var message service.WSMessage + _ = conn.SetReadDeadline(time.Now().Add(agentWSReadTimeout())) + if err := websocket.JSON.Receive(conn, &message); err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + slog.Debug("relay ws receive timeout", "node_id", node.NodeID) + return + } + slog.Debug("relay ws receive failed", "node_id", node.NodeID, "error", err) + return + } + slog.Debug("relay ws message received", "node_id", node.NodeID, "type", message.Type) + switch message.Type { + case "ping": + if !service.SendRelayWSPong(node.NodeID) { + slog.Debug("relay ws pong enqueue failed", "node_id", node.NodeID) + } + case "pong": + slog.Debug("relay ws pong received", "node_id", node.NodeID) + default: + slog.Debug("relay ws unsupported message type", "node_id", node.NodeID, "type", message.Type) + } + } + }).ServeHTTP(c.Writer, c.Request) +} + +func streamRelayWSMessages(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("relay ws send failed", "node_id", client.ID(), "error", err) + return + } + } + } +} diff --git a/openflare_server/controller/tunnel.go b/openflare_server/controller/tunnel.go new file mode 100644 index 00000000..6a8bb2ec --- /dev/null +++ b/openflare_server/controller/tunnel.go @@ -0,0 +1,84 @@ +package controller + +import ( + "openflare/service" + + "github.com/gin-gonic/gin" +) + +func GetTunnels(c *gin.Context) { + tunnels, err := service.ListTunnels() + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, tunnels) +} + +func GetTunnel(c *gin.Context) { + id, ok := parseIDParam(c) + if !ok { + return + } + tunnel, err := service.GetTunnel(id) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, tunnel) +} + +func CreateTunnel(c *gin.Context) { + var input service.TunnelInput + if !bindJSON(c, &input) { + return + } + tunnel, err := service.CreateTunnel(input) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, tunnel) +} + +func UpdateTunnel(c *gin.Context) { + id, ok := parseIDParam(c) + if !ok { + return + } + var input service.TunnelInput + if !bindJSON(c, &input) { + return + } + tunnel, err := service.UpdateTunnel(id, input) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, tunnel) +} + +func DeleteTunnel(c *gin.Context) { + id, ok := parseIDParam(c) + if !ok { + return + } + if err := service.DeleteTunnel(id); err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, nil) +} + +func RotateTunnelToken(c *gin.Context) { + id, ok := parseIDParam(c) + if !ok { + return + } + tunnel, err := service.RotateTunnelToken(id) + if err != nil { + respondFailure(c, err.Error()) + return + } + respondSuccess(c, tunnel) +} diff --git a/openflare_server/middleware/relay-auth.go b/openflare_server/middleware/relay-auth.go new file mode 100644 index 00000000..bdbc54a9 --- /dev/null +++ b/openflare_server/middleware/relay-auth.go @@ -0,0 +1,34 @@ +package middleware + +import ( + "github.com/gin-gonic/gin" + "net/http" + "openflare/service" +) + +// RelayAuth authenticates Relay requests using the shared agent token, +// and verifies the node is a tunnel_relay type. +func RelayAuth() func(c *gin.Context) { + return func(c *gin.Context) { + token := c.GetHeader("X-Agent-Token") + node, err := service.AuthenticateAgentToken(token) + if err != nil { + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": "无权进行此操作，Agent Token 无效", + }) + c.Abort() + return + } + if node.NodeType != "tunnel_relay" { + c.JSON(http.StatusForbidden, gin.H{ + "success": false, + "message": "此节点不是 TunnelRelay 类型", + }) + c.Abort() + return + } + c.Set("relay_node", node) + c.Next() + } +} diff --git a/openflare_server/middleware/tunnel-auth.go b/openflare_server/middleware/tunnel-auth.go new file mode 100644 index 00000000..9f8d69bc --- /dev/null +++ b/openflare_server/middleware/tunnel-auth.go @@ -0,0 +1,25 @@ +package middleware + +import ( + "github.com/gin-gonic/gin" + "net/http" + "openflare/service" +) + +// TunnelAuth authenticates OpenFlared client requests using X-Tunnel-Token. +func TunnelAuth() func(c *gin.Context) { + return func(c *gin.Context) { + token := c.GetHeader("X-Tunnel-Token") + tunnel, err := service.AuthenticateTunnelToken(token) + if err != nil { + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": "无权进行此操作，Tunnel Token 无效", + }) + c.Abort() + return + } + c.Set("tunnel", tunnel) + c.Next() + } +} diff --git a/openflare_server/model/main.go b/openflare_server/model/main.go index 94980bdf..d0fb5c9a 100644 --- a/openflare_server/model/main.go +++ b/openflare_server/model/main.go @@ -34,6 +34,7 @@ func registeredModels() []any { &ProxyRoute{}, &ConfigVersion{}, &Node{}, + &Tunnel{}, &NodeSystemProfile{}, &ApplyLog{}, &NodeMetricSnapshot{}, diff --git a/openflare_server/model/migrate/v16.go b/openflare_server/model/migrate/v16.go new file mode 100644 index 00000000..7a189f65 --- /dev/null +++ b/openflare_server/model/migrate/v16.go @@ -0,0 +1,82 @@ +// v16 升级内容：新增 Tunnel 隧道表，为 nodes 增加 TunnelRelay 相关字段，为 proxy_routes 增加隧道上游支持字段。 +// 背景说明：引入 Tunnel 组件后，边缘节点可通过隧道中继回源，需要区分节点类型（edge_node/relay_node）、记录中继绑定端口， +// 并在代理路由中支持 tunnel 上游类型以替代直连回源，同时需要独立的 tunnels 表管理隧道实例。 +package migrate + +import ( + "fmt" + + "gorm.io/gorm" +) + +type nodeV16 struct { + NodeType string `gorm:"column:node_type;not null;default:'edge_node'"` + RelayBindPort int `gorm:"column:relay_bind_port"` +} + +type tunnelV16 struct{} + +type proxyRouteV16 struct { + UpstreamType string `gorm:"column:upstream_type;not null;default:'direct'"` + TunnelID *uint `gorm:"column:tunnel_id"` +} + +func init() { + Register(V16()) +} + +func V16() Migration { + return Migration{ + FromVersion: 15, + ToVersion: 16, + Migrate: migrateV16, + Validate: validateV16, + } +} + +func (nodeV16) TableName() string { + return "nodes" +} + +func (tunnelV16) TableName() string { + return "tunnels" +} + +func (proxyRouteV16) TableName() string { + return "proxy_routes" +} + +func migrateV16(ctx Context, db *gorm.DB, backend string) error { + if err := ctx.ApplyCurrentSchema(db, backend); err != nil { + return err + } + if err := db.Exec("UPDATE nodes SET node_type = 'edge_node' WHERE node_type = '' OR node_type IS NULL").Error; err != nil { + return fmt.Errorf("backfill nodes.node_type: %w", err) + } + if err := db.Exec("UPDATE proxy_routes SET upstream_type = 'direct' WHERE upstream_type = '' OR upstream_type IS NULL").Error; err != nil { + return fmt.Errorf("backfill proxy_routes.upstream_type: %w", err) + } + return nil +} + +func validateV16(ctx Context, db *gorm.DB, backend string) error { + if err := ctx.ValidateDatabaseSchemaVersion(db, backend, 15); err != nil { + return err + } + if db == nil || !db.Migrator().HasTable(&tunnelV16{}) { + return fmt.Errorf("table tunnels is missing") + } + if !db.Migrator().HasColumn(&nodeV16{}, "node_type") { + return fmt.Errorf("column nodes.node_type is missing") + } + if !db.Migrator().HasColumn(&proxyRouteV16{}, "upstream_type") { + return fmt.Errorf("column proxy_routes.upstream_type is missing") + } + if !db.Migrator().HasColumn(&nodeV16{}, "relay_bind_port") { + return fmt.Errorf("column nodes.relay_bind_port is missing") + } + if !db.Migrator().HasColumn(&proxyRouteV16{}, "tunnel_id") { + return fmt.Errorf("column proxy_routes.tunnel_id is missing") + } + return nil +} diff --git a/openflare_server/model/migrations.go b/openflare_server/model/migrations.go index 7b35dc23..c616543a 100644 --- a/openflare_server/model/migrations.go +++ b/openflare_server/model/migrations.go @@ -74,6 +74,10 @@ func (databaseSchemaMigrationContext) ValidateDatabaseSchemaVersion(db *gorm.DB, return validateDatabaseSchemaV13(db, backend) case 14: return validateDatabaseSchemaV14(db, backend) + case 15: + return validateDatabaseSchemaV15(db, backend) + case 16: + return validateDatabaseSchemaV16(db, backend) default: return fmt.Errorf("database schema validation for v%d is not defined", version) } @@ -1140,6 +1144,32 @@ func validateDatabaseSchemaV14(db *gorm.DB, backend string) error { return nil } +func validateDatabaseSchemaV15(db *gorm.DB, backend string) error { + if err := validateDatabaseSchemaV14(db, backend); err != nil { + return err + } + if !db.Migrator().HasColumn(&Node{}, "ip_manual_override") { + return fmt.Errorf("column nodes.ip_manual_override is missing") + } + return nil +} + +func validateDatabaseSchemaV16(db *gorm.DB, backend string) error { + if err := validateDatabaseSchemaV15(db, backend); err != nil { + return err + } + if !db.Migrator().HasTable(&Tunnel{}) { + return fmt.Errorf("table tunnels is missing") + } + if !db.Migrator().HasColumn(&Node{}, "node_type") { + return fmt.Errorf("column nodes.node_type is missing") + } + if !db.Migrator().HasColumn(&ProxyRoute{}, "upstream_type") { + return fmt.Errorf("column proxy_routes.upstream_type is missing") + } + return nil +} + func databaseSchemaMigrations() []databaseSchemaMigration { ctx := databaseSchemaMigrationContext{} migrations := []databaseSchemaMigration{} diff --git a/openflare_server/model/node.go b/openflare_server/model/node.go index a7f8e679..2e8c98f1 100644 --- a/openflare_server/model/node.go +++ b/openflare_server/model/node.go @@ -28,6 +28,18 @@ type Node struct { LastError string `json:"last_error" gorm:"type:text"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` + // Node type: edge_node (default) | tunnel_relay + NodeType string `json:"node_type" gorm:"size:32;not null;default:'edge_node'"` + // TunnelRelay specific fields + RelayBindPort int `json:"relay_bind_port" gorm:"not null;default:0"` + RelayVhostHTTPPort int `json:"relay_vhost_http_port" gorm:"not null;default:0"` + RelayAuthToken string `json:"-" gorm:"size:128"` + RelayAgentAccessAddr string `json:"relay_agent_access_addr" gorm:"size:255"` + RelayClientAccessAddr string `json:"relay_client_access_addr" gorm:"size:255"` + RelayClientProxyURL string `json:"relay_client_proxy_url" gorm:"size:512"` + RelayStatus string `json:"relay_status" gorm:"size:16;not null;default:'unknown'"` + RelayFrpVersion string `json:"relay_frp_version" gorm:"size:64"` + RelayVersion string `json:"relay_version" gorm:"size:64"` } func ListNodes() (nodes []*Node, err error) { @@ -72,3 +84,8 @@ func (node *Node) Update() error { func (node *Node) Delete() error { return DB.Delete(node).Error } + +func ListNodesByType(nodeType string) (nodes []*Node, err error) { + err = DB.Where("node_type = ?", nodeType).Order("id desc").Find(&nodes).Error + return nodes, err +} diff --git a/openflare_server/model/proxy_route.go b/openflare_server/model/proxy_route.go index 17dd35e6..26cadc86 100644 --- a/openflare_server/model/proxy_route.go +++ b/openflare_server/model/proxy_route.go @@ -3,35 +3,39 @@ package model import "time" type ProxyRoute struct { - ID uint `json:"id" gorm:"primaryKey"` - SiteName string `json:"site_name" gorm:"size:255;not null;default:''"` - Domain string `json:"domain" gorm:"uniqueIndex;size:255;not null"` - Domains string `json:"domains" gorm:"type:text;not null;default:'[]'"` - OriginID *uint `json:"origin_id" gorm:"index"` - OriginURL string `json:"origin_url" gorm:"size:2048;not null"` - OriginHost string `json:"origin_host" gorm:"size:255"` - Upstreams string `json:"upstreams" gorm:"type:text;not null;default:'[]'"` - Enabled bool `json:"enabled" gorm:"not null;default:true"` - EnableHTTPS bool `json:"enable_https" gorm:"column:enable_https;not null;default:false"` - CertID *uint `json:"cert_id"` - CertIDs string `json:"cert_ids" gorm:"type:text;not null;default:'[]'"` - DomainCertIDs string `json:"domain_cert_ids" gorm:"type:text;not null;default:'[]'"` - RedirectHTTP bool `json:"redirect_http" gorm:"not null;default:false"` - LimitConnPerServer int `json:"limit_conn_per_server" gorm:"not null;default:0"` - LimitConnPerIP int `json:"limit_conn_per_ip" gorm:"not null;default:0"` - LimitRate string `json:"limit_rate" gorm:"size:32;not null;default:''"` - CacheEnabled bool `json:"cache_enabled" gorm:"not null;default:false"` - CachePolicy string `json:"cache_policy" gorm:"size:32;not null;default:''"` - CacheRules string `json:"cache_rules" gorm:"type:text;not null;default:'[]'"` - CustomHeaders string `json:"custom_headers" gorm:"type:text;not null;default:'[]'"` - PoWEnabled bool `json:"pow_enabled" gorm:"column:pow_enabled;not null;default:false"` - PoWConfig string `json:"pow_config" gorm:"column:pow_config;type:text;not null;default:'{}'"` - BasicAuthEnabled bool `json:"basic_auth_enabled" gorm:"not null;default:false"` - BasicAuthUsername string `json:"basic_auth_username" gorm:"size:255;not null;default:''"` - BasicAuthPassword string `json:"basic_auth_password" gorm:"size:255;not null;default:''"` - Remark string `json:"remark" gorm:"size:255"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint `json:"id" gorm:"primaryKey"` + SiteName string `json:"site_name" gorm:"size:255;not null;default:''"` + Domain string `json:"domain" gorm:"uniqueIndex;size:255;not null"` + Domains string `json:"domains" gorm:"type:text;not null;default:'[]'"` + OriginID *uint `json:"origin_id" gorm:"index"` + OriginURL string `json:"origin_url" gorm:"size:2048;not null"` + OriginHost string `json:"origin_host" gorm:"size:255"` + Upstreams string `json:"upstreams" gorm:"type:text;not null;default:'[]'"` + Enabled bool `json:"enabled" gorm:"not null;default:true"` + EnableHTTPS bool `json:"enable_https" gorm:"column:enable_https;not null;default:false"` + CertID *uint `json:"cert_id"` + CertIDs string `json:"cert_ids" gorm:"type:text;not null;default:'[]'"` + DomainCertIDs string `json:"domain_cert_ids" gorm:"type:text;not null;default:'[]'"` + RedirectHTTP bool `json:"redirect_http" gorm:"not null;default:false"` + LimitConnPerServer int `json:"limit_conn_per_server" gorm:"not null;default:0"` + LimitConnPerIP int `json:"limit_conn_per_ip" gorm:"not null;default:0"` + LimitRate string `json:"limit_rate" gorm:"size:32;not null;default:''"` + CacheEnabled bool `json:"cache_enabled" gorm:"not null;default:false"` + CachePolicy string `json:"cache_policy" gorm:"size:32;not null;default:''"` + CacheRules string `json:"cache_rules" gorm:"type:text;not null;default:'[]'"` + CustomHeaders string `json:"custom_headers" gorm:"type:text;not null;default:'[]'"` + PoWEnabled bool `json:"pow_enabled" gorm:"column:pow_enabled;not null;default:false"` + PoWConfig string `json:"pow_config" gorm:"column:pow_config;type:text;not null;default:'{}'"` + BasicAuthEnabled bool `json:"basic_auth_enabled" gorm:"not null;default:false"` + BasicAuthUsername string `json:"basic_auth_username" gorm:"size:255;not null;default:''"` + BasicAuthPassword string `json:"basic_auth_password" gorm:"size:255;not null;default:''"` + Remark string `json:"remark" gorm:"size:255"` + UpstreamType string `json:"upstream_type" gorm:"size:32;not null;default:'direct'"` + TunnelID *uint `json:"tunnel_id" gorm:"index"` + TunnelTargetAddr string `json:"tunnel_target_addr" gorm:"size:512"` + TunnelTargetProtocol string `json:"tunnel_target_protocol" gorm:"size:16"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } func ListProxyRoutes() (routes []*ProxyRoute, err error) { @@ -61,32 +65,36 @@ func (route *ProxyRoute) Insert() error { func (route *ProxyRoute) Update() error { return DB.Model(&ProxyRoute{}).Where("id = ?", route.ID).Updates(map[string]any{ - "site_name": route.SiteName, - "domain": route.Domain, - "domains": route.Domains, - "origin_id": route.OriginID, - "origin_url": route.OriginURL, - "origin_host": route.OriginHost, - "upstreams": route.Upstreams, - "enabled": route.Enabled, - "enable_https": route.EnableHTTPS, - "cert_id": route.CertID, - "cert_ids": route.CertIDs, - "domain_cert_ids": route.DomainCertIDs, - "redirect_http": route.RedirectHTTP, - "limit_conn_per_server": route.LimitConnPerServer, - "limit_conn_per_ip": route.LimitConnPerIP, - "limit_rate": route.LimitRate, - "cache_enabled": route.CacheEnabled, - "cache_policy": route.CachePolicy, - "cache_rules": route.CacheRules, - "custom_headers": route.CustomHeaders, - "pow_enabled": route.PoWEnabled, - "pow_config": route.PoWConfig, - "basic_auth_enabled": route.BasicAuthEnabled, - "basic_auth_username": route.BasicAuthUsername, - "basic_auth_password": route.BasicAuthPassword, - "remark": route.Remark, + "site_name": route.SiteName, + "domain": route.Domain, + "domains": route.Domains, + "origin_id": route.OriginID, + "origin_url": route.OriginURL, + "origin_host": route.OriginHost, + "upstreams": route.Upstreams, + "enabled": route.Enabled, + "enable_https": route.EnableHTTPS, + "cert_id": route.CertID, + "cert_ids": route.CertIDs, + "domain_cert_ids": route.DomainCertIDs, + "redirect_http": route.RedirectHTTP, + "limit_conn_per_server": route.LimitConnPerServer, + "limit_conn_per_ip": route.LimitConnPerIP, + "limit_rate": route.LimitRate, + "cache_enabled": route.CacheEnabled, + "cache_policy": route.CachePolicy, + "cache_rules": route.CacheRules, + "custom_headers": route.CustomHeaders, + "pow_enabled": route.PoWEnabled, + "pow_config": route.PoWConfig, + "basic_auth_enabled": route.BasicAuthEnabled, + "basic_auth_username": route.BasicAuthUsername, + "basic_auth_password": route.BasicAuthPassword, + "remark": route.Remark, + "upstream_type": route.UpstreamType, + "tunnel_id": route.TunnelID, + "tunnel_target_addr": route.TunnelTargetAddr, + "tunnel_target_protocol": route.TunnelTargetProtocol, }).Error } diff --git a/openflare_server/model/tunnel.go b/openflare_server/model/tunnel.go new file mode 100644 index 00000000..362f3340 --- /dev/null +++ b/openflare_server/model/tunnel.go @@ -0,0 +1,56 @@ +package model + +import "time" + +type Tunnel struct { + ID uint `json:"id" gorm:"primaryKey"` + TunnelID string `json:"tunnel_id" gorm:"uniqueIndex;size:64;not null"` + Name string `json:"name" gorm:"size:128;not null"` + TunnelToken string `json:"-" gorm:"size:128;index"` + Status string `json:"status" gorm:"size:16;not null;default:'offline'"` + ClientVersion string `json:"client_version" gorm:"size:64"` + FrpVersion string `json:"frp_version" gorm:"size:64"` + LastSeenAt time.Time `json:"last_seen_at"` + LastError string `json:"last_error" gorm:"type:text"` + CurrentVersion string `json:"current_version" gorm:"size:32"` + CurrentChecksum string `json:"current_checksum" gorm:"size:64"` + ConnectedRelays string `json:"connected_relays" gorm:"type:text;not null;default:'[]'"` + Remark string `json:"remark" gorm:"size:255"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +func ListTunnels() (tunnels []*Tunnel, err error) { + err = DB.Order("id desc").Find(&tunnels).Error + return tunnels, err +} + +func GetTunnelByID(id uint) (*Tunnel, error) { + tunnel := &Tunnel{} + err := DB.First(tunnel, id).Error + return tunnel, err +} + +func GetTunnelByTunnelID(tunnelID string) (*Tunnel, error) { + tunnel := &Tunnel{} + err := DB.Where("tunnel_id = ?", tunnelID).First(tunnel).Error + return tunnel, err +} + +func GetTunnelByTunnelToken(token string) (*Tunnel, error) { + tunnel := &Tunnel{} + err := DB.Where("tunnel_token = ?", token).First(tunnel).Error + return tunnel, err +} + +func (tunnel *Tunnel) Insert() error { + return DB.Create(tunnel).Error +} + +func (tunnel *Tunnel) Update() error { + return DB.Save(tunnel).Error +} + +func (tunnel *Tunnel) Delete() error { + return DB.Delete(tunnel).Error +} diff --git a/openflare_server/router/api-router.go b/openflare_server/router/api-router.go index d1e53d4c..d2d78053 100644 --- a/openflare_server/router/api-router.go +++ b/openflare_server/router/api-router.go @@ -191,6 +191,16 @@ func SetApiRouter(router *gin.Engine) { applyLogRoute.GET("/", controller.GetApplyLogs) applyLogRoute.POST("/cleanup", controller.CleanupApplyLogs) } + tunnelRoute := apiRouter.Group("/tunnels") + tunnelRoute.Use(middleware.AdminAuth()) + { + tunnelRoute.GET("/", controller.GetTunnels) + tunnelRoute.GET("/:id", controller.GetTunnel) + tunnelRoute.POST("/", controller.CreateTunnel) + tunnelRoute.POST("/:id/update", controller.UpdateTunnel) + tunnelRoute.POST("/:id/delete", controller.DeleteTunnel) + tunnelRoute.POST("/:id/rotate-token", controller.RotateTunnelToken) + } accessLogRoute := apiRouter.Group("/access-logs") accessLogRoute.Use(middleware.AdminAuth()) { @@ -217,5 +227,19 @@ func SetApiRouter(router *gin.Engine) { authorizedRoute.POST("/apply-logs", controller.AgentReportApplyLog) } } + relayRoute := apiRouter.Group("/relay") + relayRoute.Use(middleware.RelayAuth()) + { + 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/service/agent.go b/openflare_server/service/agent.go index 3bcd54e6..4312f07c 100644 --- a/openflare_server/service/agent.go +++ b/openflare_server/service/agent.go @@ -142,6 +142,16 @@ type NodeView struct { LatestApplyAt *time.Time `json:"latest_apply_at"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` + // TunnelRelay fields + NodeType string `json:"node_type"` + RelayBindPort int `json:"relay_bind_port"` + RelayVhostHTTPPort int `json:"relay_vhost_http_port"` + RelayAgentAccessAddr string `json:"relay_agent_access_addr"` + RelayClientAccessAddr string `json:"relay_client_access_addr"` + RelayClientProxyURL string `json:"relay_client_proxy_url"` + RelayStatus string `json:"relay_status"` + RelayFrpVersion string `json:"relay_frp_version"` + RelayVersion string `json:"relay_version"` } func HeartbeatNode(node *model.Node, payload AgentNodePayload) (*HeartbeatResponse, error) { diff --git a/openflare_server/service/agent_ws.go b/openflare_server/service/agent_ws.go index f39c459c..a4c4700e 100644 --- a/openflare_server/service/agent_ws.go +++ b/openflare_server/service/agent_ws.go @@ -3,7 +3,6 @@ package service import ( "encoding/json" "log/slog" - "sync" ) const ( @@ -22,11 +21,6 @@ type AgentWSInboundMessage struct { Payload json.RawMessage `json:"payload,omitempty"` } -type AgentWSOutboundMessage struct { - Type string `json:"type"` - Payload any `json:"payload,omitempty"` -} - type AgentWSBroadcastResult struct { Version string `json:"version"` Checksum string `json:"checksum"` @@ -35,133 +29,29 @@ type AgentWSBroadcastResult struct { FailedNodes []string `json:"failed_nodes"` } -type AgentWSClient struct { - nodeID string - send chan AgentWSOutboundMessage - done chan struct{} - once sync.Once +var DefaultAgentWSHub = NewWSHub("agent") + +func RegisterAgentWSClient(nodeID string) *WSClient { + return DefaultAgentWSHub.Register(nodeID) } -func (client *AgentWSClient) NodeID() string { - if client == nil { - return "" - } - return client.nodeID -} - -func (client *AgentWSClient) Messages() <-chan AgentWSOutboundMessage { - if client == nil { - return nil - } - return client.send -} - -func (client *AgentWSClient) Done() <-chan struct{} { - if client == nil { - return nil - } - return client.done -} - -func (client *AgentWSClient) Send(message AgentWSOutboundMessage) bool { - if client == nil { - return false - } - select { - case <-client.done: - return false - case client.send <- message: - return true - default: - return false - } -} - -func (client *AgentWSClient) Close() { - if client == nil { - return - } - client.once.Do(func() { - close(client.done) - }) -} - -type agentWSHub struct { - mu sync.RWMutex - clients map[string]*AgentWSClient -} - -var defaultAgentWSHub = &agentWSHub{ - clients: make(map[string]*AgentWSClient), -} - -func RegisterAgentWSClient(nodeID string) *AgentWSClient { - client := &AgentWSClient{ - nodeID: nodeID, - send: make(chan AgentWSOutboundMessage, 16), - done: make(chan struct{}), - } - defaultAgentWSHub.mu.Lock() - if existing := defaultAgentWSHub.clients[nodeID]; existing != nil { - slog.Debug("agent ws replacing existing connection", "node_id", nodeID) - existing.Close() - } - defaultAgentWSHub.clients[nodeID] = client - count := len(defaultAgentWSHub.clients) - defaultAgentWSHub.mu.Unlock() - slog.Debug("agent ws connection registered", "node_id", nodeID, "client_count", count) - return client -} - -func UnregisterAgentWSClient(client *AgentWSClient) { - if client == nil { - return - } - defaultAgentWSHub.mu.Lock() - if current := defaultAgentWSHub.clients[client.nodeID]; current == client { - delete(defaultAgentWSHub.clients, client.nodeID) - } - count := len(defaultAgentWSHub.clients) - defaultAgentWSHub.mu.Unlock() - client.Close() - slog.Debug("agent ws connection unregistered", "node_id", client.nodeID, "client_count", count) +func UnregisterAgentWSClient(client *WSClient) { + DefaultAgentWSHub.Unregister(client) } func DisconnectAgentWSClient(nodeID string) { - defaultAgentWSHub.mu.Lock() - client := defaultAgentWSHub.clients[nodeID] - if client != nil { - delete(defaultAgentWSHub.clients, nodeID) - } - count := len(defaultAgentWSHub.clients) - defaultAgentWSHub.mu.Unlock() - - if client != nil { - client.Close() - slog.Debug("agent ws connection forcefully disconnected", "node_id", nodeID, "client_count", count) - } + DefaultAgentWSHub.Disconnect(nodeID) } func IsAgentWSConnected(nodeID string) bool { - defaultAgentWSHub.mu.RLock() - client := defaultAgentWSHub.clients[nodeID] - defaultAgentWSHub.mu.RUnlock() - if client == nil { - return false - } - select { - case <-client.done: - return false - default: - return true - } + return DefaultAgentWSHub.IsConnected(nodeID) } func SendAgentWSSettings(nodeID string, settings *AgentSettings) bool { if settings == nil { return false } - return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{ + return DefaultAgentWSHub.SendMessage(nodeID, WSMessage{ Type: AgentWSMessageTypeSettings, Payload: settings, }) @@ -171,7 +61,7 @@ func SendAgentWSActiveConfig(nodeID string, activeConfig *ActiveConfigMeta) bool if activeConfig == nil { return false } - return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{ + return DefaultAgentWSHub.SendMessage(nodeID, WSMessage{ Type: AgentWSMessageTypeActiveConfig, Payload: activeConfig, }) @@ -181,60 +71,37 @@ func SendAgentWSForceSyncConfig(nodeID string, activeConfig *ActiveConfigMeta) b if activeConfig == nil { return false } - return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{ + return DefaultAgentWSHub.SendMessage(nodeID, WSMessage{ Type: AgentWSMessageTypeForceSyncConfig, Payload: activeConfig, }) } func SendAgentWSPong(nodeID string) bool { - return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{ + return DefaultAgentWSHub.SendMessage(nodeID, WSMessage{ Type: AgentWSMessageTypePong, }) } -func sendAgentWSMessage(nodeID string, message AgentWSOutboundMessage) bool { - defaultAgentWSHub.mu.RLock() - client := defaultAgentWSHub.clients[nodeID] - defaultAgentWSHub.mu.RUnlock() - if client == nil { - return false - } - ok := client.Send(message) - if !ok { - slog.Debug("agent ws send queued message failed", "node_id", nodeID, "type", message.Type) - } - return ok -} - func BroadcastAgentWSActiveConfig(activeConfig *ActiveConfigMeta) AgentWSBroadcastResult { - result := AgentWSBroadcastResult{} if activeConfig == nil { slog.Debug("agent ws broadcast skipped because active config is nil") - return result + return AgentWSBroadcastResult{} } - result.Version = activeConfig.Version - result.Checksum = activeConfig.Checksum - defaultAgentWSHub.mu.RLock() - clients := make([]*AgentWSClient, 0, len(defaultAgentWSHub.clients)) - for _, client := range defaultAgentWSHub.clients { - clients = append(clients, client) - } - defaultAgentWSHub.mu.RUnlock() - - result.ClientCount = len(clients) - message := AgentWSOutboundMessage{ + res := DefaultAgentWSHub.Broadcast(WSMessage{ Type: AgentWSMessageTypeActiveConfig, Payload: activeConfig, + }) + + result := AgentWSBroadcastResult{ + Version: activeConfig.Version, + Checksum: activeConfig.Checksum, + ClientCount: res.ClientCount, + SuccessCount: res.SuccessCount, + FailedNodes: res.FailedIDs, } - for _, client := range clients { - if client.Send(message) { - result.SuccessCount++ - continue - } - result.FailedNodes = append(result.FailedNodes, client.NodeID()) - } + slog.Debug("agent ws broadcast active config", "version", result.Version, "checksum", result.Checksum, diff --git a/openflare_server/service/config_version.go b/openflare_server/service/config_version.go index b863ef46..7858b843 100644 --- a/openflare_server/service/config_version.go +++ b/openflare_server/service/config_version.go @@ -340,10 +340,12 @@ func PublishConfigVersion(createdBy string, force bool) (*ReleaseResult, error) } return nil, err } - BroadcastAgentWSActiveConfig(&ActiveConfigMeta{ + activeConfig := &ActiveConfigMeta{ Version: record.Version, Checksum: record.Checksum, - }) + } + BroadcastAgentWSActiveConfig(activeConfig) + BroadcastFlaredWSActiveConfig(activeConfig) return &ReleaseResult{ Version: record, Routes: bundle.Routes, diff --git a/openflare_server/service/flared_ws.go b/openflare_server/service/flared_ws.go new file mode 100644 index 00000000..54e5decf --- /dev/null +++ b/openflare_server/service/flared_ws.go @@ -0,0 +1,54 @@ +package service + +import ( + "log/slog" +) + +var DefaultFlaredWSHub = NewWSHub("flared") + +func RegisterFlaredWSClient(tunnelID string) *WSClient { + return DefaultFlaredWSHub.Register(tunnelID) +} + +func UnregisterFlaredWSClient(client *WSClient) { + DefaultFlaredWSHub.Unregister(client) +} + +func IsFlaredWSConnected(tunnelID string) bool { + return DefaultFlaredWSHub.IsConnected(tunnelID) +} + +func SendFlaredWSPong(tunnelID string) bool { + return DefaultFlaredWSHub.SendMessage(tunnelID, WSMessage{ + Type: "pong", + }) +} + +func SendFlaredWSActiveConfig(tunnelID string, activeConfig *ActiveConfigMeta) bool { + if activeConfig == nil { + return false + } + return DefaultFlaredWSHub.SendMessage(tunnelID, WSMessage{ + Type: "active_config", + Payload: activeConfig, + }) +} + +func BroadcastFlaredWSActiveConfig(activeConfig *ActiveConfigMeta) WSBroadcastResult { + if activeConfig == nil { + slog.Debug("flared ws broadcast skipped because active config is nil") + return WSBroadcastResult{} + } + res := DefaultFlaredWSHub.Broadcast(WSMessage{ + Type: "active_config", + Payload: activeConfig, + }) + slog.Debug("flared ws broadcast active config", + "version", activeConfig.Version, + "checksum", activeConfig.Checksum, + "client_count", res.ClientCount, + "success_count", res.SuccessCount, + "failed_tunnels", res.FailedIDs, + ) + return res +} diff --git a/openflare_server/service/https_phase1_test.go b/openflare_server/service/https_phase1_test.go index ecfdd23e..a59007b1 100644 --- a/openflare_server/service/https_phase1_test.go +++ b/openflare_server/service/https_phase1_test.go @@ -61,7 +61,7 @@ func TestCreateTLSCertificateAndRenderHTTPSConfig(t *testing.T) { if !strings.Contains(result.Version.MainConfig, "error_log __OPENFLARE_ERROR_LOG__ warn;") { t.Fatal("expected main config to include managed error log placeholder") } - if !strings.Contains(result.Version.MainConfig, "log_by_lua_file __OPENFLARE_LUA_DIR__/log.lua;") { + if !strings.Contains(result.Version.MainConfig, "log_by_lua_file __OPENFLARE_LUA_DIR__/observability/log.lua;") { t.Fatal("expected main config to include managed openresty lua log hook") } if !strings.Contains(result.Version.MainConfig, "listen __OPENFLARE_OBSERVABILITY_LISTEN__;") { @@ -893,7 +893,7 @@ func TestPreviewAndDiffConfigVersion(t *testing.T) { if !strings.Contains(preview.MainConfig, "include __OPENFLARE_ROUTE_CONFIG__;") { t.Fatal("expected preview main config to include managed route config placeholder") } - if !strings.Contains(preview.MainConfig, "log_by_lua_file __OPENFLARE_LUA_DIR__/log.lua;") { + if !strings.Contains(preview.MainConfig, "log_by_lua_file __OPENFLARE_LUA_DIR__/observability/log.lua;") { t.Fatal("expected preview main config to include managed openresty lua log hook") } if !strings.Contains(preview.RenderedConfig, `proxy_set_header X-Release "candidate";`) { diff --git a/openflare_server/service/node.go b/openflare_server/service/node.go index aa1636b6..2045b358 100644 --- a/openflare_server/service/node.go +++ b/openflare_server/service/node.go @@ -24,6 +24,13 @@ type NodeInput struct { GeoLatitude *float64 `json:"geo_latitude"` GeoLongitude *float64 `json:"geo_longitude"` GeoManualOverride bool `json:"geo_manual_override"` + // TunnelRelay fields + NodeType string `json:"node_type"` + RelayBindPort int `json:"relay_bind_port"` + RelayVhostHTTPPort int `json:"relay_vhost_http_port"` + RelayAgentAccessAddr string `json:"relay_agent_access_addr"` + RelayClientAccessAddr string `json:"relay_client_access_addr"` + RelayClientProxyURL string `json:"relay_client_proxy_url"` } type NodeAgentUpdateInput struct { @@ -73,6 +80,7 @@ func CreateNode(input NodeInput) (*NodeView, error) { NginxVersion: "", Status: NodeStatusPending, AutoUpdateEnabled: input.AutoUpdateEnabled, + NodeType: normalizeNodeType(input.NodeType), } node.NodeID, err = newServerNodeID() if err != nil { @@ -82,6 +90,17 @@ func CreateNode(input NodeInput) (*NodeView, error) { if err != nil { return nil, err } + if node.NodeType == "tunnel_relay" { + node.RelayBindPort = normalizeRelayPort(input.RelayBindPort, 7000) + node.RelayVhostHTTPPort = normalizeRelayPort(input.RelayVhostHTTPPort, 8080) + node.RelayAuthToken, err = newRandomToken() + if err != nil { + return nil, err + } + node.RelayAgentAccessAddr = strings.TrimSpace(input.RelayAgentAccessAddr) + node.RelayClientAccessAddr = strings.TrimSpace(input.RelayClientAccessAddr) + node.RelayClientProxyURL = strings.TrimSpace(input.RelayClientProxyURL) + } if !node.GeoManualOverride { applyGeoInfoFromIP(node, node.IP) } @@ -114,6 +133,17 @@ func UpdateNode(id uint, input NodeInput) (*NodeView, error) { node.GeoLongitude = geoLongitude node.GeoManualOverride = geoManualOverride node.AutoUpdateEnabled = input.AutoUpdateEnabled + if node.NodeType == "tunnel_relay" { + node.RelayAgentAccessAddr = strings.TrimSpace(input.RelayAgentAccessAddr) + node.RelayClientAccessAddr = strings.TrimSpace(input.RelayClientAccessAddr) + node.RelayClientProxyURL = strings.TrimSpace(input.RelayClientProxyURL) + if input.RelayBindPort > 0 { + node.RelayBindPort = input.RelayBindPort + } + if input.RelayVhostHTTPPort > 0 { + node.RelayVhostHTTPPort = input.RelayVhostHTTPPort + } + } if !node.GeoManualOverride { applyGeoInfoFromIP(node, strings.TrimSpace(node.IP)) } @@ -313,6 +343,18 @@ func buildNodeView(node *model.Node) *NodeView { if view.UpdateChannel == "" { view.UpdateChannel = ReleaseChannelStable.String() } + view.NodeType = node.NodeType + if view.NodeType == "" { + view.NodeType = "edge_node" + } + view.RelayBindPort = node.RelayBindPort + view.RelayVhostHTTPPort = node.RelayVhostHTTPPort + view.RelayAgentAccessAddr = node.RelayAgentAccessAddr + view.RelayClientAccessAddr = node.RelayClientAccessAddr + view.RelayClientProxyURL = node.RelayClientProxyURL + view.RelayStatus = node.RelayStatus + view.RelayFrpVersion = node.RelayFrpVersion + view.RelayVersion = node.RelayVersion return view } @@ -587,3 +629,19 @@ func newServerNodeID() (string, error) { } return "node-" + token, nil } + +func normalizeNodeType(raw string) string { + switch strings.ToLower(strings.TrimSpace(raw)) { + case "tunnel_relay": + return "tunnel_relay" + default: + return "edge_node" + } +} + +func normalizeRelayPort(port int, defaultPort int) int { + if port <= 0 || port > 65535 { + return defaultPort + } + return port +} diff --git a/openflare_server/service/proxy_route.go b/openflare_server/service/proxy_route.go index b7dffa01..8f7abf7f 100644 --- a/openflare_server/service/proxy_route.go +++ b/openflare_server/service/proxy_route.go @@ -31,73 +31,81 @@ type ProxyRouteCustomHeaderInput struct { } type ProxyRouteInput struct { - SiteName string `json:"site_name"` - Domain string `json:"domain"` - Domains []string `json:"domains"` - OriginID *uint `json:"origin_id"` - OriginURL string `json:"origin_url"` - OriginScheme string `json:"origin_scheme"` - OriginAddress string `json:"origin_address"` - OriginPort string `json:"origin_port"` - OriginURI string `json:"origin_uri"` - OriginHost string `json:"origin_host"` - Upstreams []string `json:"upstreams"` - Enabled bool `json:"enabled"` - EnableHTTPS bool `json:"enable_https"` - CertID *uint `json:"cert_id"` - CertIDs []uint `json:"cert_ids"` - DomainCertIDs []uint `json:"domain_cert_ids"` - RedirectHTTP bool `json:"redirect_http"` - LimitConnPerServer int `json:"limit_conn_per_server"` - LimitConnPerIP int `json:"limit_conn_per_ip"` - LimitRate string `json:"limit_rate"` - CacheEnabled bool `json:"cache_enabled"` - CachePolicy string `json:"cache_policy"` - CacheRules []string `json:"cache_rules"` - CustomHeaders []ProxyRouteCustomHeaderInput `json:"custom_headers"` - PoWEnabled bool `json:"pow_enabled"` - PoWConfig string `json:"pow_config"` - BasicAuthEnabled bool `json:"basic_auth_enabled"` - BasicAuthUsername string `json:"basic_auth_username"` - BasicAuthPassword string `json:"basic_auth_password"` - Remark string `json:"remark"` + SiteName string `json:"site_name"` + Domain string `json:"domain"` + Domains []string `json:"domains"` + OriginID *uint `json:"origin_id"` + OriginURL string `json:"origin_url"` + OriginScheme string `json:"origin_scheme"` + OriginAddress string `json:"origin_address"` + OriginPort string `json:"origin_port"` + OriginURI string `json:"origin_uri"` + OriginHost string `json:"origin_host"` + Upstreams []string `json:"upstreams"` + Enabled bool `json:"enabled"` + EnableHTTPS bool `json:"enable_https"` + CertID *uint `json:"cert_id"` + CertIDs []uint `json:"cert_ids"` + DomainCertIDs []uint `json:"domain_cert_ids"` + RedirectHTTP bool `json:"redirect_http"` + LimitConnPerServer int `json:"limit_conn_per_server"` + LimitConnPerIP int `json:"limit_conn_per_ip"` + LimitRate string `json:"limit_rate"` + CacheEnabled bool `json:"cache_enabled"` + CachePolicy string `json:"cache_policy"` + CacheRules []string `json:"cache_rules"` + CustomHeaders []ProxyRouteCustomHeaderInput `json:"custom_headers"` + PoWEnabled bool `json:"pow_enabled"` + PoWConfig string `json:"pow_config"` + BasicAuthEnabled bool `json:"basic_auth_enabled"` + BasicAuthUsername string `json:"basic_auth_username"` + BasicAuthPassword string `json:"basic_auth_password"` + Remark string `json:"remark"` + UpstreamType string `json:"upstream_type"` + TunnelID *uint `json:"tunnel_id"` + TunnelTargetAddr string `json:"tunnel_target_addr"` + TunnelTargetProtocol string `json:"tunnel_target_protocol"` } type ProxyRouteView struct { - ID uint `json:"id"` - SiteName string `json:"site_name"` - Domain string `json:"domain"` - Domains []string `json:"domains"` - PrimaryDomain string `json:"primary_domain"` - DomainCount int `json:"domain_count"` - OriginID *uint `json:"origin_id"` - OriginURL string `json:"origin_url"` - OriginHost string `json:"origin_host"` - Upstreams string `json:"upstreams"` - UpstreamList []string `json:"upstream_list"` - Enabled bool `json:"enabled"` - EnableHTTPS bool `json:"enable_https"` - CertID *uint `json:"cert_id"` - CertIDs []uint `json:"cert_ids"` - DomainCertIDs []uint `json:"domain_cert_ids"` - RedirectHTTP bool `json:"redirect_http"` - LimitConnPerServer int `json:"limit_conn_per_server"` - LimitConnPerIP int `json:"limit_conn_per_ip"` - LimitRate string `json:"limit_rate"` - CacheEnabled bool `json:"cache_enabled"` - CachePolicy string `json:"cache_policy"` - CacheRules string `json:"cache_rules"` - CacheRuleList []string `json:"cache_rule_list"` - CustomHeaders string `json:"custom_headers"` - CustomHeaderList []ProxyRouteCustomHeaderInput `json:"custom_header_list"` - PoWEnabled bool `json:"pow_enabled"` - PoWConfig *ProxyRoutePoWConfig `json:"pow_config"` - BasicAuthEnabled bool `json:"basic_auth_enabled"` - BasicAuthUsername string `json:"basic_auth_username"` - BasicAuthPassword string `json:"basic_auth_password"` - Remark string `json:"remark"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint `json:"id"` + SiteName string `json:"site_name"` + Domain string `json:"domain"` + Domains []string `json:"domains"` + PrimaryDomain string `json:"primary_domain"` + DomainCount int `json:"domain_count"` + OriginID *uint `json:"origin_id"` + OriginURL string `json:"origin_url"` + OriginHost string `json:"origin_host"` + Upstreams string `json:"upstreams"` + UpstreamList []string `json:"upstream_list"` + Enabled bool `json:"enabled"` + EnableHTTPS bool `json:"enable_https"` + CertID *uint `json:"cert_id"` + CertIDs []uint `json:"cert_ids"` + DomainCertIDs []uint `json:"domain_cert_ids"` + RedirectHTTP bool `json:"redirect_http"` + LimitConnPerServer int `json:"limit_conn_per_server"` + LimitConnPerIP int `json:"limit_conn_per_ip"` + LimitRate string `json:"limit_rate"` + CacheEnabled bool `json:"cache_enabled"` + CachePolicy string `json:"cache_policy"` + CacheRules string `json:"cache_rules"` + CacheRuleList []string `json:"cache_rule_list"` + CustomHeaders string `json:"custom_headers"` + CustomHeaderList []ProxyRouteCustomHeaderInput `json:"custom_header_list"` + PoWEnabled bool `json:"pow_enabled"` + PoWConfig *ProxyRoutePoWConfig `json:"pow_config"` + BasicAuthEnabled bool `json:"basic_auth_enabled"` + BasicAuthUsername string `json:"basic_auth_username"` + BasicAuthPassword string `json:"basic_auth_password"` + Remark string `json:"remark"` + UpstreamType string `json:"upstream_type"` + TunnelID *uint `json:"tunnel_id"` + TunnelTargetAddr string `json:"tunnel_target_addr"` + TunnelTargetProtocol string `json:"tunnel_target_protocol"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } func ListProxyRoutes() ([]*ProxyRouteView, error) { @@ -164,16 +172,27 @@ func buildProxyRoute(route *model.ProxyRoute, input ProxyRouteInput) (*model.Pro domain := domains[0] siteName := normalizeProxyRouteSiteNameInput(route, input.SiteName, domain) - originURL, originID, err := resolveProxyRoutePrimaryOrigin(input) - if err != nil { - return nil, err + upstreamType := normalizeUpstreamType(input.UpstreamType) + var originURL string + var originID *uint + var upstreams []string + + if upstreamType == "tunnel" { + // Tunnel type: origin URL is auto-filled during config rendering + originURL = "http://127.0.0.1" + upstreams = []string{originURL} + } else { + originURL, originID, err = resolveProxyRoutePrimaryOrigin(input) + if err != nil { + return nil, err + } + upstreams, err = normalizeUpstreams(originURL, input.Upstreams) + if err != nil { + return nil, err + } } originHost := strings.TrimSpace(input.OriginHost) remark := strings.TrimSpace(input.Remark) - upstreams, err := normalizeUpstreams(originURL, input.Upstreams) - if err != nil { - return nil, err - } cachePolicy := strings.TrimSpace(input.CachePolicy) cacheRules, err := normalizeCacheRules(input.CacheEnabled, cachePolicy, input.CacheRules) if err != nil { @@ -306,6 +325,16 @@ func buildProxyRoute(route *model.ProxyRoute, input ProxyRouteInput) (*model.Pro route.BasicAuthUsername = input.BasicAuthUsername route.BasicAuthPassword = input.BasicAuthPassword route.Remark = remark + route.UpstreamType = upstreamType + if upstreamType == "tunnel" { + route.TunnelID = input.TunnelID + route.TunnelTargetAddr = strings.TrimSpace(input.TunnelTargetAddr) + route.TunnelTargetProtocol = normalizeTunnelTargetProtocol(input.TunnelTargetProtocol) + } else { + route.TunnelID = nil + route.TunnelTargetAddr = "" + route.TunnelTargetProtocol = "" + } return route, nil } @@ -359,40 +388,44 @@ func buildProxyRouteView(route *model.ProxyRoute) (*ProxyRouteView, error) { } primaryDomain := domains[0] return &ProxyRouteView{ - ID: route.ID, - SiteName: normalizeProxyRouteSiteNameInput(route, route.SiteName, primaryDomain), - Domain: primaryDomain, - Domains: domains, - PrimaryDomain: primaryDomain, - DomainCount: len(domains), - OriginID: route.OriginID, - OriginURL: route.OriginURL, - OriginHost: route.OriginHost, - Upstreams: route.Upstreams, - UpstreamList: upstreams, - Enabled: route.Enabled, - EnableHTTPS: route.EnableHTTPS, - CertID: certID, - CertIDs: certIDs, - DomainCertIDs: domainCertIDs, - RedirectHTTP: route.RedirectHTTP, - LimitConnPerServer: route.LimitConnPerServer, - LimitConnPerIP: route.LimitConnPerIP, - LimitRate: route.LimitRate, - CacheEnabled: route.CacheEnabled, - CachePolicy: route.CachePolicy, - CacheRules: route.CacheRules, - CacheRuleList: cacheRules, - CustomHeaders: route.CustomHeaders, - CustomHeaderList: customHeaders, - PoWEnabled: route.PoWEnabled, - PoWConfig: powConfig, - BasicAuthEnabled: route.BasicAuthEnabled, - BasicAuthUsername: route.BasicAuthUsername, - BasicAuthPassword: route.BasicAuthPassword, - Remark: route.Remark, - CreatedAt: route.CreatedAt, - UpdatedAt: route.UpdatedAt, + ID: route.ID, + SiteName: normalizeProxyRouteSiteNameInput(route, route.SiteName, primaryDomain), + Domain: primaryDomain, + Domains: domains, + PrimaryDomain: primaryDomain, + DomainCount: len(domains), + OriginID: route.OriginID, + OriginURL: route.OriginURL, + OriginHost: route.OriginHost, + Upstreams: route.Upstreams, + UpstreamList: upstreams, + Enabled: route.Enabled, + EnableHTTPS: route.EnableHTTPS, + CertID: certID, + CertIDs: certIDs, + DomainCertIDs: domainCertIDs, + RedirectHTTP: route.RedirectHTTP, + LimitConnPerServer: route.LimitConnPerServer, + LimitConnPerIP: route.LimitConnPerIP, + LimitRate: route.LimitRate, + CacheEnabled: route.CacheEnabled, + CachePolicy: route.CachePolicy, + CacheRules: route.CacheRules, + CacheRuleList: cacheRules, + CustomHeaders: route.CustomHeaders, + CustomHeaderList: customHeaders, + PoWEnabled: route.PoWEnabled, + PoWConfig: powConfig, + BasicAuthEnabled: route.BasicAuthEnabled, + BasicAuthUsername: route.BasicAuthUsername, + BasicAuthPassword: route.BasicAuthPassword, + Remark: route.Remark, + UpstreamType: route.UpstreamType, + TunnelID: route.TunnelID, + TunnelTargetAddr: route.TunnelTargetAddr, + TunnelTargetProtocol: route.TunnelTargetProtocol, + CreatedAt: route.CreatedAt, + UpdatedAt: route.UpdatedAt, }, nil } @@ -1208,3 +1241,21 @@ func decodeStoredPoWConfig(enabled bool, raw string) (*ProxyRoutePoWConfig, erro } return &cfg, nil } + +func normalizeUpstreamType(raw string) string { + switch strings.ToLower(strings.TrimSpace(raw)) { + case "tunnel": + return "tunnel" + default: + return "direct" + } +} + +func normalizeTunnelTargetProtocol(raw string) string { + switch strings.ToLower(strings.TrimSpace(raw)) { + case "https": + return "https" + default: + return "http" + } +} diff --git a/openflare_server/service/relay.go b/openflare_server/service/relay.go new file mode 100644 index 00000000..f1d77987 --- /dev/null +++ b/openflare_server/service/relay.go @@ -0,0 +1,294 @@ +package service + +import ( + "encoding/json" + "fmt" + "log/slog" + "openflare/common" + "openflare/model" + "strings" + "time" +) + +// RelayHeartbeatPayload is the payload sent by OpenFlareRelay in each heartbeat. +type RelayHeartbeatPayload struct { + RelayVersion string `json:"relay_version"` + FrpVersion string `json:"frp_version"` + RelayStatus string `json:"relay_status"` + FrpsConnCount int `json:"frps_connections"` + FrpsProxyCount int `json:"frps_proxy_count"` +} + +// RelayConfig is the frps configuration sent to the Relay. +type RelayConfig struct { + BindPort int `json:"bind_port"` + VhostHTTPPort int `json:"vhost_http_port"` + AuthToken string `json:"auth_token"` + LogLevel string `json:"log_level"` +} + +// RelaySettings contains runtime settings for the Relay. +type RelaySettings struct { + HeartbeatInterval int `json:"heartbeat_interval"` + WebsocketUpgradeEnabled bool `json:"websocket_upgrade_enabled"` +} + +// RelayHeartbeatResponse is the response returned to the Relay from a heartbeat. +type RelayHeartbeatResponse struct { + RelayConfig *RelayConfig `json:"relay_config"` + RelaySettings *RelaySettings `json:"relay_settings"` +} + +// HeartbeatRelay processes a relay heartbeat, updates node status, and returns config. +func HeartbeatRelay(node *model.Node, payload RelayHeartbeatPayload) (*RelayHeartbeatResponse, error) { + if node == nil { + return nil, fmt.Errorf("relay node is nil") + } + slog.Debug("relay heartbeat received", "node_id", node.NodeID) + + payload.RelayVersion = strings.TrimSpace(payload.RelayVersion) + payload.FrpVersion = strings.TrimSpace(payload.FrpVersion) + payload.RelayStatus = normalizeRelayStatus(payload.RelayStatus) + + changes := make(map[string]any) + appendRelayChange := func(key string, before any, after any) { + if before != after { + changes[key] = after + } + } + now := time.Now() + appendRelayChange("relay_version", node.RelayVersion, payload.RelayVersion) + appendRelayChange("relay_frp_version", node.RelayFrpVersion, payload.FrpVersion) + appendRelayChange("relay_status", node.RelayStatus, payload.RelayStatus) + if !node.LastSeenAt.Equal(now) { + changes["last_seen_at"] = now + } + changes["status"] = NodeStatusOnline + + node.RelayVersion = payload.RelayVersion + node.RelayFrpVersion = payload.FrpVersion + node.RelayStatus = payload.RelayStatus + node.LastSeenAt = now + node.Status = NodeStatusOnline + + if len(changes) > 0 { + if err := model.DB.Model(node).Updates(changes).Error; err != nil { + return nil, fmt.Errorf("update relay heartbeat: %w", err) + } + } + refreshAgentTokenCache(node) + + return &RelayHeartbeatResponse{ + RelayConfig: buildRelayConfig(node), + RelaySettings: buildRelaySettings(), + }, nil +} + +func buildRelayConfig(node *model.Node) *RelayConfig { + if node == nil { + return nil + } + return &RelayConfig{ + BindPort: node.RelayBindPort, + VhostHTTPPort: node.RelayVhostHTTPPort, + AuthToken: node.RelayAuthToken, + LogLevel: "info", + } +} + +func buildRelaySettings() *RelaySettings { + return &RelaySettings{ + HeartbeatInterval: common.AgentHeartbeatInterval, + WebsocketUpgradeEnabled: common.AgentWebsocketUpgradeEnabled, + } +} + +func normalizeRelayStatus(status string) string { + switch strings.ToLower(strings.TrimSpace(status)) { + case "healthy": + return "healthy" + case "unhealthy": + return "unhealthy" + default: + return "unknown" + } +} + +// FlaredHeartbeatPayload is the payload sent by OpenFlared in each heartbeat. +type FlaredHeartbeatPayload struct { + ClientVersion string `json:"client_version"` + FrpVersion string `json:"frp_version"` + TunnelStatus string `json:"tunnel_status"` + ConnectedRelays []FlaredConnectedRelay `json:"connected_relays"` + CurrentVersion string `json:"current_version"` + CurrentChecksum string `json:"current_checksum"` +} + +// FlaredConnectedRelay describes the status of a relay connection from a client. +type FlaredConnectedRelay struct { + RelayNodeID string `json:"relay_node_id"` + Status string `json:"status"` + ProxyCount int `json:"proxy_count"` +} + +// FlaredHeartbeatResponse is the response returned to the OpenFlared client. +type FlaredHeartbeatResponse struct { + ActiveConfig *ActiveConfigMeta `json:"active_config"` + TunnelSettings *RelaySettings `json:"tunnel_settings"` +} + +// FlaredTunnelConfigResponse is the full tunnel routing config sent to the client. +type FlaredTunnelConfigResponse struct { + Version string `json:"version"` + Checksum string `json:"checksum"` + Relays []FlaredRelayInfo `json:"relays"` + Proxies []FlaredProxyEntry `json:"proxies"` +} + +// FlaredRelayInfo describes a relay that the client should connect to. +type FlaredRelayInfo struct { + RelayNodeID string `json:"relay_node_id"` + Address string `json:"address"` + AuthToken string `json:"auth_token"` + ProxyURL string `json:"proxy_url"` +} + +// FlaredProxyEntry describes a single frpc proxy definition. +type FlaredProxyEntry struct { + Name string `json:"name"` + Type string `json:"type"` + LocalAddr string `json:"local_addr"` + LocalPort int `json:"local_port"` + CustomDomains []string `json:"custom_domains"` +} + +// HeartbeatFlared processes an OpenFlared client heartbeat. +func HeartbeatFlared(tunnel *model.Tunnel, payload FlaredHeartbeatPayload) (*FlaredHeartbeatResponse, error) { + if tunnel == nil { + return nil, fmt.Errorf("tunnel is nil") + } + slog.Debug("flared heartbeat received", "tunnel_id", tunnel.TunnelID) + + now := time.Now() + tunnel.ClientVersion = strings.TrimSpace(payload.ClientVersion) + tunnel.FrpVersion = strings.TrimSpace(payload.FrpVersion) + tunnel.Status = "online" + tunnel.LastSeenAt = now + tunnel.CurrentVersion = strings.TrimSpace(payload.CurrentVersion) + tunnel.CurrentChecksum = strings.TrimSpace(payload.CurrentChecksum) + + relaysJSON, err := json.Marshal(payload.ConnectedRelays) + if err == nil { + tunnel.ConnectedRelays = string(relaysJSON) + } + + if err := tunnel.Update(); err != nil { + return nil, fmt.Errorf("update tunnel heartbeat: %w", err) + } + + activeConfig, _ := GetActiveConfigMetaForAgent() + + return &FlaredHeartbeatResponse{ + ActiveConfig: activeConfig, + TunnelSettings: buildRelaySettings(), + }, nil +} + +// GetFlaredTunnelConfig builds the full tunnel routing config for an OpenFlared client. +func GetFlaredTunnelConfig(tunnel *model.Tunnel) (*FlaredTunnelConfigResponse, error) { + if tunnel == nil { + return nil, fmt.Errorf("tunnel is nil") + } + + activeVersion, err := model.GetActiveConfigVersion() + if err != nil { + return nil, fmt.Errorf("no active config version: %w", err) + } + + // Get all enabled proxy routes with tunnel upstream targeting this tunnel + routes, err := model.GetEnabledProxyRoutes() + if err != nil { + return nil, fmt.Errorf("get proxy routes: %w", err) + } + + // Get all online tunnel relay nodes + relayNodes, err := model.ListNodesByType("tunnel_relay") + if err != nil { + return nil, fmt.Errorf("get relay nodes: %w", err) + } + + // Build relay info + relays := make([]FlaredRelayInfo, 0, len(relayNodes)) + for _, node := range relayNodes { + if node.RelayStatus == "healthy" || node.Status == NodeStatusOnline { + addr := strings.TrimSpace(node.RelayClientAccessAddr) + if addr == "" { + addr = fmt.Sprintf("%s:%d", strings.TrimSpace(node.IP), node.RelayBindPort) + } + relays = append(relays, FlaredRelayInfo{ + RelayNodeID: node.NodeID, + Address: addr, + AuthToken: node.RelayAuthToken, + ProxyURL: strings.TrimSpace(node.RelayClientProxyURL), + }) + } + } + + // Build proxy entries from routes + proxies := make([]FlaredProxyEntry, 0) + for _, route := range routes { + if route.UpstreamType != "tunnel" || route.TunnelID == nil || *route.TunnelID != tunnel.ID { + continue + } + if !route.Enabled { + continue + } + domains, err := decodeStoredDomains(route.Domains, route.Domain) + if err != nil { + continue + } + localAddr, localPort := parseTunnelTargetAddr(route.TunnelTargetAddr) + proxies = append(proxies, FlaredProxyEntry{ + Name: fmt.Sprintf("%s-%s", tunnel.TunnelID, sanitizeProxyName(domains[0])), + Type: "http", + LocalAddr: localAddr, + LocalPort: localPort, + CustomDomains: domains, + }) + } + + return &FlaredTunnelConfigResponse{ + Version: activeVersion.Version, + Checksum: activeVersion.Checksum, + Relays: relays, + Proxies: proxies, + }, nil +} + +func parseTunnelTargetAddr(addr string) (string, int) { + addr = strings.TrimSpace(addr) + if addr == "" { + return "127.0.0.1", 80 + } + host, portStr, err := splitHostPort(addr) + if err != nil { + return addr, 80 + } + port := 80 + if _, err := fmt.Sscanf(portStr, "%d", &port); err != nil { + port = 80 + } + return host, port +} + +func splitHostPort(addr string) (string, string, error) { + lastColon := strings.LastIndex(addr, ":") + if lastColon < 0 { + return addr, "", fmt.Errorf("no port") + } + return addr[:lastColon], addr[lastColon+1:], nil +} + +func sanitizeProxyName(domain string) string { + return strings.ReplaceAll(strings.ReplaceAll(domain, ".", "-"), "*", "wildcard") +} diff --git a/openflare_server/service/relay_ws.go b/openflare_server/service/relay_ws.go new file mode 100644 index 00000000..38fc1e5e --- /dev/null +++ b/openflare_server/service/relay_ws.go @@ -0,0 +1,37 @@ +package service + +var DefaultRelayWSHub = NewWSHub("relay") + +func RegisterRelayWSClient(nodeID string) *WSClient { + return DefaultRelayWSHub.Register(nodeID) +} + +func UnregisterRelayWSClient(client *WSClient) { + DefaultRelayWSHub.Unregister(client) +} + +func IsRelayWSConnected(nodeID string) bool { + return DefaultRelayWSHub.IsConnected(nodeID) +} + +func SendRelayWSPing(nodeID string) bool { + return DefaultRelayWSHub.SendMessage(nodeID, WSMessage{ + Type: "ping", + }) +} + +func SendRelayWSPong(nodeID string) bool { + return DefaultRelayWSHub.SendMessage(nodeID, WSMessage{ + Type: "pong", + }) +} + +func SendRelayWSConfig(nodeID string, config *RelayConfig) bool { + if config == nil { + return false + } + return DefaultRelayWSHub.SendMessage(nodeID, WSMessage{ + Type: "relay_config", + Payload: config, + }) +} diff --git a/openflare_server/service/tunnel.go b/openflare_server/service/tunnel.go new file mode 100644 index 00000000..06389168 --- /dev/null +++ b/openflare_server/service/tunnel.go @@ -0,0 +1,183 @@ +package service + +import ( + "encoding/json" + "errors" + "log/slog" + "openflare/model" + "strings" + "time" +) + +type TunnelInput struct { + Name string `json:"name"` + Remark string `json:"remark"` +} + +type TunnelView struct { + ID uint `json:"id"` + TunnelID string `json:"tunnel_id"` + Name string `json:"name"` + TunnelToken string `json:"tunnel_token"` + Status string `json:"status"` + ClientVersion string `json:"client_version"` + FrpVersion string `json:"frp_version"` + LastSeenAt time.Time `json:"last_seen_at"` + LastError string `json:"last_error"` + CurrentVersion string `json:"current_version"` + CurrentChecksum string `json:"current_checksum"` + ConnectedRelays []string `json:"connected_relays"` + Remark string `json:"remark"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +func ListTunnels() ([]*TunnelView, error) { + tunnels, err := model.ListTunnels() + if err != nil { + return nil, err + } + views := make([]*TunnelView, 0, len(tunnels)) + for _, tunnel := range tunnels { + views = append(views, buildTunnelView(tunnel)) + } + return views, nil +} + +func GetTunnel(id uint) (*TunnelView, error) { + tunnel, err := model.GetTunnelByID(id) + if err != nil { + return nil, err + } + return buildTunnelView(tunnel), nil +} + +func CreateTunnel(input TunnelInput) (*TunnelView, error) { + name := strings.TrimSpace(input.Name) + if name == "" { + return nil, errors.New("Tunnel 名称不能为空") + } + tunnelID, err := newTunnelID() + if err != nil { + return nil, err + } + tunnelToken, err := newRandomToken() + if err != nil { + return nil, err + } + tunnel := &model.Tunnel{ + TunnelID: tunnelID, + Name: name, + TunnelToken: tunnelToken, + Status: "offline", + Remark: strings.TrimSpace(input.Remark), + } + if err := tunnel.Insert(); err != nil { + if model.IsUniqueConstraintError(err) { + return nil, errors.New("Tunnel 标识生成冲突，请重试") + } + return nil, err + } + slog.Info("tunnel created", "name", tunnel.Name, "tunnel_id", tunnel.TunnelID) + return buildTunnelView(tunnel), nil +} + +func UpdateTunnel(id uint, input TunnelInput) (*TunnelView, error) { + tunnel, err := model.GetTunnelByID(id) + if err != nil { + return nil, err + } + name := strings.TrimSpace(input.Name) + if name == "" { + return nil, errors.New("Tunnel 名称不能为空") + } + tunnel.Name = name + tunnel.Remark = strings.TrimSpace(input.Remark) + if err := tunnel.Update(); err != nil { + return nil, err + } + slog.Info("tunnel updated", "name", tunnel.Name, "tunnel_id", tunnel.TunnelID) + return buildTunnelView(tunnel), nil +} + +func DeleteTunnel(id uint) error { + tunnel, err := model.GetTunnelByID(id) + if err != nil { + return err + } + slog.Info("tunnel deleted", "name", tunnel.Name, "tunnel_id", tunnel.TunnelID) + return tunnel.Delete() +} + +func RotateTunnelToken(id uint) (*TunnelView, error) { + tunnel, err := model.GetTunnelByID(id) + if err != nil { + return nil, err + } + newToken, err := newRandomToken() + if err != nil { + return nil, err + } + tunnel.TunnelToken = newToken + if err := tunnel.Update(); err != nil { + return nil, err + } + slog.Info("tunnel token rotated", "tunnel_id", tunnel.TunnelID) + return buildTunnelView(tunnel), nil +} + +func AuthenticateTunnelToken(token string) (*model.Tunnel, error) { + token = strings.TrimSpace(token) + if token == "" { + return nil, errors.New("缺少 Tunnel Token") + } + tunnel, err := model.GetTunnelByTunnelToken(token) + if err != nil { + return nil, errors.New("Tunnel Token 无效") + } + return tunnel, nil +} + +func buildTunnelView(tunnel *model.Tunnel) *TunnelView { + if tunnel == nil { + return nil + } + relays := decodeTunnelConnectedRelays(tunnel.ConnectedRelays) + return &TunnelView{ + ID: tunnel.ID, + TunnelID: tunnel.TunnelID, + Name: tunnel.Name, + TunnelToken: tunnel.TunnelToken, + Status: tunnel.Status, + ClientVersion: tunnel.ClientVersion, + FrpVersion: tunnel.FrpVersion, + LastSeenAt: tunnel.LastSeenAt, + LastError: tunnel.LastError, + CurrentVersion: tunnel.CurrentVersion, + CurrentChecksum: tunnel.CurrentChecksum, + ConnectedRelays: relays, + Remark: tunnel.Remark, + CreatedAt: tunnel.CreatedAt, + UpdatedAt: tunnel.UpdatedAt, + } +} + +func decodeTunnelConnectedRelays(raw string) []string { + raw = strings.TrimSpace(raw) + if raw == "" || raw == "[]" { + return make([]string, 0) + } + var relays []string + if err := json.Unmarshal([]byte(raw), &relays); err != nil { + return make([]string, 0) + } + return relays +} + +func newTunnelID() (string, error) { + token, err := newRandomToken() + if err != nil { + return "", err + } + return "tun-" + token, nil +} diff --git a/openflare_server/service/ws_hub.go b/openflare_server/service/ws_hub.go new file mode 100644 index 00000000..8da8ba18 --- /dev/null +++ b/openflare_server/service/ws_hub.go @@ -0,0 +1,177 @@ +package service + +import ( + "log/slog" + "sync" +) + +type WSMessage struct { + Type string `json:"type"` + Payload any `json:"payload,omitempty"` +} + +type WSClient struct { + id string + send chan WSMessage + done chan struct{} + once sync.Once +} + +func (client *WSClient) ID() string { + if client == nil { + return "" + } + return client.id +} + +func (client *WSClient) Messages() <-chan WSMessage { + if client == nil { + return nil + } + return client.send +} + +func (client *WSClient) Done() <-chan struct{} { + if client == nil { + return nil + } + return client.done +} + +func (client *WSClient) Send(message WSMessage) bool { + if client == nil { + return false + } + select { + case <-client.done: + return false + case client.send <- message: + return true + default: + return false + } +} + +func (client *WSClient) Close() { + if client == nil { + return + } + client.once.Do(func() { + close(client.done) + }) +} + +type WSHub struct { + name string + mu sync.RWMutex + clients map[string]*WSClient +} + +func NewWSHub(name string) *WSHub { + return &WSHub{ + name: name, + clients: make(map[string]*WSClient), + } +} + +func (h *WSHub) Register(id string) *WSClient { + client := &WSClient{ + id: id, + send: make(chan WSMessage, 16), + done: make(chan struct{}), + } + h.mu.Lock() + if existing := h.clients[id]; existing != nil { + slog.Debug("ws replacing existing connection", "hub", h.name, "id", id) + existing.Close() + } + h.clients[id] = client + count := len(h.clients) + h.mu.Unlock() + slog.Debug("ws connection registered", "hub", h.name, "id", id, "client_count", count) + return client +} + +func (h *WSHub) Unregister(client *WSClient) { + if client == nil { + return + } + h.mu.Lock() + if current := h.clients[client.id]; current == client { + delete(h.clients, client.id) + } + count := len(h.clients) + h.mu.Unlock() + client.Close() + slog.Debug("ws connection unregistered", "hub", h.name, "id", client.id, "client_count", count) +} + +func (h *WSHub) Disconnect(id string) { + h.mu.Lock() + client := h.clients[id] + if client != nil { + delete(h.clients, id) + } + count := len(h.clients) + h.mu.Unlock() + + if client != nil { + client.Close() + slog.Debug("ws connection forcefully disconnected", "hub", h.name, "id", id, "client_count", count) + } +} + +func (h *WSHub) IsConnected(id string) bool { + h.mu.RLock() + client := h.clients[id] + h.mu.RUnlock() + if client == nil { + return false + } + select { + case <-client.done: + return false + default: + return true + } +} + +func (h *WSHub) SendMessage(id string, message WSMessage) bool { + h.mu.RLock() + client := h.clients[id] + h.mu.RUnlock() + if client == nil { + return false + } + ok := client.Send(message) + if !ok { + slog.Debug("ws send queued message failed", "hub", h.name, "id", id, "type", message.Type) + } + return ok +} + +type WSBroadcastResult struct { + ClientCount int `json:"client_count"` + SuccessCount int `json:"success_count"` + FailedIDs []string `json:"failed_ids"` +} + +func (h *WSHub) Broadcast(message WSMessage) WSBroadcastResult { + h.mu.RLock() + clients := make([]*WSClient, 0, len(h.clients)) + for _, client := range h.clients { + clients = append(clients, client) + } + h.mu.RUnlock() + + var result WSBroadcastResult + result.ClientCount = len(clients) + for _, client := range clients { + if client.Send(message) { + result.SuccessCount++ + continue + } + result.FailedIDs = append(result.FailedIDs, client.ID()) + } + return result +}