feat: 通道组——一次开放多台主机 / 多个端口,共用邀请与授权

- 批量创建:主机支持 IP、主机名、末段范围(10.0.1.11-13)和 主机:端口,
  与端口列表展开为 主机 × 端口,自动分配连续的访问端口,创建前可预览
- 授权以组为单位:一条邀请链接、一次审批即可访问组内全部地址;
  计时从首次连上任一地址开始,撤销一次全组断开;后加入组的通道自动获得组授权
- 网关授权状态按作用范围(通道 / 通道组)存储,通道的动态白名单统一由状态库重新计算,
  通道授权与组授权互不干扰;组内最后一个通道关闭或组被解散时,组的邀请与授权失效
- 闲置联动:组内任一地址在用时整组都不算闲置;支持整组开关
- 已有通道可并入组或移出组;组可改名、追加目标、解散(保留或删除成员)
- App 与网页面板:组卡片、新建 / 编辑通道组、组邀请与“连接与授权”,授权页列出全部地址
- 测试:网关 9 个、Agent 5 个通道组用例
This commit is contained in:
DeepFal committed 2026-10-13 20:16:43 +08:00
1 parent 86e159b3f0
commit 3797bef88a
22 files changed
+3894 -471

No files matched your search

+70 -14
View File
@@ -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
+186 -77
View File
@@ -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
+277
View File
@@ -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
}
+209
View File
@@ -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)
}
+250
View File
@@ -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()
}
+339
View File
@@ -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 })
}