From 46bfb86c1bb909ee7a93d3852997310ad74e7320 Mon Sep 17 00:00:00 2001 From: sagit <36596628+Sagit-chu@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:03:07 +0800 Subject: [PATCH] fix(user): revoke tunnel access when disabling users (#568) Revoke tunnel grants and pause each forwarding entry without self-healing missing services. Continue healthy nodes and other forwards after failures, aggregate pending nodes for retry, and show persistent partial-disable feedback. Related: #562 --- .../internal/http/handler/control_plane.go | 89 ++- .../internal/http/handler/flow_policy.go | 43 ++ go-backend/internal/http/handler/mutations.go | 16 + .../internal/http/handler/nftables_runtime.go | 10 +- .../http/handler/user_disable_test.go | 527 ++++++++++++++++++ .../store/repo/repository_mutations.go | 7 + .../contract/user_disable_contract_test.go | 87 +++ vite-frontend/src/pages/user.tsx | 28 +- 8 files changed, 775 insertions(+), 32 deletions(-) create mode 100644 go-backend/internal/http/handler/user_disable_test.go create mode 100644 go-backend/tests/contract/user_disable_contract_test.go diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 2498acd..110ffae 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -534,7 +534,22 @@ func buildForwardServiceDeleteNames(bases []string) []string { return names } +type forwardServiceControlOptions struct { + tolerateNotFound bool + tolerateOffline bool + allowSelfHeal bool + continueOnError bool +} + func (h *Handler) controlForwardServices(forward *forwardRecord, commandType string, tolerateNotFound bool) error { + return h.controlForwardServicesWithOptions(forward, commandType, forwardServiceControlOptions{ + tolerateNotFound: tolerateNotFound, + tolerateOffline: true, + allowSelfHeal: true, + }) +} + +func (h *Handler) controlForwardServicesWithOptions(forward *forwardRecord, commandType string, opts forwardServiceControlOptions) error { if h == nil || forward == nil { return errors.New("invalid forward control context") } @@ -561,6 +576,7 @@ func (h *Handler) controlForwardServices(forward *forwardRecord, commandType str candidateTunnelIDs = append(candidateTunnelIDs, userTunnelIDs...) candidateTunnelIDs = append(candidateTunnelIDs, allUserTunnelIDs...) bases := buildForwardServiceBaseCandidates(forward.ID, forward.UserID, userTunnelID, candidateTunnelIDs) + var nodeErrors []error if strings.EqualFold(strings.TrimSpace(commandType), "DeleteService") { seen := map[int64]struct{}{} for _, fp := range ports { @@ -569,16 +585,20 @@ func (h *Handler) controlForwardServices(forward *forwardRecord, commandType str } seen[fp.NodeID] = struct{}{} if err := h.deleteForwardServiceBasesOnNode(fp.NodeID, bases); err != nil { - if isNodeOfflineOrTimeoutError(err) { + if opts.tolerateOffline && isNodeOfflineOrTimeoutError(err) { continue } - if tolerateNotFound && isNotFoundError(err) { + if opts.tolerateNotFound && isNotFoundError(err) { + continue + } + if opts.continueOnError { + nodeErrors = append(nodeErrors, h.nodeCommandError(fp.NodeID, err)) continue } return err } } - return nil + return errors.Join(nodeErrors...) } seen := map[int64]struct{}{} healed := false @@ -588,37 +608,50 @@ func (h *Handler) controlForwardServices(forward *forwardRecord, commandType str } seen[fp.NodeID] = struct{}{} - nodeHandled, lastNotFoundErr, err := h.controlForwardServicesOnNode(fp.NodeID, bases, commandType) - if err != nil { - if isNodeOfflineOrTimeoutError(err) { - continue - } - return err - } - - if !nodeHandled && lastNotFoundErr != nil && !healed && shouldSelfHealForwardServiceControl(commandType) { - if healErr := h.syncForwardServices(forward, "UpdateService", true); healErr != nil { - return healErr - } - healed = true - nodeHandled, lastNotFoundErr, err = h.controlForwardServicesOnNode(fp.NodeID, bases, commandType) + nodeErr := func() error { + nodeHandled, lastNotFoundErr, err := h.controlForwardServicesOnNode(fp.NodeID, bases, commandType) if err != nil { + if opts.tolerateOffline && isNodeOfflineOrTimeoutError(err) { + return nil + } return err } - } - if nodeHandled { - continue + if opts.allowSelfHeal && !nodeHandled && lastNotFoundErr != nil && !healed && shouldSelfHealForwardServiceControl(commandType) { + if healErr := h.syncForwardServices(forward, "UpdateService", true); healErr != nil { + return healErr + } + healed = true + nodeHandled, lastNotFoundErr, err = h.controlForwardServicesOnNode(fp.NodeID, bases, commandType) + if err != nil { + return err + } + } + + if nodeHandled || opts.tolerateNotFound { + return nil + } + if lastNotFoundErr != nil { + return lastNotFoundErr + } + return errors.New("service control failed") + }() + if nodeErr != nil { + if !opts.continueOnError { + return nodeErr + } + nodeErrors = append(nodeErrors, h.nodeCommandError(fp.NodeID, nodeErr)) } - if tolerateNotFound { - continue - } - if lastNotFoundErr != nil { - return lastNotFoundErr - } - return errors.New("service control failed") } - return nil + return errors.Join(nodeErrors...) +} + +func (h *Handler) nodeCommandError(nodeID int64, err error) error { + name := strconv.FormatInt(nodeID, 10) + if node, lookupErr := h.getNodeRecord(nodeID); lookupErr == nil && node != nil { + name = defaultString(strings.TrimSpace(node.Name), name) + } + return fmt.Errorf("节点 %s (ID %d): %w", name, nodeID, err) } func (h *Handler) controlForwardServicesOnNode(nodeID int64, bases []string, commandType string) (bool, error, error) { diff --git a/go-backend/internal/http/handler/flow_policy.go b/go-backend/internal/http/handler/flow_policy.go index d85b67f..243e08e 100644 --- a/go-backend/internal/http/handler/flow_policy.go +++ b/go-backend/internal/http/handler/flow_policy.go @@ -3,6 +3,7 @@ package handler import ( "encoding/json" "errors" + "fmt" "log" "math" "strconv" @@ -393,6 +394,48 @@ func (h *Handler) getUserTunnelPolicy(userTunnelID int64) (*userTunnelPolicy, er }, nil } +// disableUserAccess retains unconfirmed forwards as active so a repeated disable +// can retry them. Login and tunnel access remain disabled when a node is offline. +func (h *Handler) disableUserAccess(userID int64, now int64) error { + permissionErr := h.repo.DisableUserTunnels(userID) + forwards, err := h.listActiveForwardsByUser(userID) + if err != nil { + return errors.Join(permissionErr, err) + } + errs := []error{permissionErr} + for i := range forwards { + forward := &forwards[i] + nftMode, entryNodeIDs, pauseErr := h.tunnelUsesNftables(forward.TunnelID) + if pauseErr == nil { + if nftMode { + // Exclude the rule from the desired plan before marking its pause + // confirmed. A failed SSH apply must remain eligible for retry. + pauseErr = h.syncNftablesNode(entryNodeIDs[0], forward.ID) + if pauseErr != nil { + pauseErr = h.nodeCommandError(entryNodeIDs[0], pauseErr) + } + } else { + // Missing services are already stopped. Self-healing would recreate + // services on every entry and could restart one already paused. + pauseErr = h.controlForwardServicesWithOptions(forward, "PauseService", forwardServiceControlOptions{ + tolerateNotFound: true, + tolerateOffline: false, + allowSelfHeal: false, + continueOnError: true, + }) + } + } + if pauseErr != nil { + errs = append(errs, fmt.Errorf("转发 %d 暂停失败: %w", forward.ID, pauseErr)) + continue + } + if err := h.repo.UpdateForwardStatus(forward.ID, 0, now); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + func (h *Handler) pauseUserForwards(userID int64, now int64) { forwards, err := h.listActiveForwardsByUser(userID) if err != nil { diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 4e57ea8..222f78d 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -205,6 +205,15 @@ func (h *Handler) userUpdate(w http.ResponseWriter, r *http.Request) { } } + if status != 1 { + // Revoke access before optional quota/group updates can fail or return. + if err := h.disableUserAccess(id, now); err != nil { + result := response.Err(-2, fmt.Sprintf("账号已禁用，以下权限或节点暂停未确认；其他转发已继续处理，请排查后重试禁用：\n%v", err)) + result.Data = map[string]interface{}{"userDisabled": true} + response.WriteJSON(w, result) + return + } + } h.repo.PropagateUserFlowToTunnels(id, flow, num, expTime, flowResetTime, flowMiB) if hasDailyQuota || hasMonthlyQuota { dailyQuotaGB := asInt64(req["dailyQuotaGB"], 0) @@ -243,6 +252,13 @@ func (h *Handler) userUpdate(w http.ResponseWriter, r *http.Request) { } } } + if status != 1 { + // Group changes may have created additional tunnel grants. + if err := h.repo.DisableUserTunnels(id); err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + } if oldUser.MaxConn != maxConn { warnings, syncErr := h.syncUserMaxConnForwards(id) if syncErr != nil { diff --git a/go-backend/internal/http/handler/nftables_runtime.go b/go-backend/internal/http/handler/nftables_runtime.go index d440d51..f5ccf39 100644 --- a/go-backend/internal/http/handler/nftables_runtime.go +++ b/go-backend/internal/http/handler/nftables_runtime.go @@ -7,6 +7,7 @@ import ( "fmt" "net" "net/http" + "slices" "strings" "time" @@ -150,7 +151,7 @@ func (h *Handler) validateNftablesTunnelStateTx(tx *gorm.DB, entryNodeIDs []int6 return nil } -func (h *Handler) buildNftablesNodePlan(nodeID int64) (runtimenft.NodePlan, *model.NodeSSHConfig, error) { +func (h *Handler) buildNftablesNodePlan(nodeID int64, excludedForwardIDs ...int64) (runtimenft.NodePlan, *model.NodeSSHConfig, error) { cfg, err := h.repo.GetNodeSSHConfig(nodeID) if err != nil { return runtimenft.NodePlan{}, nil, err @@ -162,6 +163,9 @@ func (h *Handler) buildNftablesNodePlan(nodeID int64) (runtimenft.NodePlan, *mod plan := runtimenft.NodePlan{NodeID: nodeID, Rules: make([]runtimenft.Rule, 0, len(forwards))} for i := range forwards { forward := &forwards[i] + if slices.Contains(excludedForwardIDs, forward.ID) { + continue + } tunnel, err := h.getTunnelRecord(forward.TunnelID) if err != nil || tunnel == nil || tunnel.Status != 1 { continue @@ -201,11 +205,11 @@ func (h *Handler) buildNftablesNodePlan(nodeID int64) (runtimenft.NodePlan, *mod return plan, cfg, nil } -func (h *Handler) syncNftablesNode(nodeID int64) error { +func (h *Handler) syncNftablesNode(nodeID int64, excludedForwardIDs ...int64) error { if h == nil || h.repo == nil { return errors.New("handler not initialized") } - plan, cfgModel, err := h.buildNftablesNodePlan(nodeID) + plan, cfgModel, err := h.buildNftablesNodePlan(nodeID, excludedForwardIDs...) if err != nil { return err } diff --git a/go-backend/internal/http/handler/user_disable_test.go b/go-backend/internal/http/handler/user_disable_test.go new file mode 100644 index 0000000..9acef8c --- /dev/null +++ b/go-backend/internal/http/handler/user_disable_test.go @@ -0,0 +1,527 @@ +package handler + +import ( + "bytes" + "database/sql" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "go-backend/internal/http/response" + "go-backend/internal/store/model" + "go-backend/internal/ws" +) + +func TestUserUpdateDisablesPermissionsAndRunningForwards(t *testing.T) { + for _, password := range []string{"", "replacement-password"} { + t.Run(fmt.Sprintf("password_changed=%t", password != ""), func(t *testing.T) { + a := newCleanupAgent(t) + now := time.Now().UnixMilli() + for _, userID := range []int64{2, 3} { + if err := a.h.repo.DB().Create(&model.User{ID: userID, User: fmt.Sprintf("user%d", userID), Pwd: "pwd", RoleID: 1, Status: 1, ExpTime: now + 86400000, Flow: 100, Num: 10}).Error; err != nil { + t.Fatal(err) + } + } + if err := a.h.repo.DB().Create(&model.Tunnel{ID: 1, Name: "test", Type: 1, Protocol: "tls", Status: 1, Flow: 1}).Error; err != nil { + t.Fatal(err) + } + for _, permission := range []model.UserTunnel{ + {ID: 10, UserID: 2, TunnelID: 1, Status: 1, Flow: 100, Num: 10}, + {ID: 11, UserID: 2, TunnelID: 2, Status: 0, Flow: 100, Num: 10}, + {ID: 12, UserID: 3, TunnelID: 1, Status: 1, Flow: 100, Num: 10}, + } { + if err := a.h.repo.DB().Create(&permission).Error; err != nil { + t.Fatal(err) + } + } + for _, forward := range []model.Forward{ + {ID: 20, UserID: 2, UserName: "user2", Name: "running", TunnelID: 1, RemoteAddr: "127.0.0.1:80", Status: 1}, + {ID: 21, UserID: 2, UserName: "user2", Name: "paused", TunnelID: 1, RemoteAddr: "127.0.0.1:81", Status: 0}, + {ID: 22, UserID: 3, UserName: "user3", Name: "other-user", TunnelID: 1, RemoteAddr: "127.0.0.1:82", Status: 1}, + } { + if err := a.h.repo.DB().Create(&forward).Error; err != nil { + t.Fatal(err) + } + if err := a.h.repo.DB().Create(&model.ForwardPort{ForwardID: forward.ID, NodeID: 1, Port: 31000 + int(forward.ID)}).Error; err != nil { + t.Fatal(err) + } + } + + update := func(status int) { + t.Helper() + payload, err := json.Marshal(map[string]interface{}{"id": 2, "user": "user2", "status": status, "pwd": password, "flow": 100, "num": 10, "expTime": now + 86400000}) + if err != nil { + t.Fatal(err) + } + res := httptest.NewRecorder() + a.h.userUpdate(res, httptest.NewRequest(http.MethodPost, "/api/v1/user/update", bytes.NewReader(payload))) + var result response.R + if err := json.Unmarshal(res.Body.Bytes(), &result); err != nil { + t.Fatal(err) + } + if result.Code != 0 { + t.Fatalf("update user: %d %s", result.Code, result.Msg) + } + } + + update(1) + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM user_tunnel WHERE id = 10"); got != 1 { + t.Errorf("ordinary account update disabled an enabled permission: %d", got) + } + if len(a.commandsOfType("PauseService")) != 0 { + t.Error("ordinary account update paused services") + } + update(0) + for _, check := range []struct { + table string + id int64 + want int + }{{"user", 2, 0}, {"user_tunnel", 10, 0}, {"user_tunnel", 11, 0}, {"user_tunnel", 12, 1}, {"forward", 20, 0}, {"forward", 21, 0}, {"forward", 22, 1}} { + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM "+check.table+" WHERE id = ?", check.id); got != check.want { + t.Errorf("%s %d status = %d, want %d", check.table, check.id, got, check.want) + } + } + commands := a.commandsOfType("PauseService") + if len(commands) == 0 { + t.Error("running forward never received PauseService") + } + for _, command := range commands { + var data struct { + Services []string `json:"services"` + } + if err := json.Unmarshal(command.Data, &data); err != nil { + t.Fatal(err) + } + for _, name := range data.Services { + if !strings.HasPrefix(name, "20_2_") { + t.Errorf("paused unrelated service %q", name) + } + } + } + update(0) + if len(a.commandsOfType("PauseService")) != len(commands) { + t.Error("repeated disable replayed pauses for already paused forwards") + } + + // Account edits and re-enabling must not grant disabled permissions or + // resume forwards that an administrator previously paused. + update(1) + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM user_tunnel WHERE id = 10"); got != 0 { + t.Errorf("account re-enable unexpectedly granted a disabled permission: %d", got) + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM user_tunnel WHERE id = 11"); got != 0 { + t.Errorf("re-enabled independently disabled permission: %d", got) + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM forward WHERE id = 21"); got != 0 { + t.Errorf("resumed previously paused forward: %d", got) + } + if len(a.commandsOfType("ResumeService")) != 0 { + t.Error("account re-enable unexpectedly resumed services") + } + }) + } +} + +func TestUserDisableMultiEntryNeverRecreatesMissingServices(t *testing.T) { + for _, secondNodeState := range []string{"missing", "offline", "timeout"} { + t.Run(secondNodeState, func(t *testing.T) { + a := newCleanupAgent(t) + var remoteMu sync.Mutex + remoteServices := make(map[string]bool) + var remoteCommands []string + remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var command struct { + Type string `json:"commandType"` + Data json.RawMessage `json:"data"` + } + if err := json.NewDecoder(r.Body).Decode(&command); err != nil { + t.Error(err) + return + } + remoteMu.Lock() + defer remoteMu.Unlock() + remoteCommands = append(remoteCommands, command.Type) + switch command.Type { + case "UpdateService", "AddService": + var services []struct { + Name string `json:"name"` + } + if err := json.Unmarshal(command.Data, &services); err != nil { + t.Error(err) + return + } + for _, service := range services { + remoteServices[service.Name] = true + } + case "PauseService": + if secondNodeState != "missing" { + message := "节点不在线" + if secondNodeState == "timeout" { + message = "等待节点响应超时" + } + response.WriteJSON(w, response.ErrDefault(message)) + return + } + var data struct { + Services []string `json:"services"` + } + if err := json.Unmarshal(command.Data, &data); err != nil { + t.Error(err) + return + } + for _, name := range data.Services { + if _, exists := remoteServices[name]; !exists { + response.WriteJSON(w, response.OK(map[string]interface{}{"success": false, "message": "service not found"})) + return + } + remoteServices[name] = false + } + } + response.WriteJSON(w, response.OK(map[string]interface{}{"success": true, "type": command.Type})) + })) + defer remote.Close() + now := time.Now().UnixMilli() + userID, err := a.h.repo.CreateUser("multi-entry-disable", "pwd", 1, now+86400000, 100, 1, 10, 1, 0, now) + if err != nil { + t.Fatal(err) + } + for _, value := range []interface{}{ + &model.Node{ID: 2, Name: "second-entry", Secret: "test", ServerIP: "127.0.0.1", Status: 1, IsRemote: 1, RemoteURL: sql.NullString{String: remote.URL, Valid: true}, RemoteToken: sql.NullString{String: "test-token", Valid: true}}, + &model.Tunnel{ID: 1, Name: "multi-entry", Type: 1, Status: 1, Protocol: "tls", Flow: 1}, + &model.ChainTunnel{TunnelID: 1, NodeID: 1, ChainType: "1"}, + &model.ChainTunnel{TunnelID: 1, NodeID: 2, ChainType: "1"}, + &model.UserTunnel{ID: 10, UserID: userID, TunnelID: 1, Status: 1, Flow: 100, Num: 10}, + &model.Forward{ID: 20, UserID: userID, UserName: "multi-entry-disable", Name: "running", TunnelID: 1, RemoteAddr: "127.0.0.1:80", Status: 1}, + &model.ForwardPort{ForwardID: 20, NodeID: 1, Port: 31020}, + &model.ForwardPort{ForwardID: 20, NodeID: 2, Port: 31020}, + } { + if err := a.h.repo.DB().Create(value).Error; err != nil { + t.Fatal(err) + } + } + res := postJSONToHandler(t, a.h.userUpdate, map[string]interface{}{"id": userID, "user": "multi-entry-disable", "status": 0}) + var out response.R + if err := json.Unmarshal(res.Body.Bytes(), &out); err != nil { + t.Fatal(err) + } + wantStatus := 1 + if secondNodeState == "missing" { + wantStatus = 0 + if out.Code != 0 { + t.Errorf("missing service should count as stopped: %s", out.Msg) + } + } else if out.Code == 0 { + t.Error("unconfirmed second entry pause returned success") + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM forward WHERE id = 20"); got != wantStatus { + t.Errorf("forward status = %d, want %d", got, wantStatus) + } + a.mu.Lock() + commands := append([]cleanupAgentCommand(nil), a.commands...) + a.mu.Unlock() + firstEntryRunning := true + for _, command := range commands { + switch command.Type { + case "PauseService": + firstEntryRunning = false + case "UpdateService", "AddService", "ResumeService": + firstEntryRunning = true + t.Errorf("disable sent %s to already paused entry A", command.Type) + } + } + if firstEntryRunning { + t.Error("entry A is running after disable") + } + remoteMu.Lock() + defer remoteMu.Unlock() + if len(remoteCommands) == 0 { + t.Error("entry B was never checked") + } + for _, command := range remoteCommands { + if command != "PauseService" { + t.Errorf("disable sent %s to entry B", command) + } + } + for name, running := range remoteServices { + if running { + t.Errorf("entry B service %s is still running", name) + } + } + }) + } +} + +func TestUserDisableContinuesAcrossNodesAndForwards(t *testing.T) { + for _, failure := range []string{"节点不在线", "等待节点响应超时", "pause rejected"} { + for _, failedNodeFirst := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/failed_node_first=%t", failure, failedNodeFirst), func(t *testing.T) { + a := newCleanupAgent(t) + var mu sync.Mutex + recovered := false + attempts := make(map[string]int) + remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var command struct { + Type string `json:"commandType"` + } + if err := json.NewDecoder(r.Body).Decode(&command); err != nil { + t.Error(err) + return + } + if command.Type != "PauseService" { + t.Errorf("disable sent %s", command.Type) + } + mu.Lock() + attempts[r.Header.Get("Authorization")]++ + isRecovered := recovered + mu.Unlock() + if !isRecovered { + response.WriteJSON(w, response.ErrDefault(failure)) + return + } + response.WriteJSON(w, response.OK(map[string]interface{}{"success": true, "type": command.Type})) + })) + defer remote.Close() + now := time.Now().UnixMilli() + userID, err := a.h.repo.CreateUser("continue-disable", "pwd", 1, now+86400000, 100, 1, 10, 1, 0, now) + if err != nil { + t.Fatal(err) + } + for _, value := range []interface{}{ + &model.Tunnel{ID: 1, Name: "multi-entry", Type: 2, Status: 1, Flow: 1}, + &model.Node{ID: 4, Name: "offline-middle", Secret: "test", ServerIP: "127.0.0.1", Status: 0}, + &model.ChainTunnel{TunnelID: 1, NodeID: 4, ChainType: "2"}, + &model.UserTunnel{ID: 10, UserID: userID, TunnelID: 1, Status: 1, Flow: 100, Num: 10}, + &model.Forward{ID: 20, UserID: userID, UserName: "continue-disable", Name: "mixed-nodes", TunnelID: 1, RemoteAddr: "127.0.0.1:80", Status: 1}, + &model.Forward{ID: 21, UserID: userID, UserName: "continue-disable", Name: "healthy-forward", TunnelID: 1, RemoteAddr: "127.0.0.1:81", Status: 1}, + } { + if err := a.h.repo.DB().Create(value).Error; err != nil { + t.Fatal(err) + } + } + for _, nodeID := range []int64{2, 3} { + node := model.Node{ID: nodeID, Name: fmt.Sprintf("failed-node-%d", nodeID), Secret: "test", ServerIP: "127.0.0.1", Status: 1, IsRemote: 1, RemoteURL: sql.NullString{String: remote.URL, Valid: true}, RemoteToken: sql.NullString{String: fmt.Sprintf("test-node-%d", nodeID), Valid: true}} + if err := a.h.repo.DB().Create(&node).Error; err != nil { + t.Fatal(err) + } + } + nodeOrder := []int64{1, 2, 3} + if failedNodeFirst { + nodeOrder = []int64{2, 1, 3} + } + for _, nodeID := range nodeOrder { + if err := a.h.repo.DB().Create(&model.ForwardPort{ForwardID: 20, NodeID: nodeID, Port: 31020}).Error; err != nil { + t.Fatal(err) + } + if err := a.h.repo.DB().Create(&model.ChainTunnel{TunnelID: 1, NodeID: nodeID, ChainType: "1"}).Error; err != nil { + t.Fatal(err) + } + } + if err := a.h.repo.DB().Create(&model.ForwardPort{ForwardID: 21, NodeID: 1, Port: 31021}).Error; err != nil { + t.Fatal(err) + } + update := func() response.R { + t.Helper() + res := postJSONToHandler(t, a.h.userUpdate, map[string]interface{}{"id": userID, "user": "continue-disable", "status": 0}) + var out response.R + if err := json.Unmarshal(res.Body.Bytes(), &out); err != nil { + t.Fatal(err) + } + return out + } + out := update() + if out.Code == 0 { + t.Error("partial node failure reported complete success") + } + state, ok := out.Data.(map[string]interface{}) + if !ok || state["userDisabled"] != true { + t.Errorf("partial response did not tell the UI the account is disabled: %#v", out.Data) + } + for _, nodeID := range []int64{2, 3} { + if !strings.Contains(out.Msg, fmt.Sprintf("failed-node-%d", nodeID)) || !strings.Contains(out.Msg, fmt.Sprintf("ID %d", nodeID)) { + t.Errorf("failure response did not identify node %d: %s", nodeID, out.Msg) + } + mu.Lock() + count := attempts[fmt.Sprintf("Bearer test-node-%d", nodeID)] + mu.Unlock() + if count == 0 { + t.Errorf("failure on another node prevented node %d from being attempted", nodeID) + } + } + paused := make(map[string]bool) + for _, command := range a.commandsOfType("PauseService") { + var data struct { + Services []string `json:"services"` + } + if err := json.Unmarshal(command.Data, &data); err != nil { + t.Fatal(err) + } + for _, name := range data.Services { + paused[strings.Split(name, "_")[0]] = true + } + } + for _, id := range []string{"20", "21"} { + if !paused[id] { + t.Errorf("healthy node never paused forward %s", id) + } + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM forward WHERE id = 20"); got != 1 { + t.Errorf("unconfirmed forward status = %d", got) + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM forward WHERE id = 21"); got != 0 { + t.Errorf("healthy later forward was not paused: %d", got) + } + mu.Lock() + recovered = true + mu.Unlock() + if retry := update(); retry.Code != 0 { + t.Errorf("retry after recovery failed: %s", retry.Msg) + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM forward WHERE id = 20"); got != 0 { + t.Errorf("retry did not confirm all nodes: %d", got) + } + for _, kind := range []string{"UpdateService", "AddService", "ResumeService"} { + if len(a.commandsOfType(kind)) != 0 { + t.Errorf("disable sent %s", kind) + } + } + }) + } + } +} + +func TestUserDisableRemovesNftablesRuleOnlyAfterConfirmedApply(t *testing.T) { + fixture := setupNftablesHandler(t) + h := fixture.handler + t.Cleanup(func() { _ = h.repo.Close() }) + seedNftablesSSHConfig(t, h, fixture.nodeID) + tunnelID := seedTunnelForNftables(t, h, "disable-nft", fixture.nodeID) + forward := seedForwardForNftables(t, h, tunnelID, fixture.nodeID, "203.0.113.9:8080") + now := time.Now().UnixMilli() + userID, err := h.repo.CreateUser("disable-nft", "pwd", 1, now+86400000, 100, 1, 10, 1, 0, now) + if err != nil { + t.Fatal(err) + } + if err := h.repo.DB().Model(&model.Forward{}).Where("id = ?", forward.ID).Update("user_id", userID).Error; err != nil { + t.Fatal(err) + } + if err := h.repo.DB().Create(&model.UserTunnel{UserID: userID, TunnelID: tunnelID, Status: 1, Flow: 100, Num: 10}).Error; err != nil { + t.Fatal(err) + } + otherForwardID := forward.ID + 100 + for _, value := range []interface{}{ + &model.Forward{ID: otherForwardID, UserID: 1, UserName: "admin", Name: "other-user", TunnelID: tunnelID, RemoteAddr: "203.0.113.9:8081", Status: 1}, + &model.ForwardPort{ForwardID: otherForwardID, NodeID: fixture.nodeID, Port: 20001}, + } { + if err := h.repo.DB().Create(value).Error; err != nil { + t.Fatal(err) + } + } + manager := &fakeNftablesManager{reconcileErr: errors.New("ssh unavailable")} + h.nftablesManager = manager + update := func() response.R { + t.Helper() + res := postJSONToHandler(t, h.userUpdate, map[string]interface{}{"id": userID, "user": "disable-nft", "status": 0}) + var out response.R + if err := json.Unmarshal(res.Body.Bytes(), &out); err != nil { + t.Fatal(err) + } + return out + } + if out := update(); out.Code == 0 { + t.Error("failed SSH apply returned success") + } + if got := mustQueryInt(t, h.repo, "SELECT status FROM forward WHERE id = ?", forward.ID); got != 1 { + t.Errorf("unconfirmed nftables pause marked completed: %d", got) + } + if manager.reconcileCount() != 1 || len(manager.lastPlan.Rules) != 1 || manager.lastPlan.Rules[0].ForwardID != otherForwardID { + t.Fatalf("did not attempt to remove disabled user's nftables rule: %+v", manager.lastPlan) + } + manager.reconcileErr = nil + if out := update(); out.Code != 0 { + t.Fatalf("retry nftables disable: %s", out.Msg) + } + if got := mustQueryInt(t, h.repo, "SELECT status FROM forward WHERE id = ?", forward.ID); got != 0 { + t.Errorf("confirmed nftables pause was not stored: %d", got) + } + if manager.reconcileCount() != 2 || len(manager.lastPlan.Rules) != 1 || manager.lastPlan.Rules[0].ForwardID != otherForwardID { + t.Errorf("retry did not apply a plan without disabled forward: %+v", manager.lastPlan) + } + if got := mustQueryInt(t, h.repo, "SELECT status FROM forward WHERE id = ?", otherForwardID); got != 1 { + t.Errorf("disabled another user's nftables forward: %d", got) + } +} + +func TestUserDisableReportsPauseFailureAndCanRetry(t *testing.T) { + for _, offline := range []bool{false, true} { + t.Run(fmt.Sprintf("offline=%t", offline), func(t *testing.T) { + a := newCleanupAgent(t) + now := time.Now().UnixMilli() + userID, err := a.h.repo.CreateUser("disable-retry", "pwd", 1, now+86400000, 100, 1, 10, 1, 0, now) + if err != nil { + t.Fatal(err) + } + for _, value := range []interface{}{ + &model.Tunnel{ID: 1, Name: "test", Type: 1, Status: 1, Flow: 1}, + &model.UserTunnel{ID: 10, UserID: userID, TunnelID: 1, Status: 1, Flow: 100, Num: 10}, + &model.Forward{ID: 20, UserID: userID, UserName: "disable-retry", Name: "running", TunnelID: 1, RemoteAddr: "127.0.0.1:80", Status: 1}, + &model.ForwardPort{ForwardID: 20, NodeID: 1, Port: 31020}, + } { + if err := a.h.repo.DB().Create(value).Error; err != nil { + t.Fatal(err) + } + } + onlineServer := a.h.wsServer + if offline { + a.h.wsServer = ws.NewServer(a.h.repo, "test-jwt-secret") + } else { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + response.WriteJSON(w, response.ErrDefault("pause rejected")) + })) + defer ts.Close() + if err := a.h.repo.DB().Model(&model.Node{}).Where("id = 1").Updates(map[string]interface{}{"is_remote": 1, "remote_url": ts.URL, "remote_token": "test-token"}).Error; err != nil { + t.Fatal(err) + } + } + update := func() response.R { + t.Helper() + res := postJSONToHandler(t, a.h.userUpdate, map[string]interface{}{"id": userID, "user": "disable-retry", "status": 0}) + var out response.R + if err := json.Unmarshal(res.Body.Bytes(), &out); err != nil { + t.Fatal(err) + } + return out + } + if out := update(); out.Code == 0 { + t.Error("pause failure returned success") + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM user WHERE id = ?", userID); got != 0 { + t.Errorf("failed pause restored account access: %d", got) + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM user_tunnel WHERE id = 10"); got != 0 { + t.Errorf("failed pause restored tunnel access: %d", got) + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM forward WHERE id = 20"); got != 1 { + t.Errorf("unconfirmed pause marked completed, status = %d", got) + } + a.h.wsServer = onlineServer + if err := a.h.repo.DB().Model(&model.Node{}).Where("id = 1").Update("is_remote", 0).Error; err != nil { + t.Fatal(err) + } + if out := update(); out.Code != 0 { + t.Fatalf("retry failed: %s", out.Msg) + } + if got := mustQueryInt(t, a.h.repo, "SELECT status FROM forward WHERE id = 20"); got != 0 { + t.Errorf("retry did not pause forward: %d", got) + } + if len(a.commandsOfType("PauseService")) == 0 { + t.Error("retry never sent PauseService") + } + }) + } +} diff --git a/go-backend/internal/store/repo/repository_mutations.go b/go-backend/internal/store/repo/repository_mutations.go index 9755108..5bf5556 100644 --- a/go-backend/internal/store/repo/repository_mutations.go +++ b/go-backend/internal/store/repo/repository_mutations.go @@ -151,6 +151,13 @@ func (r *Repository) PropagateUserFlowToTunnels(userID int64, flow int64, num in }).Error } +func (r *Repository) DisableUserTunnels(userID int64) error { + if r == nil || r.db == nil { + return errors.New("repository not initialized") + } + return r.db.Model(&model.UserTunnel{}).Where("user_id = ?", userID).Update("status", 0).Error +} + func (r *Repository) DeleteUserCascade(userID int64) error { if r == nil || r.db == nil { return errors.New("repository not initialized") diff --git a/go-backend/tests/contract/user_disable_contract_test.go b/go-backend/tests/contract/user_disable_contract_test.go new file mode 100644 index 0000000..ada5bca --- /dev/null +++ b/go-backend/tests/contract/user_disable_contract_test.go @@ -0,0 +1,87 @@ +package contract_test + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "go-backend/internal/auth" + "go-backend/internal/http/response" + "go-backend/internal/store/model" +) + +func TestManualUserDisableRevokesLoginTokenAndForwardAccess(t *testing.T) { + const secret = "contract-jwt-secret" + router, r := setupContractRouter(t, secret) + seedContractUser(t, r, 2, "disable-contract", 1, 1) + for _, value := range []interface{}{ + &model.Tunnel{ID: 1, Name: "disable-contract", Type: 1, Status: 1, Flow: 1}, + &model.UserTunnel{ID: 10, UserID: 2, TunnelID: 1, Status: 1, Flow: 100, Num: 10}, + &model.Forward{ID: 20, UserID: 2, UserName: "disable-contract", Name: "paused", TunnelID: 1, RemoteAddr: "127.0.0.1:80", Status: 0}, + } { + if err := r.DB().Create(value).Error; err != nil { + t.Fatal(err) + } + } + userToken, err := auth.GenerateToken(2, "disable-contract", 1, secret) + if err != nil { + t.Fatal(err) + } + adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret) + if err != nil { + t.Fatal(err) + } + request := func(path, token, body string) response.R { + t.Helper() + req := httptest.NewRequest(http.MethodPost, path, bytes.NewBufferString(body)) + req.Header.Set("Authorization", token) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + var out response.R + if err := json.Unmarshal(res.Body.Bytes(), &out); err != nil { + t.Fatal(err) + } + return out + } + if out := request("/api/v1/forward/list", userToken, `{}`); out.Code != 0 { + t.Fatalf("token invalid before disable: %s", out.Msg) + } + if out := request("/api/v1/user/update", adminToken, `{"id":2,"user":"disable-contract","status":0}`); out.Code != 0 { + t.Fatalf("disable user: %s", out.Msg) + } + for _, check := range []struct{ path, body string }{ + {"/api/v1/forward/list", `{}`}, + {"/api/v1/forward/create", `{"tunnelId":1,"name":"bypass","remoteAddr":"127.0.0.1:80"}`}, + {"/api/v1/forward/resume", `{"id":20}`}, + {"/api/v1/forward/batch-resume", `{"ids":[20]}`}, + } { + if out := request(check.path, userToken, check.body); out.Code != 401 { + t.Errorf("disabled user's old token accepted at %s: %+v", check.path, out) + } + } + if out := request("/api/v1/user/login", "", `{"username":"disable-contract","password":"contract-pass","captchaId":""}`); out.Code == 0 || !strings.Contains(out.Msg, "停用") { + t.Errorf("disabled account login: %+v", out) + } + if out := request("/api/v1/forward/resume", adminToken, `{"id":20}`); out.Code == 0 || !strings.Contains(out.Msg, "禁用") { + t.Errorf("admin bypassed disabled owner's policy: %+v", out) + } + out := request("/api/v1/tunnel/user/list", adminToken, `{"userId":2}`) + if out.Code != 0 { + t.Fatalf("list permissions: %s", out.Msg) + } + items, ok := out.Data.([]interface{}) + if !ok || len(items) != 1 { + t.Fatalf("permission list: %#v", out.Data) + } + permission, ok := items[0].(map[string]interface{}) + if !ok || permission["status"] != float64(0) { + t.Errorf("permission still displayed as enabled: %#v", items[0]) + } + if got := mustQueryInt(t, r, "SELECT status FROM forward WHERE id = 20"); got != 0 { + t.Errorf("disabled forward resumed: %d", got) + } +} diff --git a/vite-frontend/src/pages/user.tsx b/vite-frontend/src/pages/user.tsx index 7cd05f6..fc0b5c5 100644 --- a/vite-frontend/src/pages/user.tsx +++ b/vite-frontend/src/pages/user.tsx @@ -231,6 +231,9 @@ export default function UserPage() { maxConn: 0, }); const [userFormLoading, setUserFormLoading] = useState(false); + const [userSubmitWarning, setUserSubmitWarning] = useState( + null, + ); const [userFlowInput, setUserFlowInput] = useState("1000"); const [userFlowUnit, setUserFlowUnit] = useState("GB"); const [quotaResetLoading, setQuotaResetLoading] = useState(false); @@ -517,6 +520,7 @@ export default function UserPage() { }; const handleAdd = () => { + setUserSubmitWarning(null); setIsEdit(false); setUserFlowInput("1000"); setUserFlowUnit("GB"); @@ -537,6 +541,7 @@ export default function UserPage() { }; const handleEdit = async (user: User) => { + setUserSubmitWarning(null); setIsEdit(true); const trafficInput = trafficInputFor(user.flow, user.flowMiB); @@ -618,6 +623,7 @@ export default function UserPage() { } setUserFormLoading(true); + setUserSubmitWarning(null); try { const submitData: any = { ...userForm, @@ -660,7 +666,19 @@ export default function UserPage() { await loadUsers(); } } else { - toast.error(response.msg || (isEdit ? "更新失败" : "创建失败")); + const userDisabled = + isEdit && + response.data !== null && + typeof response.data === "object" && + "userDisabled" in response.data && + response.data.userDisabled === true; + + if (userDisabled) { + setUserSubmitWarning(response.msg); + await loadUsers(); + } else { + toast.error(response.msg || (isEdit ? "更新失败" : "创建失败")); + } } } catch { toast.error(isEdit ? "更新失败" : "创建失败"); @@ -1532,6 +1550,14 @@ export default function UserPage() { {isEdit ? "编辑用户" : "新增用户"} + {userSubmitWarning ? ( +
+ {userSubmitWarning} +
+ ) : null}