+
+
+
@@ -204,6 +212,8 @@ make server # 产物：dist/portrelay-server-linux-{amd64,arm64} 2. 访问方式保持默认的 **仅授权链接**，然后点 **邀请**，选择授权后可访问时长、可用设备数，以及是否需要审批。 3. 把生成的文案或二维码发给同事。同事打开链接、填写姓名后，就可以用数据库客户端或 SSH 连接 `192.168.1.10:33660`。 +需要一次开放多台主机时（比如一组服务器），点 **新建通道组**，主机填 `10.0.1.11-13`、端口填 `22`，预览确认端口映射后创建，再点 **邀请整组**。已有的通道也可以在新建组时勾选加入。 + ## 配置参考 **网关 `portrelay-server`** diff --git a/README_EN.md b/README_EN.md index ca0b026..fdc9da3 100644 --- a/README_EN.md +++ b/README_EN.md @@ -58,11 +58,13 @@ PortRelay turns that into three steps: - When approval is required, the app shows a system notification and you approve or deny with one click. The requester's page refreshes on its own. - **The access clock starts at the first connection**, so a teammate can accept an invite now and use it later. - When access expires or is revoked, **existing connections are dropped too**. Unauthorized connections are reset (TCP RST) and logged. +- **Tunnel groups** open several hosts and ports at once (for example `10.0.1.11-13` × `22, 8080`) on a contiguous block of gateway ports. A group shares one invite link and one set of grants: a teammate is authorized once for the whole group, and one revoke disconnects them everywhere. **Tunnel lifecycle** - **Idle auto-close.** The idle timer starts when a tunnel has no connections, no active grants, no pending requests and no invite links with uses left. By default the tunnel closes after 30 minutes of idle time. - An optional **maximum open time** closes the tunnel at a fixed point, whether or not it is in use. +- While any address in a group is in use, the whole group counts as in use; members are switched on and off together. - PortRelay records the first-connection time, the actual connected time and why each tunnel was closed. - Editing a tunnel keeps existing connections. Moving a tunnel binds the new port before releasing the old one. @@ -104,6 +106,12 @@ PortRelay turns that into three steps: +**Tunnel groups.** A group card sums up the state and traffic of the whole group and lists every access address. Each address can be switched, invited or moved out on its own. + +
+
+
@@ -213,6 +221,8 @@ Open ⚙️ Settings and fill in: 2. Keep the default access mode, **仅授权链接** (invite-only). Click **邀请** (Invite), then choose how long access lasts, how many devices may use the link, and whether approval is required. 3. Send the generated message or QR code. Once the teammate has entered their name, they can connect their database client or SSH to `192.168.1.10:33660`. +To open several hosts at once (a set of servers, say), click **新建通道组** (New tunnel group), enter hosts `10.0.1.11-13` and ports `22`, check the previewed port mapping and create it, then click **邀请整组** (Invite group). Existing tunnels can be ticked to join the group as well. + ## Configuration **Gateway: `portrelay-server`** diff --git a/client/agent.go b/client/agent.go index 60e9b47..49d77af 100644 --- a/client/agent.go +++ b/client/agent.go @@ -166,10 +166,14 @@ type TunnelActivity struct { // inUse: 有在线连接、有效授权（含尚未首次连接的）、待审批申请或仍可使用的邀请 func tunnelInUse(m protocol.TunnelMetrics) bool { - if m.ActiveConns > 0 || len(m.DynamicLeases) > 0 || len(m.Pending) > 0 { + return m.ActiveConns > 0 || accessInUse(m.DynamicLeases, m.Pending, m.Invites) +} + +func accessInUse(leases []protocol.DynamicLease, pending []protocol.AccessRequest, invites []protocol.Invite) bool { + if len(leases) > 0 || len(pending) > 0 { return true } - for _, inv := range m.Invites { + for _, inv := range invites { if inv.MaxUses <= 0 || len(inv.Uses) < inv.MaxUses { return true } @@ -188,6 +192,18 @@ func (a *Agent) trackUsage(now time.Time) { for _, m := range metrics.Tunnels { byID[m.TunnelID] = m } + // 通道组整体在用：组内任一通道在用，或组的邀请 / 授权 / 申请仍有效。组内通道一起计闲置、一起关闭。 + groupBusy := make(map[string]bool) + for _, gm := range metrics.Groups { + if accessInUse(gm.DynamicLeases, gm.Pending, gm.Invites) { + groupBusy[gm.GroupID] = true + } + } + for _, t := range cfg.Tunnels { + if m, ok := byID[t.ID]; ok && t.Enabled && t.GroupID != "" && m.Status == "online" && tunnelInUse(m) { + groupBusy[t.GroupID] = true + } + } type closing struct { id, reason string @@ -226,7 +242,7 @@ func (a *Agent) trackUsage(now time.Time) { firstConn = append(firstConn, t.ID) } } - u.inUse = tunnelInUse(m) + u.inUse = tunnelInUse(m) || (t.GroupID != "" && groupBusy[t.GroupID]) if u.inUse { u.lastActive = now } @@ -292,6 +308,20 @@ func (a *Agent) RateSeries(id string) RateSeries { return a.rates.series(id, time.Now()) } +// GroupRateSeries sums the rate curves of several tunnels (a tunnel group) +func (a *Agent) GroupRateSeries(ids []string) RateSeries { + now := time.Now() + sum := (&rateHistory{}).series(now) + for _, id := range ids { + s := a.rates.series(id, now) + for k := range sum.Up { + sum.Up[k] += s.Up[k] + sum.Down[k] += s.Down[k] + } + } + return sum +} + // Activity returns the live usage of an enabled tunnel (nil when not tracked yet) func (a *Agent) Activity(t protocol.TunnelConfig) *TunnelActivity { a.usageMu.Lock() @@ -448,7 +478,7 @@ func (a *Agent) connectAndRun(cfg Config) error { for _, t := range a.cm.Get().Tunnels { if t.Enabled { log.Printf("[Agent] 注册通道: %s (:%d -> %s) [责任人: %s]", t.Name, t.ListenPort, t.Target, t.Owner) - _ = a.sendControl(protocol.RegisterTunnelMessage{Type: protocol.TypeRegisterTunnel, Tunnel: t}) + _ = a.sendControl(protocol.RegisterTunnelMessage{Type: protocol.TypeRegisterTunnel, Tunnel: a.wire(t)}) } } @@ -696,12 +726,23 @@ func (a *Agent) SyncTunnel(t protocol.TunnelConfig, edited bool) { return } if edited { - _ = a.sendControl(protocol.UpdateTunnelMessage{Type: protocol.TypeUpdateTunnel, Tunnel: t}) + _ = a.sendControl(protocol.UpdateTunnelMessage{Type: protocol.TypeUpdateTunnel, Tunnel: a.wire(t)}) } else { - _ = a.sendControl(protocol.RegisterTunnelMessage{Type: protocol.TypeRegisterTunnel, Tunnel: t}) + _ = a.sendControl(protocol.RegisterTunnelMessage{Type: protocol.TypeRegisterTunnel, Tunnel: a.wire(t)}) } } +// wire fills in what the relay server needs besides the stored config (the group's display name) +func (a *Agent) wire(t protocol.TunnelConfig) protocol.TunnelConfig { + t.GroupName = "" + if t.GroupID != "" { + if g, ok := a.cm.FindGroup(t.GroupID); ok { + t.GroupName = g.Name + } + } + return t +} + func (a *Agent) RemoveTunnel(id string) { a.mu.Lock() delete(a.regErrors, id) @@ -712,8 +753,14 @@ func (a *Agent) RemoveTunnel(id string) { _ = a.sendControl(protocol.UnregisterMessage{Type: protocol.TypeUnregister, TunnelID: id}) } +// Scope selects what an invite, request or lease applies to: one tunnel, or a whole group +type Scope struct { + TunnelID string + GroupID string +} + // CreateInvite asks the relay server for a new invite link and waits for the result -func (a *Agent) CreateInvite(tunnelID, note string, validMinutes, maxUses int, leaseDuration string, requireApproval bool) (protocol.Invite, error) { +func (a *Agent) CreateInvite(scope Scope, note string, validMinutes, maxUses int, leaseDuration string, requireApproval bool) (protocol.Invite, error) { if !a.IsConnected() { return protocol.Invite{}, fmt.Errorf("未连接中继服务器") } @@ -729,7 +776,7 @@ func (a *Agent) CreateInvite(tunnelID, note string, validMinutes, maxUses int, l }() if err := a.sendControl(protocol.CreateInviteMessage{ - Type: protocol.TypeCreateInvite, RequestID: reqID, TunnelID: tunnelID, + Type: protocol.TypeCreateInvite, RequestID: reqID, TunnelID: scope.TunnelID, GroupID: scope.GroupID, Note: note, ValidMinutes: validMinutes, MaxUses: maxUses, LeaseDuration: leaseDuration, RequireApproval: requireApproval, }); err != nil { return protocol.Invite{}, err @@ -746,27 +793,35 @@ func (a *Agent) CreateInvite(tunnelID, note string, validMinutes, maxUses int, l } // RevokeInvite invalidates an invite link and the leases it granted -func (a *Agent) RevokeInvite(tunnelID, code string) error { +func (a *Agent) RevokeInvite(scope Scope, code string) error { if !a.IsConnected() { return fmt.Errorf("未连接中继服务器") } - return a.sendControl(protocol.RevokeInviteMessage{Type: protocol.TypeRevokeInvite, TunnelID: tunnelID, Code: code}) + return a.sendControl(protocol.RevokeInviteMessage{Type: protocol.TypeRevokeInvite, TunnelID: scope.TunnelID, GroupID: scope.GroupID, Code: code}) } // DecideRequest approves or rejects a pending access request -func (a *Agent) DecideRequest(tunnelID, requestID string, approve bool) error { +func (a *Agent) DecideRequest(scope Scope, requestID string, approve bool) error { if !a.IsConnected() { return fmt.Errorf("未连接中继服务器") } - return a.sendControl(protocol.DecideRequestMessage{Type: protocol.TypeDecideRequest, TunnelID: tunnelID, RequestID: requestID, Approve: approve}) + return a.sendControl(protocol.DecideRequestMessage{Type: protocol.TypeDecideRequest, TunnelID: scope.TunnelID, GroupID: scope.GroupID, RequestID: requestID, Approve: approve}) } // RevokeLease removes a temporary whitelist lease on the relay server (all leases when ip is empty) -func (a *Agent) RevokeLease(tunnelID, ip string) error { +func (a *Agent) RevokeLease(scope Scope, ip string) error { if !a.IsConnected() { return fmt.Errorf("未连接中继服务器") } - return a.sendControl(protocol.RevokeLeaseMessage{Type: protocol.TypeRevokeLease, TunnelID: tunnelID, IP: ip}) + return a.sendControl(protocol.RevokeLeaseMessage{Type: protocol.TypeRevokeLease, TunnelID: scope.TunnelID, GroupID: scope.GroupID, IP: ip}) +} + +// DeleteGroup tells the relay server to drop a dissolved group's invites and leases +func (a *Agent) DeleteGroup(groupID string) { + if !a.IsConnected() { + return + } + _ = a.sendControl(protocol.DeleteGroupMessage{Type: protocol.TypeDeleteGroup, GroupID: groupID}) } func (a *Agent) KillConn(connID string) { @@ -801,6 +856,7 @@ func (a *Agent) GetLatestMetrics() protocol.MetricsMessage { defer a.mu.RUnlock() m := a.latestMetrics m.Tunnels = append([]protocol.TunnelMetrics(nil), m.Tunnels...) + m.Groups = append([]protocol.GroupMetrics(nil), m.Groups...) m.ActiveConns = append([]protocol.ActiveConnInfo(nil), m.ActiveConns...) m.SecurityEvents = append([]protocol.SecurityEvent(nil), m.SecurityEvents...) return m diff --git a/client/api.go b/client/api.go index c6cb3c3..8416547 100644 --- a/client/api.go +++ b/client/api.go @@ -42,6 +42,8 @@ func (s *APIServer) Handler() http.Handler { mux.HandleFunc("/api/status", s.handleStatus) mux.HandleFunc("/api/tunnels", s.handleTunnels) mux.HandleFunc("/api/tunnels/", s.handleTunnelItem) + mux.HandleFunc("/api/groups", s.handleGroups) + mux.HandleFunc("/api/groups/", s.handleGroupItem) mux.HandleFunc("/api/probe", s.handleProbe) mux.HandleFunc("/api/recommend-port", s.handleRecommendPort) mux.HandleFunc("/api/connections", s.handleConnections) @@ -212,47 +214,13 @@ func (s *APIServer) handleTunnelItem(w http.ResponseWriter, r *http.Request) { return } - // POST /api/tunnels/{id}/leases/revoke {"ip": "..."}；ip 为空表示撤销该通道全部临时放通 - if len(parts) == 5 && parts[3] == "leases" && parts[4] == "revoke" && r.Method == http.MethodPost { - var req struct { - IP string `json:"ip"` - } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err != io.EOF { - jsonError(w, http.StatusBadRequest, "Invalid JSON") - return - } + // 授权相关：leases/revoke、invites、requests（与通道组共用同一套处理） + if len(parts) >= 4 && isAccessPath(parts[3]) { if _, ok := s.cm.FindTunnel(tunnelID); !ok { jsonError(w, http.StatusNotFound, "通道未找到") return } - if err := s.agent.RevokeLease(tunnelID, strings.TrimSpace(req.IP)); err != nil { - jsonError(w, http.StatusConflict, err.Error()) - return - } - jsonResponse(w, http.StatusOK, map[string]string{"status": "revoked"}) - return - } - - // 邀请链接：POST 新建；DELETE /invites/{code} 作废；GET /invites/{code}/share 获取分享文案 - if len(parts) >= 4 && parts[3] == "invites" { - s.handleInvites(w, r, tunnelID, parts[4:]) - return - } - - // POST /api/tunnels/{id}/requests/{rid} {"approve": true|false} - if len(parts) == 5 && parts[3] == "requests" && r.Method == http.MethodPost { - var req struct { - Approve bool `json:"approve"` - } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - jsonError(w, http.StatusBadRequest, "Invalid JSON") - return - } - if err := s.agent.DecideRequest(tunnelID, parts[4], req.Approve); err != nil { - jsonError(w, http.StatusConflict, err.Error()) - return - } - jsonResponse(w, http.StatusOK, map[string]string{"status": "ok"}) + s.handleAccess(w, r, Scope{TunnelID: tunnelID}, parts[3:]) return } @@ -303,44 +271,59 @@ func (s *APIServer) handleTunnelItem(w http.ResponseWriter, r *http.Request) { // handleToggle 支持 {"enabled": true|false} 指定目标状态（幂等，防止重复点击来回翻转）；不带 body 时取反 func (s *APIServer) handleToggle(w http.ResponseWriter, r *http.Request, tunnelID string) { + want, err := decodeToggle(r) + if err != nil { + jsonError(w, http.StatusBadRequest, "Invalid JSON") + return + } + t, err := s.setEnabled(tunnelID, want) + if err != nil { + jsonError(w, http.StatusNotFound, err.Error()) + return + } + jsonResponse(w, http.StatusOK, t) +} + +// decodeToggle reads {"enabled": bool}; nil means "flip" +func decodeToggle(r *http.Request) (*bool, error) { var req struct { Enabled *bool `json:"enabled"` } if r.Body != nil { if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err != io.EOF { - jsonError(w, http.StatusBadRequest, "Invalid JSON") - return + return nil, err } } + return req.Enabled, nil +} +// setEnabled turns a tunnel on or off (want nil flips it), merging session traffic and syncing the server +func (s *APIServer) setEnabled(tunnelID string, want *bool) (protocol.TunnelConfig, error) { current, err := s.cm.CurrentEnabled(tunnelID) if err != nil { - jsonError(w, http.StatusNotFound, err.Error()) - return + return protocol.TunnelConfig{}, err } - want := !current - if req.Enabled != nil { - want = *req.Enabled + target := !current + if want != nil { + target = *want } - var sessIn, sessOut int64 - if !want && current { + if !target && current { s.agent.FlushUsage(tunnelID) } - if want != current { + if target != current { // 关闭时把本次会话流量并入历史；开启时清掉可能残留的旧指标 sessIn, sessOut = s.agent.ConsumeSessionBytes(tunnelID) } - t, changed, err := s.cm.SetTunnelEnabled(tunnelID, want, sessIn, sessOut) + t, changed, err := s.cm.SetTunnelEnabled(tunnelID, target, sessIn, sessOut) if err != nil { - jsonError(w, http.StatusNotFound, err.Error()) - return + return t, err } if changed { s.agent.dropUsage(tunnelID) s.agent.SyncTunnel(t, false) } - jsonResponse(w, http.StatusOK, t) + return t, nil } // inviteBase is the URL prefix of invite links: http://<对外访问地址>:<控制端口>/i/ @@ -358,13 +341,26 @@ func (s *APIServer) inviteBase(cfg Config) string { // shareText is the ready-to-send message (e.g. for WeChat) for an invite func shareText(cfg Config, t protocol.TunnelConfig, inv protocol.Invite, link string) string { + return shareMessage(inv, fmt.Sprintf("通道：%s\n访问地址：%s:%d", t.Name, cfg.PublicHostOrDefault(), t.ListenPort), link) +} + +// groupShareText lists every open address of a group with the target behind it +func groupShareText(cfg Config, g TunnelGroup, members []protocol.TunnelConfig, inv protocol.Invite, link string) string { + var b strings.Builder + fmt.Fprintf(&b, "通道组：%s（%d 个地址）\n访问地址：", g.Name, len(members)) + for _, t := range members { + fmt.Fprintf(&b, "\n %s:%d → %s", cfg.PublicHostOrDefault(), t.ListenPort, t.Target) + } + return shareMessage(inv, b.String(), link) +} + +func shareMessage(inv protocol.Invite, body, link string) string { var b strings.Builder b.WriteString("【PortRelay 访问邀请】") if inv.Note != "" { b.WriteString("（给 " + inv.Note + "）") } - b.WriteString("\n通道：" + t.Name) - b.WriteString(fmt.Sprintf("\n访问地址：%s:%d", cfg.PublicHostOrDefault(), t.ListenPort)) + b.WriteString("\n" + body) action := "填写姓名后即可访问" if inv.RequireApproval { action = "填写姓名提交申请，管理员批准后即可访问" @@ -379,12 +375,104 @@ func shareText(cfg Config, t protocol.TunnelConfig, inv protocol.Invite, link st return b.String() } -func (s *APIServer) handleInvites(w http.ResponseWriter, r *http.Request, tunnelID string, rest []string) { - t, ok := s.cm.FindTunnel(tunnelID) - if !ok { - jsonError(w, http.StatusNotFound, "通道未找到") - return +func isAccessPath(p string) bool { + return p == "leases" || p == "invites" || p == "requests" +} + +// handleAccess serves the access endpoints shared by tunnels and groups: +// +// POST leases/revoke {"ip": "..."}（ip 为空表示撤销全部） +// POST invites 新建邀请 +// DELETE invites/{code} 作废邀请 +// GET invites/{code}/share 分享文案 +// POST requests/{rid} {"approve": true|false} +func (s *APIServer) handleAccess(w http.ResponseWriter, r *http.Request, sc Scope, rest []string) { + switch { + case len(rest) == 2 && rest[0] == "leases" && rest[1] == "revoke" && r.Method == http.MethodPost: + var req struct { + IP string `json:"ip"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err != io.EOF { + jsonError(w, http.StatusBadRequest, "Invalid JSON") + return + } + if err := s.agent.RevokeLease(sc, strings.TrimSpace(req.IP)); err != nil { + jsonError(w, http.StatusConflict, err.Error()) + return + } + jsonResponse(w, http.StatusOK, map[string]string{"status": "revoked"}) + + case rest[0] == "invites": + s.handleInvites(w, r, sc, rest[1:]) + + case len(rest) == 2 && rest[0] == "requests" && r.Method == http.MethodPost: + var req struct { + Approve bool `json:"approve"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + jsonError(w, http.StatusBadRequest, "Invalid JSON") + return + } + if err := s.agent.DecideRequest(sc, rest[1], req.Approve); err != nil { + jsonError(w, http.StatusConflict, err.Error()) + return + } + jsonResponse(w, http.StatusOK, map[string]string{"status": "ok"}) + + default: + http.NotFound(w, r) } +} + +// scopeShare builds the share message for an invite of a tunnel or group +func (s *APIServer) scopeShare(cfg Config, sc Scope, inv protocol.Invite, link string) string { + if sc.GroupID != "" { + g, _ := s.cm.FindGroup(sc.GroupID) + var open []protocol.TunnelConfig + for _, t := range s.cm.GroupMembers(sc.GroupID) { + if t.Enabled { + open = append(open, t) + } + } + return groupShareText(cfg, g, open, inv, link) + } + t, _ := s.cm.FindTunnel(sc.TunnelID) + return shareText(cfg, t, inv, link) +} + +// scopeOpen reports whether the tunnel, or at least one tunnel of the group, is enabled +func (s *APIServer) scopeOpen(sc Scope) bool { + if sc.GroupID != "" { + for _, t := range s.cm.GroupMembers(sc.GroupID) { + if t.Enabled { + return true + } + } + return false + } + t, ok := s.cm.FindTunnel(sc.TunnelID) + return ok && t.Enabled +} + +// scopeInvites returns the live invites of a tunnel or group +func scopeInvites(m protocol.MetricsMessage, sc Scope) []protocol.Invite { + if sc.GroupID != "" { + for _, g := range m.Groups { + if g.GroupID == sc.GroupID { + return g.Invites + } + } + return nil + } + for _, t := range m.Tunnels { + if t.TunnelID == sc.TunnelID { + return t.Invites + } + } + return nil +} + +func (s *APIServer) handleInvites(w http.ResponseWriter, r *http.Request, sc Scope, rest []string) { cfg := s.cm.Get() switch { @@ -415,11 +503,15 @@ func (s *APIServer) handleInvites(w http.ResponseWriter, r *http.Request, tunnel if req.LeaseDuration == "" { req.LeaseDuration = protocol.LeaseDuration2h } - if !t.Enabled { - jsonError(w, http.StatusConflict, "通道未开启，请先开启通道再生成邀请链接") + if !s.scopeOpen(sc) { + msg := "通道未开启，请先开启通道再生成邀请链接" + if sc.GroupID != "" { + msg = "通道组内没有开启的通道，请先开启再生成邀请链接" + } + jsonError(w, http.StatusConflict, msg) return } - inv, err := s.agent.CreateInvite(tunnelID, strings.TrimSpace(req.Note), req.ValidMinutes, maxUses, req.LeaseDuration, req.RequireApproval) + inv, err := s.agent.CreateInvite(sc, strings.TrimSpace(req.Note), req.ValidMinutes, maxUses, req.LeaseDuration, req.RequireApproval) if err != nil { jsonError(w, http.StatusConflict, err.Error()) return @@ -428,27 +520,22 @@ func (s *APIServer) handleInvites(w http.ResponseWriter, r *http.Request, tunnel jsonResponse(w, http.StatusCreated, map[string]any{ "invite": inv, "link": link, - "share_text": shareText(cfg, t, inv, link), + "share_text": s.scopeShare(cfg, sc, inv, link), }) case len(rest) == 1 && r.Method == http.MethodDelete: - if err := s.agent.RevokeInvite(tunnelID, rest[0]); err != nil { + if err := s.agent.RevokeInvite(sc, rest[0]); err != nil { jsonError(w, http.StatusConflict, err.Error()) return } jsonResponse(w, http.StatusOK, map[string]string{"status": "revoked"}) case len(rest) == 2 && rest[1] == "share" && r.Method == http.MethodGet: - for _, m := range s.agent.GetLatestMetrics().Tunnels { - if m.TunnelID != tunnelID { - continue - } - for _, inv := range m.Invites { - if inv.Code == rest[0] { - link := s.inviteBase(cfg) + inv.Code - jsonResponse(w, http.StatusOK, map[string]any{"invite": inv, "link": link, "share_text": shareText(cfg, t, inv, link)}) - return - } + for _, inv := range scopeInvites(s.agent.GetLatestMetrics(), sc) { + if inv.Code == rest[0] { + link := s.inviteBase(cfg) + inv.Code + jsonResponse(w, http.StatusOK, map[string]any{"invite": inv, "link": link, "share_text": s.scopeShare(cfg, sc, inv, link)}) + return } } jsonError(w, http.StatusNotFound, "邀请链接不存在或已过期") @@ -504,10 +591,10 @@ func (s *APIServer) handleRecommendPort(w http.ResponseWriter, r *http.Request) // handleConnections returns live client connections (optionally for one tunnel) and recent security events func (s *APIServer) handleConnections(w http.ResponseWriter, r *http.Request) { metrics := s.agent.GetLatestMetrics() - tunnelID := r.URL.Query().Get("tunnel_id") + match := s.tunnelFilter(r) conns := make([]protocol.ActiveConnInfo, 0, len(metrics.ActiveConns)) for _, c := range metrics.ActiveConns { - if tunnelID == "" || c.TunnelID == tunnelID { + if match(c.TunnelID) { conns = append(conns, c) } } @@ -537,8 +624,29 @@ func (s *APIServer) handleKillConn(w http.ResponseWriter, r *http.Request) { } func (s *APIServer) handleClientHistory(w http.ResponseWriter, r *http.Request) { - tunnelID := r.URL.Query().Get("tunnel_id") - jsonResponse(w, http.StatusOK, s.cm.GetClientHistory(tunnelID)) + match := s.tunnelFilter(r) + out := make([]protocol.ClientHistoryRecord, 0) + for _, rec := range s.cm.GetClientHistory("") { + if match(rec.TunnelID) { + out = append(out, rec) + } + } + jsonResponse(w, http.StatusOK, out) +} + +// tunnelFilter matches tunnels selected by ?tunnel_id= or ?group_id= (everything when neither is set) +func (s *APIServer) tunnelFilter(r *http.Request) func(id string) bool { + if gid := r.URL.Query().Get("group_id"); gid != "" { + ids := make(map[string]bool) + for _, t := range s.cm.GroupMembers(gid) { + ids[t.ID] = true + } + return func(id string) bool { return ids[id] } + } + if tid := r.URL.Query().Get("tunnel_id"); tid != "" { + return func(id string) bool { return id == tid } + } + return func(string) bool { return true } } func (s *APIServer) handleSettings(w http.ResponseWriter, r *http.Request) { @@ -595,6 +703,7 @@ func (s *APIServer) handleSSE(w http.ResponseWriter, r *http.Request) { payload := s.statusPayload() payload["metrics"] = metrics payload["tunnels"] = s.tunnelsWithMetrics(metrics) + payload["groups"] = s.groupsView(metrics) data, _ := json.Marshal(payload) if _, err := fmt.Fprintf(w, "data: %s\n\n", data); err != nil { return false diff --git a/client/api_groups.go b/client/api_groups.go new file mode 100644 index 0000000..cd063f4 --- /dev/null +++ b/client/api_groups.go @@ -0,0 +1,277 @@ +package client + +import ( + "encoding/json" + "io" + "net/http" + "strings" + + "github.com/DeepFal/PortRelay/core/protocol" +) + +// GroupView joins a tunnel group with its members and live shared access state +type GroupView struct { + TunnelGroup + TunnelIDs []string `json:"tunnel_ids"` // 成员，按监听端口排序 + Metrics protocol.GroupMetrics `json:"metrics"` + // RateHistory 开启中成员的近 10 分钟流量之和 + RateHistory *RateSeries `json:"rate_history,omitempty"` +} + +func (s *APIServer) groupsView(metrics protocol.MetricsMessage) []GroupView { + shared := make(map[string]protocol.GroupMetrics, len(metrics.Groups)) + for _, g := range metrics.Groups { + shared[g.GroupID] = g + } + cfg := s.cm.Get() + out := make([]GroupView, 0, len(cfg.Groups)) + for _, g := range cfg.Groups { + v := GroupView{TunnelGroup: g, TunnelIDs: make([]string, 0), Metrics: protocol.GroupMetrics{GroupID: g.ID}} + if m, ok := shared[g.ID]; ok { + v.Metrics = m + } + var open []string + for _, t := range s.cm.GroupMembers(g.ID) { + v.TunnelIDs = append(v.TunnelIDs, t.ID) + if t.Enabled { + open = append(open, t.ID) + } + } + if len(open) > 0 { + series := s.agent.GroupRateSeries(open) + v.RateHistory = &series + } + out = append(out, v) + } + return out +} + +// groupRequest creates a group or adds members to one +type groupRequest struct { + Name string `json:"name"` + Owner string `json:"owner"` + // Hosts / Ports 展开为新建成员，见 ExpandTargets + Hosts string `json:"hosts"` + Ports string `json:"ports"` + // TunnelIDs 加入该组的已有通道 + TunnelIDs []string `json:"tunnel_ids"` + // 新建成员的通道设置 + Whitelist []string `json:"whitelist"` + IdleCloseMinutes *int `json:"idle_close_minutes"` + DurationMinutes int `json:"duration_minutes"` + MaxBandwidthMB float64 `json:"max_bandwidth_mb"` +} + +func (req groupRequest) spec() (GroupSpec, error) { + spec := GroupSpec{ + Name: req.Name, + Owner: req.Owner, + MoveIDs: req.TunnelIDs, + Template: protocol.TunnelConfig{ + Whitelist: req.Whitelist, + IdleCloseMinutes: req.IdleCloseMinutes, + DurationMinutes: req.DurationMinutes, + MaxBandwidthMB: req.MaxBandwidthMB, + }, + } + if strings.TrimSpace(req.Hosts) != "" { + targets, err := ExpandTargets(req.Hosts, req.Ports) + if err != nil { + return spec, err + } + spec.Targets = targets + } + return spec, nil +} + +func decodeGroupRequest(r *http.Request) (groupRequest, error) { + var req groupRequest + err := json.NewDecoder(r.Body).Decode(&req) + if err == io.EOF { + err = nil + } + return req, err +} + +// syncGroupChanges pushes newly created members and changed existing ones to the relay server +func (s *APIServer) syncGroupChanges(created, changed []protocol.TunnelConfig) { + for _, t := range created { + s.agent.SyncTunnel(t, false) + } + for _, t := range changed { + if t.Enabled { + s.agent.SyncTunnel(t, true) + } + } +} + +// handleGroups: GET 列表；POST 新建通道组（批量新建目标 + / 或加入已有通道） +func (s *APIServer) handleGroups(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + jsonResponse(w, http.StatusOK, s.groupsView(s.agent.GetLatestMetrics())) + case http.MethodPost: + req, err := decodeGroupRequest(r) + if err != nil { + jsonError(w, http.StatusBadRequest, "Invalid JSON") + return + } + spec, err := req.spec() + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return + } + g, created, changed, err := s.cm.SaveGroup("", spec) + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return + } + s.syncGroupChanges(created, changed) + jsonResponse(w, http.StatusCreated, map[string]any{"group": g, "created": nonNil(created)}) + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } +} + +// handleGroupItem: +// +// POST /api/groups/preview 预览批量展开结果（不保存） +// PUT /api/groups/{id} 改名 / 责任人，并可追加成员 +// DELETE /api/groups/{id}?delete_members=1 解散（默认保留成员为独立通道） +// POST /api/groups/{id}/toggle {"enabled": bool} 整组开关 +// ... /api/groups/{id}/leases|invites|requests 与单个通道相同的授权接口 +func (s *APIServer) handleGroupItem(w http.ResponseWriter, r *http.Request) { + parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/") + if len(parts) < 3 || parts[2] == "" { + jsonError(w, http.StatusBadRequest, "Missing group ID") + return + } + if parts[2] == "preview" && len(parts) == 3 && r.Method == http.MethodPost { + s.handleGroupPreview(w, r) + return + } + id := parts[2] + g, ok := s.cm.FindGroup(id) + if !ok { + jsonError(w, http.StatusNotFound, "通道组不存在") + return + } + + if len(parts) >= 4 && isAccessPath(parts[3]) { + s.handleAccess(w, r, Scope{GroupID: id}, parts[3:]) + return + } + if len(parts) == 4 && parts[3] == "toggle" && r.Method == http.MethodPost { + want, err := decodeToggle(r) + if err != nil { + jsonError(w, http.StatusBadRequest, "Invalid JSON") + return + } + if want == nil { + // 不指定时：有任一成员开启则整组关闭，否则整组开启 + on := false + for _, t := range s.cm.GroupMembers(id) { + on = on || t.Enabled + } + flip := !on + want = &flip + } + out := make([]protocol.TunnelConfig, 0) + for _, t := range s.cm.GroupMembers(id) { + updated, err := s.setEnabled(t.ID, want) + if err != nil { + jsonError(w, http.StatusConflict, err.Error()) + return + } + out = append(out, updated) + } + jsonResponse(w, http.StatusOK, out) + return + } + if len(parts) != 3 { + http.NotFound(w, r) + return + } + + switch r.Method { + case http.MethodPut: + req, err := decodeGroupRequest(r) + if err != nil { + jsonError(w, http.StatusBadRequest, "Invalid JSON") + return + } + if strings.TrimSpace(req.Name) == "" { + req.Name = g.Name + } + spec, err := req.spec() + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return + } + updated, created, changed, err := s.cm.SaveGroup(id, spec) + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return + } + s.syncGroupChanges(created, changed) + jsonResponse(w, http.StatusOK, map[string]any{"group": updated, "created": nonNil(created)}) + + case http.MethodDelete: + deleteMembers := r.URL.Query().Get("delete_members") == "1" + members, err := s.cm.RemoveGroup(id, deleteMembers) + if err != nil { + jsonError(w, http.StatusNotFound, err.Error()) + return + } + for _, t := range members { + if deleteMembers { + s.agent.ConsumeSessionBytes(t.ID) + s.agent.dropUsage(t.ID) + s.agent.RemoveTunnel(t.ID) + } else if t.Enabled { + s.agent.SyncTunnel(t, true) + } + } + s.agent.DeleteGroup(id) + jsonResponse(w, http.StatusOK, map[string]string{"status": "deleted"}) + + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } +} + +// handleGroupPreview expands hosts × ports and assigns ports without saving +func (s *APIServer) handleGroupPreview(w http.ResponseWriter, r *http.Request) { + req, err := decodeGroupRequest(r) + if err != nil { + jsonError(w, http.StatusBadRequest, "Invalid JSON") + return + } + spec, err := req.spec() + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return + } + members, err := s.cm.PreviewGroupMembers(spec) + if err != nil { + jsonError(w, http.StatusBadRequest, err.Error()) + return + } + type item struct { + Name string `json:"name"` + Target string `json:"target"` + ListenPort int `json:"listen_port"` + } + out := make([]item, 0, len(members)) + for _, t := range members { + out = append(out, item{t.Name, t.Target, t.ListenPort}) + } + jsonResponse(w, http.StatusOK, map[string]any{"members": out}) +} + +func nonNil(ts []protocol.TunnelConfig) []protocol.TunnelConfig { + if ts == nil { + return []protocol.TunnelConfig{} + } + return ts +} diff --git a/client/batch.go b/client/batch.go new file mode 100644 index 0000000..2af5571 --- /dev/null +++ b/client/batch.go @@ -0,0 +1,209 @@ +package client + +import ( + "fmt" + "net" + "regexp" + "strconv" + "strings" +) + +// maxBatchTargets 一次批量创建的目标上限，避免误操作占满端口池 +const maxBatchTargets = 64 + +var ( + listSeparators = regexp.MustCompile(`[\s,，、;；]+`) + shortRange = regexp.MustCompile(`^(\d{1,3}\.\d{1,3}\.\d{1,3}\.)(\d{1,3})-(\d{1,3})$`) + fullRange = regexp.MustCompile(`^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})-(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$`) + hostnameRe = regexp.MustCompile(`^[A-Za-z0-9]([A-Za-z0-9.-]*[A-Za-z0-9])?$`) +) + +func splitList(s string) []string { + var out []string + for _, item := range listSeparators.Split(strings.TrimSpace(s), -1) { + if item != "" { + out = append(out, item) + } + } + return out +} + +// parsePorts accepts "22, 8080-8082" +func parsePorts(s string) ([]int, error) { + var out []int + for _, item := range splitList(s) { + lo, hi := item, item + if i := strings.Index(item, "-"); i > 0 { + lo, hi = item[:i], item[i+1:] + } + a, err1 := strconv.Atoi(lo) + b, err2 := strconv.Atoi(hi) + if err1 != nil || err2 != nil || a <= 0 || b > 65535 || a > b { + return nil, fmt.Errorf("无效的端口 %q", item) + } + if b-a >= maxBatchTargets { + return nil, fmt.Errorf("端口范围 %q 过大", item) + } + for p := a; p <= b; p++ { + out = append(out, p) + } + } + return out, nil +} + +// expandIPv4Range expands "10.0.1.11-13" or "10.0.1.11-10.0.1.13" +func expandIPv4Range(item string) ([]string, bool, error) { + var from, to net.IP + if m := shortRange.FindStringSubmatch(item); m != nil { + from, to = net.ParseIP(m[1]+m[2]).To4(), net.ParseIP(m[1]+m[3]).To4() + } else if m := fullRange.FindStringSubmatch(item); m != nil { + from, to = net.ParseIP(m[1]).To4(), net.ParseIP(m[2]).To4() + } else { + return nil, false, nil + } + if from == nil || to == nil { + return nil, true, fmt.Errorf("无效的地址范围 %q", item) + } + a := uint32(from[0])<<24 | uint32(from[1])<<16 | uint32(from[2])<<8 | uint32(from[3]) + b := uint32(to[0])<<24 | uint32(to[1])<<16 | uint32(to[2])<<8 | uint32(to[3]) + if a > b { + return nil, true, fmt.Errorf("地址范围 %q 起止颠倒", item) + } + if b-a >= maxBatchTargets { + return nil, true, fmt.Errorf("地址范围 %q 过大", item) + } + var out []string + for v := a; v <= b; v++ { + out = append(out, net.IPv4(byte(v>>24), byte(v>>16), byte(v>>8), byte(v)).String()) + } + return out, true, nil +} + +func validHost(h string) bool { + return net.ParseIP(h) != nil || hostnameRe.MatchString(h) +} + +// ExpandTargets turns host and port lists into "host:port" targets, host by host. +// +// hosts 用逗号、空格或换行分隔，每一项可以是 IP、主机名、IPv4 范围（"10.0.1.11-13" 或 +// "10.0.1.11-10.0.1.13"），也可以直接写成 "主机:端口"；ports 形如 "22, 8080-8082"， +// 作用于没有写端口的主机。结果去重并保持输入顺序。 +func ExpandTargets(hosts, ports string) ([]string, error) { + portList, err := parsePorts(ports) + if err != nil { + return nil, err + } + seen := make(map[string]bool) + var out []string + add := func(t string) error { + if !seen[t] { + seen[t] = true + out = append(out, t) + } + if len(out) > maxBatchTargets { + return fmt.Errorf("一次最多添加 %d 个目标", maxBatchTargets) + } + return nil + } + items := splitList(hosts) + if len(items) == 0 { + return nil, fmt.Errorf("请填写至少一个目标主机") + } + for _, item := range items { + // 自带端口的完整目标："10.0.1.11:22"、"[fd00::1]:22" + if h, p, err := net.SplitHostPort(item); err == nil { + n, perr := strconv.Atoi(p) + if !validHost(h) || perr != nil || n <= 0 || n > 65535 { + return nil, fmt.Errorf("无效的目标 %q", item) + } + if err := add(net.JoinHostPort(h, p)); err != nil { + return nil, err + } + continue + } + hostsOfItem, isRange, err := expandIPv4Range(item) + if err != nil { + return nil, err + } + if !isRange { + if !validHost(item) { + return nil, fmt.Errorf("无效的主机 %q", item) + } + hostsOfItem = []string{item} + } + if len(portList) == 0 { + return nil, fmt.Errorf("请填写端口，或把目标写成 主机:端口") + } + for _, h := range hostsOfItem { + for _, p := range portList { + if err := add(net.JoinHostPort(h, strconv.Itoa(p))); err != nil { + return nil, err + } + } + } + } + return out, nil +} + +// memberNames names batch-created tunnels after the group: "组名 (主机)" when every target +// shares one port, otherwise "组名 (主机:端口)". +func memberNames(group string, targets []string) []string { + ports := make(map[string]bool) + for _, t := range targets { + if _, p, err := net.SplitHostPort(t); err == nil { + ports[p] = true + } + } + names := make([]string, len(targets)) + for i, t := range targets { + label := t + if len(ports) == 1 { + if h, _, err := net.SplitHostPort(t); err == nil { + label = h + } + } + names[i] = fmt.Sprintf("%s (%s)", group, label) + } + return names +} + +// allocatePortsLocked picks n free listen ports, preferring one contiguous block so the mapping +// is easy to read. Caller must hold cm.mu. +func (cm *ConfigManager) allocatePortsLocked(n int) ([]int, error) { + used := make(map[int]bool) + for _, t := range cm.cfg.Tunnels { + used[t.ListenPort] = true + } + // 与单个通道的推荐顺序一致：先 33660 往上，再从端口池开头补 + var order []int + for p := 33660; p <= PortMax; p++ { + order = append(order, p) + } + for p := PortMin; p < 33660; p++ { + order = append(order, p) + } + free := func(p int) bool { return p != ControlPort && !used[p] } + + for i := range order { + ok := true + for k := 0; k < n; k++ { + if i+k >= len(order) || !free(order[i+k]) || (k > 0 && order[i+k] != order[i+k-1]+1) { + ok = false + break + } + } + if ok { + return order[i : i+n], nil + } + } + var out []int + for _, p := range order { + if free(p) { + out = append(out, p) + if len(out) == n { + return out, nil + } + } + } + return nil, fmt.Errorf("端口池剩余 %d 个空闲端口，不够分配 %d 个", len(out), n) +} diff --git a/client/config.go b/client/config.go index 6ac3b32..4960546 100644 --- a/client/config.go +++ b/client/config.go @@ -11,6 +11,7 @@ import ( "net" "os" "path/filepath" + "sort" "strconv" "strings" "sync" @@ -49,12 +50,21 @@ func generateAuthCode() string { return string(b) } +// TunnelGroup 通道组：组内通道共用邀请、授权与审批，闲置计时联动 +type TunnelGroup struct { + ID string `json:"id"` + Name string `json:"name"` + Owner string `json:"owner,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + // Config represents persistent client settings type Config struct { ServerAddr string `json:"server_addr"` // 中继服务器控制地址 (host:port) Token string `json:"token"` PublicHost string `json:"public_host,omitempty"` // 访问者连接通道时使用的网关地址 Tunnels []protocol.TunnelConfig `json:"tunnels"` + Groups []TunnelGroup `json:"groups,omitempty"` ClientHistory []protocol.ClientHistoryRecord `json:"client_history,omitempty"` } @@ -208,6 +218,7 @@ func (cm *ConfigManager) Get() Config { for i, t := range cm.cfg.Tunnels { out.Tunnels[i] = cloneTunnel(t) } + out.Groups = append([]TunnelGroup(nil), cm.cfg.Groups...) out.ClientHistory = append([]protocol.ClientHistoryRecord(nil), cm.cfg.ClientHistory...) return out } @@ -316,9 +327,24 @@ func normalizeAndValidate(t *protocol.TunnelConfig) error { // 放通时长与是否审批已改为在每个邀请上设置，通道级字段不再使用 t.LeaseDuration = "" t.RequireApproval = false + // 组名只在同步到服务端时填写，配置里以组 ID 为准 + t.GroupName = "" return nil } +// groupExistsLocked reports whether a group ID is known ("" means no group). Caller must hold cm.mu. +func (cm *ConfigManager) groupExistsLocked(id string) bool { + if id == "" { + return true + } + for _, g := range cm.cfg.Groups { + if g.ID == id { + return true + } + } + return false +} + // checkPortLocked returns an error if another tunnel already uses the port. Caller must hold cm.mu. func (cm *ConfigManager) checkPortLocked(id string, port int) error { for _, existing := range cm.cfg.Tunnels { @@ -350,6 +376,9 @@ func (cm *ConfigManager) AddTunnel(t protocol.TunnelConfig) (protocol.TunnelConf if err := cm.checkPortLocked(t.ID, t.ListenPort); err != nil { return t, err } + if !cm.groupExistsLocked(t.GroupID) { + return t, fmt.Errorf("通道组不存在") + } t.Enabled = true now := time.Now() @@ -409,6 +438,9 @@ func (cm *ConfigManager) PatchTunnel(id string, patch []byte) (protocol.TunnelCo if err := cm.checkPortLocked(id, updated.ListenPort); err != nil { return existing, existing.ListenPort, err } + if !cm.groupExistsLocked(updated.GroupID) { + return existing, existing.ListenPort, fmt.Errorf("通道组不存在") + } // 运行中修改时长：从本次开启时间重新计算到期时间 if updated.Enabled && updated.DurationMinutes != existing.DurationMinutes { @@ -700,3 +732,221 @@ func (cm *ConfigManager) RecommendPort() int { } return 0 } + +// ---- 通道组 ---- + +// FindGroup returns the group with the given ID +func (cm *ConfigManager) FindGroup(id string) (TunnelGroup, bool) { + cm.mu.RLock() + defer cm.mu.RUnlock() + for _, g := range cm.cfg.Groups { + if g.ID == id { + return g, true + } + } + return TunnelGroup{}, false +} + +// GroupMembers returns the tunnels of a group ordered by listen port +func (cm *ConfigManager) GroupMembers(id string) []protocol.TunnelConfig { + cm.mu.RLock() + defer cm.mu.RUnlock() + return cm.groupMembersLocked(id) +} + +func (cm *ConfigManager) groupMembersLocked(id string) []protocol.TunnelConfig { + var out []protocol.TunnelConfig + for _, t := range cm.cfg.Tunnels { + if id != "" && t.GroupID == id { + out = append(out, cloneTunnel(t)) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].ListenPort < out[j].ListenPort }) + return out +} + +// GroupSpec describes a group and the members to add to it +type GroupSpec struct { + Name string + Owner string + // Targets 新建成员的目标（已展开的 主机:端口） + Targets []string + // Template 新建成员的通道设置：白名单、闲置关闭、最长开启时长、限速 + Template protocol.TunnelConfig + // MoveIDs 加入该组的已有通道 + MoveIDs []string +} + +// newMembersLocked builds (but does not store) the tunnels for spec.Targets. Caller must hold cm.mu. +func (cm *ConfigManager) newMembersLocked(groupID string, spec GroupSpec, now time.Time) ([]protocol.TunnelConfig, error) { + if len(spec.Targets) == 0 { + return nil, nil + } + ports, err := cm.allocatePortsLocked(len(spec.Targets)) + if err != nil { + return nil, err + } + names := memberNames(spec.Name, spec.Targets) + out := make([]protocol.TunnelConfig, 0, len(spec.Targets)) + for i, target := range spec.Targets { + t := cloneTunnel(spec.Template) + t.ID = uuid.New().String() + t.Name, t.Target, t.ListenPort = names[i], target, ports[i] + t.GroupID = groupID + if strings.TrimSpace(t.Owner) == "" { + t.Owner = spec.Owner + } + if err := normalizeAndValidate(&t); err != nil { + return nil, fmt.Errorf("%s：%w", target, err) + } + t.AuthCode = generateAuthCode() + t.HistoricalBytesIn, t.HistoricalBytesOut, t.LastDurationSec = 0, 0, 0 + t.CloseReason, t.FirstConnectedAt, t.ConnectedSec, t.LastConnectedSec = "", nil, 0, 0 + t.DynamicWhitelist = nil + if t.IdleCloseMinutes == nil { + v := DefaultIdleCloseMinutes + t.IdleCloseMinutes = &v + } + t.Enabled = true + start := now + t.StartedAt = &start + t.ExpiresAt = nil + if t.DurationMinutes > 0 { + exp := now.Add(time.Duration(t.DurationMinutes) * time.Minute) + t.ExpiresAt = &exp + } + out = append(out, t) + } + return out, nil +} + +// PreviewGroupMembers returns the tunnels that saving spec would create, without saving anything +func (cm *ConfigManager) PreviewGroupMembers(spec GroupSpec) ([]protocol.TunnelConfig, error) { + if strings.TrimSpace(spec.Name) == "" { + spec.Name = "通道组" + } + spec.Name = strings.TrimSpace(spec.Name) + cm.mu.RLock() + defer cm.mu.RUnlock() + return cm.newMembersLocked("", spec, time.Now()) +} + +// SaveGroup creates a group (id == "") or renames / adds members to an existing one. +// Returns the group, the tunnels it created and the existing members whose config changed +// (moved in, or the group was renamed) — the caller pushes both to the relay server. +func (cm *ConfigManager) SaveGroup(id string, spec GroupSpec) (TunnelGroup, []protocol.TunnelConfig, []protocol.TunnelConfig, error) { + spec.Name = strings.TrimSpace(spec.Name) + spec.Owner = strings.TrimSpace(spec.Owner) + if spec.Name == "" { + return TunnelGroup{}, nil, nil, fmt.Errorf("请填写通道组名称") + } + if len([]rune(spec.Name)) > 64 { + return TunnelGroup{}, nil, nil, fmt.Errorf("通道组名称不能超过 64 个字") + } + + cm.mu.Lock() + defer cm.mu.Unlock() + + gi := -1 + for i, g := range cm.cfg.Groups { + if id != "" && g.ID == id { + gi = i + } else if g.Name == spec.Name { + return TunnelGroup{}, nil, nil, fmt.Errorf("已有同名的通道组「%s」", spec.Name) + } + } + if id != "" && gi < 0 { + return TunnelGroup{}, nil, nil, fmt.Errorf("通道组不存在") + } + if id == "" && len(spec.Targets) == 0 && len(spec.MoveIDs) == 0 { + return TunnelGroup{}, nil, nil, fmt.Errorf("请添加至少一个目标，或选择要加入的已有通道") + } + + group := TunnelGroup{ID: id, Name: spec.Name, Owner: spec.Owner, CreatedAt: time.Now()} + if gi >= 0 { + group.CreatedAt = cm.cfg.Groups[gi].CreatedAt + } else { + group.ID = uuid.New().String() + } + + move := make(map[string]bool) + for _, mid := range spec.MoveIDs { + found := false + for _, t := range cm.cfg.Tunnels { + if t.ID != mid { + continue + } + found = true + if t.GroupID != "" && t.GroupID != group.ID { + return TunnelGroup{}, nil, nil, fmt.Errorf("「%s」已属于其它通道组，请先将其移出", t.Name) + } + } + if !found { + return TunnelGroup{}, nil, nil, fmt.Errorf("通道 %s 未找到", mid) + } + move[mid] = true + } + + created, err := cm.newMembersLocked(group.ID, spec, time.Now()) + if err != nil { + return TunnelGroup{}, nil, nil, err + } + + renamed := gi >= 0 && cm.cfg.Groups[gi].Name != group.Name + var changed []protocol.TunnelConfig + for i, t := range cm.cfg.Tunnels { + switch { + case move[t.ID] && t.GroupID != group.ID: + cm.cfg.Tunnels[i].GroupID = group.ID + changed = append(changed, cloneTunnel(cm.cfg.Tunnels[i])) + case renamed && t.GroupID == group.ID: + changed = append(changed, cloneTunnel(t)) + } + } + cm.cfg.Tunnels = append(cm.cfg.Tunnels, created...) + if gi >= 0 { + cm.cfg.Groups[gi] = group + } else { + cm.cfg.Groups = append(cm.cfg.Groups, group) + } + out := make([]protocol.TunnelConfig, len(created)) + for i, t := range created { + out[i] = cloneTunnel(t) + } + return group, out, changed, cm.save() +} + +// RemoveGroup dissolves a group. With deleteMembers its tunnels are deleted too, otherwise they +// stay as standalone tunnels. Returns the affected members (as they are after the change). +func (cm *ConfigManager) RemoveGroup(id string, deleteMembers bool) ([]protocol.TunnelConfig, error) { + cm.mu.Lock() + defer cm.mu.Unlock() + groups := make([]TunnelGroup, 0, len(cm.cfg.Groups)) + found := false + for _, g := range cm.cfg.Groups { + if g.ID == id { + found = true + continue + } + groups = append(groups, g) + } + if !found { + return nil, fmt.Errorf("通道组不存在") + } + var affected []protocol.TunnelConfig + kept := make([]protocol.TunnelConfig, 0, len(cm.cfg.Tunnels)) + for _, t := range cm.cfg.Tunnels { + if t.GroupID != id { + kept = append(kept, t) + continue + } + t.GroupID = "" + affected = append(affected, cloneTunnel(t)) + if !deleteMembers { + kept = append(kept, t) + } + } + cm.cfg.Groups = groups + cm.cfg.Tunnels = kept + return affected, cm.save() +} diff --git a/client/group_test.go b/client/group_test.go new file mode 100644 index 0000000..25bff24 --- /dev/null +++ b/client/group_test.go @@ -0,0 +1,339 @@ +package client + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "reflect" + "strings" + "testing" + "time" + + "github.com/DeepFal/PortRelay/core/protocol" +) + +func TestExpandTargets(t *testing.T) { + cases := []struct { + hosts, ports string + want []string + }{ + {"10.0.1.11-13", "22", []string{"10.0.1.11:22", "10.0.1.12:22", "10.0.1.13:22"}}, + {"10.0.1.11, 10.0.1.12", "22, 8080", []string{"10.0.1.11:22", "10.0.1.11:8080", "10.0.1.12:22", "10.0.1.12:8080"}}, + {"10.0.1.11-10.0.1.12", "8080-8081", []string{"10.0.1.11:8080", "10.0.1.11:8081", "10.0.1.12:8080", "10.0.1.12:8081"}}, + {"db.internal:3306\n10.0.1.5", "22", []string{"db.internal:3306", "10.0.1.5:22"}}, + {"10.0.1.1，10.0.1.2、10.0.1.1", "22", []string{"10.0.1.1:22", "10.0.1.2:22"}}, + {"fd00::1", "22", []string{"[fd00::1]:22"}}, + {"10.0.2.9:5432", "", []string{"10.0.2.9:5432"}}, + } + for _, c := range cases { + got, err := ExpandTargets(c.hosts, c.ports) + if err != nil || !reflect.DeepEqual(got, c.want) { + t.Errorf("ExpandTargets(%q, %q) = %v, %v; want %v", c.hosts, c.ports, got, err, c.want) + } + } + for _, bad := range [][2]string{ + {"", "22"}, + {"10.0.1.11", ""}, + {"bad host!", "22"}, + {"10.0.1.1-200", "22"}, + {"10.0.1.13-11", "22"}, + {"10.0.1.11", "0"}, + {"10.0.1.11", "70000"}, + {"10.0.1.1-40", "22, 23"}, + } { + if got, err := ExpandTargets(bad[0], bad[1]); err == nil { + t.Errorf("ExpandTargets(%q, %q) should fail, got %v", bad[0], bad[1], got) + } + } +} + +func TestMemberNames(t *testing.T) { + if got := memberNames("订单服务", []string{"10.0.1.11:22", "10.0.1.12:22"}); got[0] != "订单服务 (10.0.1.11)" { + t.Errorf("single port should name by host: %v", got) + } + if got := memberNames("订单服务", []string{"10.0.1.11:22", "10.0.1.11:8080"}); got[1] != "订单服务 (10.0.1.11:8080)" { + t.Errorf("several ports should name by host:port: %v", got) + } +} + +func TestAllocatePortsPrefersContiguousBlock(t *testing.T) { + cm := newTestConfig(t, "127.0.0.1:1", testToken) + for _, p := range []int{33660, 33662} { + if _, err := cm.AddTunnel(protocol.TunnelConfig{Name: fmt.Sprint(p), Target: "10.0.0.1:22", ListenPort: p}); err != nil { + t.Fatal(err) + } + } + cm.mu.Lock() + three, err3 := cm.allocatePortsLocked(3) + one, err1 := cm.allocatePortsLocked(1) + cm.mu.Unlock() + if err3 != nil || !reflect.DeepEqual(three, []int{33663, 33664, 33665}) { + t.Fatalf("3 ports: %v %v", three, err3) + } + if err1 != nil || !reflect.DeepEqual(one, []int{33661}) { + t.Fatalf("1 port: %v %v", one, err1) + } +} + +type groupResp struct { + Group TunnelGroup `json:"group"` + Created []protocol.TunnelConfig `json:"created"` +} + +func (e *env) group(t *testing.T, id string) GroupView { + t.Helper() + _, body := e.call(t, "GET", "/api/groups", nil) + var list []GroupView + _ = json.Unmarshal(body, &list) + for _, g := range list { + if g.ID == id { + return g + } + } + t.Fatalf("group %s not found", id) + return GroupView{} +} + +func grantViaInvite(t *testing.T, e *env, code, name string) { + t.Helper() + resp, err := http.PostForm(fmt.Sprintf("http://%s/auth", e.addr), url.Values{"code": {code}, "name": {name}}) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("grant: %d", resp.StatusCode) + } +} + +func TestGroupEndToEnd(t *testing.T) { + e := startEnv(t, testToken) + waitFor(t, "agent connected", e.agent.IsConnected) + echo1, echo2 := startEcho(t), startEcho(t) + + status, body := e.call(t, "POST", "/api/groups", map[string]any{"name": "订单服务", "owner": "运维组", "hosts": echo1 + ", " + echo2}) + if status != http.StatusCreated { + t.Fatalf("create group: %d %s", status, body) + } + var g groupResp + _ = json.Unmarshal(body, &g) + if len(g.Created) != 2 || g.Created[1].ListenPort != g.Created[0].ListenPort+1 { + t.Fatalf("members should get a contiguous port block: %+v", g.Created) + } + m1, m2 := g.Created[0], g.Created[1] + for _, m := range g.Created { + if m.GroupID != g.Group.ID || m.Owner != "运维组" || len(m.Whitelist) != 0 || !strings.HasPrefix(m.Name, "订单服务 (") { + t.Fatalf("member not set up as a link-only group member: %+v", m) + } + } + waitFor(t, "members online", func() bool { + return e.tunnel(t, m1.ID).Metrics.Status == "online" && e.tunnel(t, m2.ID).Metrics.Status == "online" + }) + if _, err := echoVia(m1.ListenPort, "x"); err == nil { + t.Fatal("group members are link-only by default") + } + + status, body = e.call(t, "POST", "/api/groups/"+g.Group.ID+"/invites", map[string]any{"note": "小王", "lease_duration": "8h"}) + if status != http.StatusCreated { + t.Fatalf("group invite: %d %s", status, body) + } + var inv struct { + Invite protocol.Invite `json:"invite"` + ShareText string `json:"share_text"` + } + _ = json.Unmarshal(body, &inv) + for _, want := range []string{"通道组：订单服务（2 个地址）", fmt.Sprintf("127.0.0.1:%d → %s", m1.ListenPort, echo1), fmt.Sprintf("127.0.0.1:%d → %s", m2.ListenPort, echo2), "8 小时"} { + if !strings.Contains(inv.ShareText, want) { + t.Errorf("share text missing %q:\n%s", want, inv.ShareText) + } + } + waitFor(t, "group invite listed", func() bool { return len(e.group(t, g.Group.ID).Metrics.Invites) == 1 }) + if status, _ := e.call(t, "GET", "/api/groups/"+g.Group.ID+"/invites/"+inv.Invite.Code+"/share", nil); status != http.StatusOK { + t.Fatalf("share of existing group invite: %d", status) + } + + grantViaInvite(t, e, inv.Invite.Code, "小王") + for _, m := range g.Created { + if got, err := echoVia(m.ListenPort, "hi"); err != nil || got != "hi" { + t.Fatalf("member %d not reachable after group grant: %q %v", m.ListenPort, got, err) + } + } + waitFor(t, "group lease started", func() bool { + l := e.group(t, g.Group.ID).Metrics.DynamicLeases + return len(l) == 1 && l[0].Name == "小王" && !l[0].AwaitingConnect + }) + if gv := e.group(t, g.Group.ID); !reflect.DeepEqual(gv.TunnelIDs, []string{m1.ID, m2.ID}) || gv.RateHistory == nil { + t.Fatalf("group view: %+v", gv) + } + + if status, _ := e.call(t, "POST", "/api/groups/"+g.Group.ID+"/leases/revoke", map[string]string{}); status != http.StatusOK { + t.Fatal("revoke group leases failed") + } + waitFor(t, "group access revoked", func() bool { + _, err1 := echoVia(m1.ListenPort, "x") + _, err2 := echoVia(m2.ListenPort, "x") + return err1 != nil && err2 != nil + }) + + // 整组关闭 / 开启 + e.call(t, "POST", "/api/groups/"+g.Group.ID+"/toggle", map[string]bool{"enabled": false}) + if e.tunnel(t, m1.ID).Enabled || e.tunnel(t, m2.ID).Enabled { + t.Fatal("group toggle should close every member") + } + e.call(t, "POST", "/api/groups/"+g.Group.ID+"/toggle", map[string]bool{"enabled": true}) + if !e.tunnel(t, m1.ID).Enabled || !e.tunnel(t, m2.ID).Enabled { + t.Fatal("group toggle should open every member") + } + + // 解散：成员保留为独立通道 + if status, _ := e.call(t, "DELETE", "/api/groups/"+g.Group.ID, nil); status != http.StatusOK { + t.Fatal("dissolve failed") + } + if e.tunnel(t, m1.ID).GroupID != "" || len(e.cm.Get().Groups) != 0 { + t.Fatal("dissolved group should leave standalone tunnels") + } +} + +func TestGroupFromExistingTunnelsRenameAndMoveOut(t *testing.T) { + e := startEnv(t, testToken) + waitFor(t, "agent connected", e.agent.IsConnected) + var ids []string + var ports []int + for i := 0; i < 2; i++ { + port := freeRelayPort(t, ports...) + _, body := e.call(t, "POST", "/api/tunnels", map[string]any{"name": fmt.Sprint("t", i), "target": "127.0.0.1:1", "listen_port": port}) + var tc protocol.TunnelConfig + _ = json.Unmarshal(body, &tc) + ids, ports = append(ids, tc.ID), append(ports, port) + } + + status, body := e.call(t, "POST", "/api/groups", map[string]any{"name": "支付服务", "tunnel_ids": ids}) + if status != http.StatusCreated { + t.Fatalf("group from existing tunnels: %d %s", status, body) + } + var g groupResp + _ = json.Unmarshal(body, &g) + if e.tunnel(t, ids[0]).GroupID != g.Group.ID || e.tunnel(t, ids[1]).GroupID != g.Group.ID { + t.Fatal("existing tunnels should join the group") + } + _, body = e.call(t, "POST", "/api/groups/"+g.Group.ID+"/invites", map[string]any{}) + var inv struct { + Invite protocol.Invite `json:"invite"` + } + _ = json.Unmarshal(body, &inv) + page := func() string { + resp, err := http.Get(fmt.Sprintf("http://%s/i/%s", e.addr, inv.Invite.Code)) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + return string(b) + } + if p := page(); !strings.Contains(p, "支付服务（2 个地址）") { + t.Fatalf("invite page should list the group: %s", p) + } + + // 改名会同步到服务端 + if status, body := e.call(t, "PUT", "/api/groups/"+g.Group.ID, map[string]any{"name": "支付服务后端"}); status != http.StatusOK { + t.Fatalf("rename: %d %s", status, body) + } + waitFor(t, "new group name on the invite page", func() bool { return strings.Contains(page(), "支付服务后端（2 个地址）") }) + + // 移出组；不能加入不存在的组 + if status, _ := e.call(t, "PUT", "/api/tunnels/"+ids[0], map[string]any{"group_id": ""}); status != http.StatusOK { + t.Fatal("move out failed") + } + waitFor(t, "invite page lists the remaining member", func() bool { return strings.Contains(page(), "（1 个地址）") }) + if status, _ := e.call(t, "PUT", "/api/tunnels/"+ids[0], map[string]any{"group_id": "nope"}); status != http.StatusBadRequest { + t.Fatal("joining an unknown group must fail") + } + + // 一个通道只能属于一个组；组名不能重复 + if status, _ := e.call(t, "POST", "/api/groups", map[string]any{"name": "其它", "tunnel_ids": []string{ids[1]}}); status != http.StatusBadRequest { + t.Fatal("a tunnel cannot be in two groups") + } + if status, _ := e.call(t, "POST", "/api/groups", map[string]any{"name": "支付服务后端", "tunnel_ids": []string{ids[0]}}); status != http.StatusBadRequest { + t.Fatal("duplicate group names must be rejected") + } + + // 解散并删除成员 + if status, _ := e.call(t, "DELETE", "/api/groups/"+g.Group.ID+"?delete_members=1", nil); status != http.StatusOK { + t.Fatal("delete group with members failed") + } + if _, ok := e.cm.FindTunnel(ids[1]); ok { + t.Fatal("members should be deleted with the group") + } + if _, ok := e.cm.FindTunnel(ids[0]); !ok { + t.Fatal("the tunnel moved out earlier must remain") + } +} + +func TestGroupPreviewDoesNotSave(t *testing.T) { + e := startEnv(t, testToken) + status, body := e.call(t, "POST", "/api/groups/preview", map[string]any{"name": "订单服务", "hosts": "10.0.1.11-12", "ports": "22"}) + if status != http.StatusOK { + t.Fatalf("preview: %d %s", status, body) + } + var out struct { + Members []struct { + Name string `json:"name"` + Target string `json:"target"` + ListenPort int `json:"listen_port"` + } `json:"members"` + } + _ = json.Unmarshal(body, &out) + if len(out.Members) != 2 || out.Members[0].Name != "订单服务 (10.0.1.11)" || out.Members[1].ListenPort != out.Members[0].ListenPort+1 { + t.Fatalf("preview: %+v", out.Members) + } + if len(e.cm.Get().Tunnels) != 0 { + t.Fatal("preview must not create tunnels") + } + if status, _ := e.call(t, "POST", "/api/groups/preview", map[string]any{"hosts": "10.0.1.11"}); status != http.StatusBadRequest { + t.Fatal("missing ports should be rejected") + } +} + +func TestGroupIdleCloseIsLinked(t *testing.T) { + e := startEnvWith(t, testToken, func(a *Agent) { a.idleUnit = 3 * time.Second }) + waitFor(t, "agent connected", e.agent.IsConnected) + _, body := e.call(t, "POST", "/api/groups", map[string]any{"name": "g", "hosts": "127.0.0.1", "ports": "1, 2", "idle_close_minutes": 1}) + var g groupResp + _ = json.Unmarshal(body, &g) + a, b := g.Created[0], g.Created[1] + waitFor(t, "members online", func() bool { + return e.tunnel(t, a.ID).Metrics.Status == "online" && e.tunnel(t, b.ID).Metrics.Status == "online" + }) + + // 只给成员 a 发邀请：b 也算在用，不会单独闲置关闭 + _, ib := e.call(t, "POST", "/api/tunnels/"+a.ID+"/invites", map[string]any{}) + var ir struct { + Invite protocol.Invite `json:"invite"` + } + _ = json.Unmarshal(ib, &ir) + time.Sleep(6 * time.Second) + if !e.tunnel(t, b.ID).Enabled { + t.Fatal("a member must stay open while another member is in use") + } + e.call(t, "DELETE", "/api/tunnels/"+a.ID+"/invites/"+ir.Invite.Code, nil) + waitFor(t, "both members closed together", func() bool { return !e.tunnel(t, a.ID).Enabled && !e.tunnel(t, b.ID).Enabled }) + + // 组邀请同样让整组保持开启 + e.call(t, "POST", "/api/groups/"+g.Group.ID+"/toggle", map[string]bool{"enabled": true}) + waitFor(t, "members online again", func() bool { + return e.tunnel(t, a.ID).Metrics.Status == "online" && e.tunnel(t, b.ID).Metrics.Status == "online" + }) + _, gb := e.call(t, "POST", "/api/groups/"+g.Group.ID+"/invites", map[string]any{}) + var gr struct { + Invite protocol.Invite `json:"invite"` + } + _ = json.Unmarshal(gb, &gr) + time.Sleep(6 * time.Second) + if !e.tunnel(t, a.ID).Enabled || !e.tunnel(t, b.ID).Enabled { + t.Fatal("an unused group invite keeps the whole group open") + } + e.call(t, "DELETE", "/api/groups/"+g.Group.ID+"/invites/"+gr.Invite.Code, nil) + waitFor(t, "group closed after its invite was revoked", func() bool { return !e.tunnel(t, a.ID).Enabled && !e.tunnel(t, b.ID).Enabled }) +} diff --git a/cmd/portrelay-agent/web/index.html b/cmd/portrelay-agent/web/index.html index 8a19c62..80ff523 100644 --- a/cmd/portrelay-agent/web/index.html +++ b/cmd/portrelay-agent/web/index.html @@ -247,6 +247,7 @@ background: #1e293b; border: 1px solid var(--card-border); border-radius: 16px; width: 520px; max-width: 90vw; padding: 24px; box-shadow: 0 20px 40px rgba(0, 0, 0, 0.5); + max-height: 90vh; overflow-y: auto; } .modal-header { display: flex; justify-content: space-between; align-items: center; margin-bottom: 20px; @@ -301,6 +302,32 @@ padding: 24px; display: flex; flex-direction: column; } .drawer.open { right: 0; } + + /* 通道组 */ + .group-icon { color: #38bdf8; font-size: 16px; } + .group-summary { display: flex; align-items: center; gap: 12px; flex-wrap: wrap; font-size: 13px; color: var(--text-muted); } + .group-summary .route-endpoint { font-family: ui-monospace, Menlo, Monaco, Consolas, monospace; } + .member-list { border-top: 1px solid rgba(148, 163, 184, 0.14); margin-bottom: 14px; } + .member-row { display: flex; align-items: center; gap: 10px; padding: 8px 0; border-bottom: 1px solid rgba(148, 163, 184, 0.1); font-size: 13px; } + .member-row.off { opacity: 0.6; } + .member-route { display: flex; align-items: center; gap: 8px; font-family: ui-monospace, Menlo, Monaco, Consolas, monospace; white-space: nowrap; } + .member-route .route-arrow { font-size: 13px; } + .member-name { color: var(--text-muted); font-size: 12px; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; max-width: 200px; } + .member-meta { margin-left: auto; color: var(--text-muted); font-size: 12px; white-space: nowrap; font-variant-numeric: tabular-nums; } + .member-hint { color: var(--text-muted); font-size: 12px; padding: 8px 0; } + .member-row .chip { font-size: 11px; font-weight: 600; padding: 1px 7px; border-radius: 5px; cursor: pointer; background: rgba(52, 211, 153, 0.12); color: #34d399; } + .mdot { width: 7px; height: 7px; border-radius: 50%; flex-shrink: 0; background: #64748b; } + .mdot.on { background: #34d399; } .mdot.err { background: #f87171; } .mdot.wait { background: #fbbf24; } + .member-actions { display: flex; gap: 2px; } + .icon-btn { background: none; border: 0; cursor: pointer; font-size: 13px; padding: 2px 4px; border-radius: 4px; opacity: 0.75; } + .icon-btn:hover:not(:disabled) { background: rgba(255, 255, 255, 0.08); opacity: 1; } + .icon-btn:disabled { opacity: 0.25; cursor: default; } + .check-list { max-height: 140px; overflow-y: auto; display: flex; flex-direction: column; gap: 4px; } + .form-group label.check-item { display: flex; align-items: center; gap: 8px; margin: 0; font-size: 13px; font-weight: normal; color: var(--text); cursor: pointer; } + .check-item input { width: auto !important; } + .check-item code { color: var(--text-muted); font-size: 12px; } + .preview-row { font-family: ui-monospace, Menlo, Monaco, Consolas, monospace; font-size: 12px; } + @media (max-width: 860px) { .member-name, .member-meta { display: none; } } .drawer-header { display: flex; justify-content: space-between; align-items: center; margin-bottom: 20px; } .drawer-content { flex: 1; overflow-y: auto; } @@ -321,6 +348,7 @@
正在连接服务器... + @@ -413,6 +441,10 @@ +