Merge branch 'main' into opencode/kind-planet

go-backend/internal/http/handler/control_plane.go

@@ -457,7 +457,17 @@ func (h *Handler) applyNodeProtocolChange(nodeID int64, httpVal, tlsVal, socksVa
 }
 
 func (h *Handler) sendNodeCommand(nodeID int64, commandType string, data interface{}, tolerateExists bool, tolerateNotFound bool) (ws.CommandResult, error) {
-	result, err := h.wsServer.SendCommand(nodeID, commandType, data, 12*time.Second)
+	var (
+		result ws.CommandResult
+		err    error
+	)
+
+	node, nodeErr := h.getNodeRecord(nodeID)
+	if nodeErr == nil && node != nil && node.IsRemote == 1 {
+		result, err = h.sendRemoteNodeCommand(node, commandType, data)
+	} else {
+		result, err = h.wsServer.SendCommand(nodeID, commandType, data, 12*time.Second)
+	}
 	if err == nil {
 		return result, nil
 	}
@@ -475,6 +485,44 @@ func (h *Handler) sendNodeCommand(nodeID int64, commandType string, data interfa
 	return result, err
 }
 
+func (h *Handler) sendRemoteNodeCommand(node *nodeRecord, commandType string, data interface{}) (ws.CommandResult, error) {
+	if node == nil {
+		return ws.CommandResult{}, errors.New("节点不存在")
+	}
+	remoteURL := strings.TrimSpace(node.RemoteURL)
+	remoteToken := strings.TrimSpace(node.RemoteToken)
+	if remoteURL == "" || remoteToken == "" {
+		return ws.CommandResult{}, errors.New("远程节点缺少共享配置")
+	}
+
+	fc := client.NewFederationClient()
+	res, err := fc.Command(remoteURL, remoteToken, h.federationLocalDomain(), client.RuntimeNodeCommandRequest{
+		CommandType: commandType,
+		Data:        data,
+	})
+	if err != nil {
+		return ws.CommandResult{}, err
+	}
+	if res == nil {
+		return ws.CommandResult{}, errors.New("远程节点未返回命令结果")
+	}
+
+	result := ws.CommandResult{
+		Type:    res.Type,
+		Success: res.Success,
+		Message: res.Message,
+		Data:    res.Data,
+	}
+	if !result.Success {
+		msg := strings.TrimSpace(result.Message)
+		if msg == "" {
+			msg = "命令执行失败"
+		}
+		return result, errors.New(msg)
+	}
+	return result, nil
+}
+
 func (h *Handler) diagnoseForwardRuntime(forward *forwardRecord) (map[string]interface{}, error) {
 	if forward == nil {
 		return nil, errForwardNotFound

go-backend/internal/http/handler/federation.go

@@ -91,6 +91,11 @@ type federationRuntimeDiagnoseRequest struct {
 	Timeout int    `json:"timeout"`
 }
 
+type federationRuntimeCommandRequest struct {
+	CommandType string      `json:"commandType"`
+	Data        interface{} `json:"data"`
+}
+
 type peerShareUsedPort struct {
 	RuntimeID   int64  `json:"runtimeId"`
 	Port        int    `json:"port"`
@@ -1199,6 +1204,51 @@ func (h *Handler) federationRuntimeDiagnose(w http.ResponseWriter, r *http.Reque
 	response.WriteJSON(w, response.OK(res.Data))
 }
 
+func (h *Handler) federationRuntimeCommand(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		response.WriteJSON(w, response.ErrDefault("Invalid method"))
+		return
+	}
+
+	token := extractBearerToken(r)
+	share, err := h.repo.GetPeerShareByToken(token)
+	if err != nil || share == nil {
+		response.WriteJSON(w, response.Err(401, "Unauthorized"))
+		return
+	}
+
+	var req federationRuntimeCommandRequest
+	if err := decodeJSON(r.Body, &req); err != nil {
+		response.WriteJSON(w, response.ErrDefault("Invalid JSON"))
+		return
+	}
+	cmd := strings.TrimSpace(req.CommandType)
+	if cmd == "" {
+		response.WriteJSON(w, response.ErrDefault("commandType is required"))
+		return
+	}
+	if !isFederationRuntimeCommandAllowed(cmd) {
+		response.WriteJSON(w, response.ErrDefault("command not allowed"))
+		return
+	}
+
+	res, err := h.sendNodeCommand(share.NodeID, cmd, req.Data, false, false)
+	if err != nil {
+		response.WriteJSON(w, response.ErrDefault(err.Error()))
+		return
+	}
+	response.WriteJSON(w, response.OK(res))
+}
+
+func isFederationRuntimeCommandAllowed(commandType string) bool {
+	switch strings.ToLower(strings.TrimSpace(commandType)) {
+	case "addservice", "updateservice", "deleteservice", "pauseservice", "resumeservice", "addchains", "deletechains", "addlimiters", "deletelimiters", "tcpping", "reload":
+		return true
+	default:
+		return false
+	}
+}
+
 func (h *Handler) pickPeerSharePort(share *sqlite.PeerShare, requestedPort int) (int, error) {
 	if share == nil {
 		return 0, fmt.Errorf("share not found")

go-backend/internal/http/handler/federation_runtime_test.go

@@ -152,6 +152,71 @@ func TestPrepareTunnelCreateStateRemoteAutoPortDefersToFederation(t *testing.T)
 	}
 }
 
+func TestPrepareTunnelCreateStateAllowsOfflineRemoteMiddleNode(t *testing.T) {
+	repo, err := sqlite.Open(filepath.Join(t.TempDir(), "panel.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer repo.Close()
+
+	h := &Handler{repo: repo}
+	now := time.Now().UnixMilli()
+
+	insertNode := func(name string, status int, portRange string, isRemote int) int64 {
+		res, execErr := repo.DB().Exec(`
+			INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx, is_remote, remote_url, remote_token, remote_config)
+			VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		`, name, name+"-secret", "10.0.0.1", "10.0.0.1", "", portRange, "", "v1", 1, 1, 1, now, now, status, "[::]", "[::]", 0, isRemote, "http://peer", "peer-token", `{"shareId":2}`)
+		if execErr != nil {
+			t.Fatalf("insert node %s: %v", name, execErr)
+		}
+		id, idErr := res.LastInsertId()
+		if idErr != nil {
+			t.Fatalf("node id %s: %v", name, idErr)
+		}
+		return id
+	}
+
+	entryID := insertNode("entry-local", 1, "32000-32010", 0)
+	remoteMiddleID := insertNode("middle-remote", 0, "33000-33010", 1)
+	outID := insertNode("out-local", 1, "34000-34010", 0)
+
+	tx, err := repo.DB().Begin()
+	if err != nil {
+		t.Fatalf("begin tx: %v", err)
+	}
+	defer tx.Rollback()
+
+	req := map[string]interface{}{
+		"name": "remote-middle-offline-status",
+		"inNodeId": []interface{}{
+			map[string]interface{}{"nodeId": float64(entryID), "protocol": "tls", "strategy": "round"},
+		},
+		"chainNodes": []interface{}{
+			[]interface{}{
+				map[string]interface{}{"nodeId": float64(remoteMiddleID), "protocol": "tls", "strategy": "round", "port": float64(0)},
+			},
+		},
+		"outNodeId": []interface{}{
+			map[string]interface{}{"nodeId": float64(outID), "protocol": "tls", "strategy": "round", "port": float64(0)},
+		},
+	}
+
+	state, err := h.prepareTunnelCreateState(tx, req, 2, 0)
+	if err != nil {
+		t.Fatalf("prepare state should allow offline remote middle node: %v", err)
+	}
+	if len(state.ChainHops) != 1 || len(state.ChainHops[0]) != 1 {
+		t.Fatalf("expected one middle hop node, got %+v", state.ChainHops)
+	}
+	if state.ChainHops[0][0].NodeID != remoteMiddleID {
+		t.Fatalf("expected remote middle node id %d, got %d", remoteMiddleID, state.ChainHops[0][0].NodeID)
+	}
+	if state.Nodes[remoteMiddleID] == nil || state.Nodes[remoteMiddleID].IsRemote != 1 {
+		t.Fatalf("expected remote middle node metadata in state")
+	}
+}
+
 func TestFederationRuntimeReservePortRejectsWhenShareFlowExceeded(t *testing.T) {
 	repo, err := sqlite.Open(filepath.Join(t.TempDir(), "panel.db"))
 	if err != nil {

go-backend/internal/http/handler/handler.go

@@ -93,6 +93,12 @@ func (h *Handler) Register(mux *http.ServeMux) {
 	mux.HandleFunc("/api/v1/config/list", h.getConfigs)
 	mux.HandleFunc("/api/v1/config/update", h.updateConfigs)
 	mux.HandleFunc("/api/v1/config/update-single", h.updateSingleConfig)
+	mux.HandleFunc("/api/v1/backup/export", h.backupExport)
+	mux.HandleFunc("/api/v1/backup/import", h.backupImport)
+	mux.HandleFunc("/api/v1/backup/restore", h.backupImport)
+	mux.HandleFunc("/api/v1/api/v1/backup/export", h.backupExport)
+	mux.HandleFunc("/api/v1/api/v1/backup/import", h.backupImport)
+	mux.HandleFunc("/api/v1/api/v1/backup/restore", h.backupImport)
 	mux.HandleFunc("/api/v1/captcha/check", h.checkCaptcha)
 	mux.HandleFunc("/api/v1/captcha/verify", h.captchaVerify)
 	mux.HandleFunc("/api/v1/user/package", h.userPackage)
@@ -169,6 +175,7 @@ func (h *Handler) Register(mux *http.ServeMux) {
 	mux.HandleFunc("/api/v1/federation/runtime/apply-role", h.authPeer(h.federationRuntimeApplyRole))
 	mux.HandleFunc("/api/v1/federation/runtime/release-role", h.authPeer(h.federationRuntimeReleaseRole))
 	mux.HandleFunc("/api/v1/federation/runtime/diagnose", h.authPeer(h.federationRuntimeDiagnose))
+	mux.HandleFunc("/api/v1/federation/runtime/command", h.authPeer(h.federationRuntimeCommand))
 	mux.HandleFunc("/api/v1/federation/node/import", h.nodeImport)
 
 	mux.HandleFunc("/flow/test", h.flowTest)
@@ -1140,3 +1147,84 @@ func (h *Handler) verifyCloudflareTurnstile(token, secretKey string) bool {
 	}
 	return body.Success
 }
+
+type backupExportRequest struct {
+	Types []string `json:"types"`
+}
+
+func (h *Handler) backupExport(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		response.WriteJSON(w, response.ErrDefault("请求失败"))
+		return
+	}
+
+	var req backupExportRequest
+	if err := decodeJSON(r.Body, &req); err != nil {
+		response.WriteJSON(w, response.Err(500, "请求参数错误"))
+		return
+	}
+
+	var backup interface{}
+	var err error
+
+	if len(req.Types) == 0 {
+		backup, err = h.repo.ExportAll()
+	} else {
+		backup, err = h.repo.ExportPartial(req.Types)
+	}
+
+	if err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
+
+	w.Header().Set("Content-Disposition", "attachment; filename=backup.json")
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(backup); err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
+}
+
+type backupImportRequest struct {
+	Types []string `json:"types"`
+	sqlite.BackupData
+}
+
+func (h *Handler) backupImport(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		response.WriteJSON(w, response.ErrDefault("请求失败"))
+		return
+	}
+
+	var req backupImportRequest
+	if err := decodeJSON(r.Body, &req); err != nil {
+		response.WriteJSON(w, response.Err(500, "请求参数错误"))
+		return
+	}
+
+	if len(req.Types) == 0 {
+		response.WriteJSON(w, response.Err(500, "请选择要导入的数据类型"))
+		return
+	}
+
+	autoBackup, err := h.repo.ExportAll()
+	if err != nil {
+		response.WriteJSON(w, response.Err(-2, fmt.Sprintf("导入前自动备份失败: %v", err)))
+		return
+	}
+
+	if req.BackupData.Version == "" {
+		response.WriteJSON(w, response.Err(500, "备份数据格式错误"))
+		return
+	}
+
+	result, err := h.repo.Import(&req.BackupData, req.Types)
+	if err != nil {
+		response.WriteJSON(w, response.Err(-2, fmt.Sprintf("导入失败: %v", err)))
+		return
+	}
+
+	result.AutoBackup = autoBackup
+	response.WriteJSON(w, response.OK(result))
+}

go-backend/internal/http/handler/mutations.go

@@ -1713,10 +1713,19 @@ func (h *Handler) groupUserAssign(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	defer func() { _ = tx.Rollback() }()
+	previousUserIDs, err := queryInt64ListTx(tx, `SELECT user_id FROM user_group_user WHERE user_group_id = ?`, req.GroupID)
+	if err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
 	_, _ = tx.Exec(`DELETE FROM user_group_user WHERE user_group_id = ?`, req.GroupID)
 	for _, uid := range req.UserIDs {
 		_, _ = tx.Exec(`INSERT INTO user_group_user(user_group_id, user_id, created_time) VALUES(?, ?, ?) ON CONFLICT DO NOTHING`, req.GroupID, uid, time.Now().UnixMilli())
 	}
+	if err := revokeGroupGrantsForRemovedUsersTx(tx, req.GroupID, previousUserIDs, req.UserIDs); err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
 	if err := tx.Commit(); err != nil {
 		response.WriteJSON(w, response.Err(-2, err.Error()))
 		return
@@ -1748,10 +1757,35 @@ func (h *Handler) groupPermissionRemove(w http.ResponseWriter, r *http.Request)
 	if id <= 0 {
 		return
 	}
+	tx, err := h.repo.DB().Begin()
+	if err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
+	defer func() { _ = tx.Rollback() }()
+
 	var ug, tg int64
-	_ = h.repo.DB().QueryRow(`SELECT user_group_id, tunnel_group_id FROM group_permission WHERE id = ?`, id).Scan(&ug, &tg)
-	_, _ = h.repo.DB().Exec(`DELETE FROM group_permission WHERE id = ?`, id)
-	_, _ = h.repo.DB().Exec(`DELETE FROM group_permission_grant WHERE user_group_id = ? AND tunnel_group_id = ?`, ug, tg)
+	err = tx.QueryRow(`SELECT user_group_id, tunnel_group_id FROM group_permission WHERE id = ?`, id).Scan(&ug, &tg)
+	if err != nil && err != sql.ErrNoRows {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
+
+	if _, err := tx.Exec(`DELETE FROM group_permission WHERE id = ?`, id); err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
+	if err == nil {
+		if err := revokeGroupPermissionPairTx(tx, ug, tg); err != nil {
+			response.WriteJSON(w, response.Err(-2, err.Error()))
+			return
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
 	response.WriteJSON(w, response.OKEmpty())
 }
 
@@ -1908,6 +1942,144 @@ func queryInt64List(db *store.DB, q string, args ...interface{}) ([]int64, error
 	return out, rows.Err()
 }
 
+func queryInt64ListTx(tx *store.Tx, q string, args ...interface{}) ([]int64, error) {
+	rows, err := tx.Query(q, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := make([]int64, 0)
+	for rows.Next() {
+		var v int64
+		if err := rows.Scan(&v); err != nil {
+			return nil, err
+		}
+		out = append(out, v)
+	}
+	return out, rows.Err()
+}
+
+func revokeGroupGrantsForRemovedUsersTx(tx *store.Tx, userGroupID int64, previousUserIDs, currentUserIDs []int64) error {
+	currentSet := make(map[int64]struct{}, len(currentUserIDs))
+	for _, uid := range currentUserIDs {
+		if uid > 0 {
+			currentSet[uid] = struct{}{}
+		}
+	}
+
+	removedUserIDs := make([]int64, 0)
+	for _, uid := range previousUserIDs {
+		if uid <= 0 {
+			continue
+		}
+		if _, ok := currentSet[uid]; !ok {
+			removedUserIDs = append(removedUserIDs, uid)
+		}
+	}
+	if len(removedUserIDs) == 0 {
+		return nil
+	}
+
+	for _, userID := range removedUserIDs {
+		rows, err := tx.Query(`
+			SELECT g.user_tunnel_id, g.created_by_group
+			FROM group_permission_grant g
+			JOIN user_tunnel ut ON ut.id = g.user_tunnel_id
+			WHERE g.user_group_id = ? AND ut.user_id = ?
+		`, userGroupID, userID)
+		if err != nil {
+			return err
+		}
+
+		groupCreatedTunnelIDs := make(map[int64]struct{})
+		for rows.Next() {
+			var userTunnelID int64
+			var createdByGroup int
+			if err := rows.Scan(&userTunnelID, &createdByGroup); err != nil {
+				rows.Close()
+				return err
+			}
+			if createdByGroup == 1 && userTunnelID > 0 {
+				groupCreatedTunnelIDs[userTunnelID] = struct{}{}
+			}
+		}
+		if err := rows.Err(); err != nil {
+			rows.Close()
+			return err
+		}
+		rows.Close()
+
+		if _, err := tx.Exec(`
+			DELETE FROM group_permission_grant
+			WHERE user_group_id = ?
+			  AND user_tunnel_id IN (SELECT id FROM user_tunnel WHERE user_id = ?)
+		`, userGroupID, userID); err != nil {
+			return err
+		}
+
+		for userTunnelID := range groupCreatedTunnelIDs {
+			var remaining int
+			if err := tx.QueryRow(`SELECT COUNT(1) FROM group_permission_grant WHERE user_tunnel_id = ?`, userTunnelID).Scan(&remaining); err != nil {
+				return err
+			}
+			if remaining == 0 {
+				if _, err := tx.Exec(`DELETE FROM user_tunnel WHERE id = ?`, userTunnelID); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func revokeGroupPermissionPairTx(tx *store.Tx, userGroupID, tunnelGroupID int64) error {
+	rows, err := tx.Query(`
+		SELECT user_tunnel_id, created_by_group
+		FROM group_permission_grant
+		WHERE user_group_id = ? AND tunnel_group_id = ?
+	`, userGroupID, tunnelGroupID)
+	if err != nil {
+		return err
+	}
+
+	groupCreatedTunnelIDs := make(map[int64]struct{})
+	for rows.Next() {
+		var userTunnelID int64
+		var createdByGroup int
+		if err := rows.Scan(&userTunnelID, &createdByGroup); err != nil {
+			rows.Close()
+			return err
+		}
+		if createdByGroup == 1 && userTunnelID > 0 {
+			groupCreatedTunnelIDs[userTunnelID] = struct{}{}
+		}
+	}
+	if err := rows.Err(); err != nil {
+		rows.Close()
+		return err
+	}
+	rows.Close()
+
+	if _, err := tx.Exec(`DELETE FROM group_permission_grant WHERE user_group_id = ? AND tunnel_group_id = ?`, userGroupID, tunnelGroupID); err != nil {
+		return err
+	}
+
+	for userTunnelID := range groupCreatedTunnelIDs {
+		var remaining int
+		if err := tx.QueryRow(`SELECT COUNT(1) FROM group_permission_grant WHERE user_tunnel_id = ?`, userTunnelID).Scan(&remaining); err != nil {
+			return err
+		}
+		if remaining == 0 {
+			if _, err := tx.Exec(`DELETE FROM user_tunnel WHERE id = ?`, userTunnelID); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
 func queryPairs(db *store.DB, q string, args ...interface{}) ([][2]int64, error) {
 	rows, err := db.Query(q, args...)
 	if err != nil {
@@ -2061,7 +2233,7 @@ func (h *Handler) prepareTunnelCreateState(tx *store.Tx, req map[string]interfac
 			}
 			return nil, err
 		}
-		if node.Status != 1 {
+		if node.IsRemote != 1 && node.Status != 1 {
 			return nil, errors.New("部分节点不在线")
 		}
 		state.Nodes[nodeID] = node