//! 节点互联服务:双向授权同步的 peer 管理、HMAC 认证、握手与增量交换。 //! 加密通道 = 站点既有 HTTPS + 互联密钥 HMAC-SHA256 签名(防篡改/防重放),无需额外隧道。 // Logan.Li package sync import ( "bytes" "crypto/hmac" "crypto/rand" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" "net/http" "strings" "sync" "time" "backend/config" "backend/internal/dto" "backend/internal/model" "gorm.io/gorm" ) const ( syncBatchSize = 500 // 单实体单轮增量上限,宽松分批 // Logan.Li handshakePath = "/v1/sync/handshake" exchangePath = "/v1/sync/exchange" sigTimeWindow = 5 * time.Minute // 签名时间戳容差(防重放) // Logan.Li httpTimeout = 20 * time.Second protocolVersion = "1" ) // Service 节点互联业务逻辑层 // Logan.Li type Service struct { db *gorm.DB cfg *config.Config nodeID string client *http.Client mu sync.Mutex // 同一时刻仅一轮交换,防游标错乱 // Logan.Li } // NewService 创建 service 实例;节点 ID 由 JWT_SECRET 派生(两节点密钥不同,天然互异且稳定) // Logan.Li func NewService(db *gorm.DB, cfg *config.Config) *Service { sum := sha256.Sum256([]byte("opsx-sync-node:" + cfg.JWTSecret)) return &Service{ db: db, cfg: cfg, nodeID: hex.EncodeToString(sum[:8]), client: &http.Client{Timeout: httpTimeout}, } } // NodeID 本节点标识 // Logan.Li func (s *Service) NodeID() string { return s.nodeID } // GenerateSharedKey 生成 32 字节 hex 互联密钥 // Logan.Li func GenerateSharedKey() string { buf := make([]byte, 32) if _, err := rand.Read(buf); err != nil { panic("生成互联密钥失败: " + err.Error()) } return hex.EncodeToString(buf) } // ============ Peer 管理 ============ // ListPeers 节点列表 // Logan.Li func (s *Service) ListPeers() ([]dto.PeerResponse, error) { var peers []model.PeerNode if err := s.db.Order("id ASC").Find(&peers).Error; err != nil { return nil, err } results := make([]dto.PeerResponse, 0, len(peers)) for _, p := range peers { results = append(results, toPeerResponse(p)) } return results, nil } // SavePeer 创建或更新节点(ID=0 创建) // Logan.Li func (s *Service) SavePeer(req dto.PeerRequest) (*dto.PeerResponse, error) { base := normalizeBaseURL(req.BaseURL) key := strings.TrimSpace(req.SharedKey) if len(key) < 32 { return nil, errors.New("互联密钥至少 32 位(建议用生成按钮)") } var peer model.PeerNode if req.ID > 0 { if err := s.db.First(&peer, req.ID).Error; err != nil { return nil, errors.New("节点不存在") } peer.Name = req.Name peer.BaseURL = base peer.SharedKey = key peer.Status = model.PeerStatusPending // 配置变更后需重新握手确认 // Logan.Li } else { var count int64 s.db.Model(&model.PeerNode{}).Where("base_url = ?", base).Count(&count) if count > 0 { return nil, errors.New("该地址的节点已存在") } peer = model.PeerNode{Name: req.Name, BaseURL: base, SharedKey: key, Status: model.PeerStatusPending} } if err := s.db.Save(&peer).Error; err != nil { return nil, err } resp := toPeerResponse(peer) return &resp, nil } // DeletePeer 删除节点 // Logan.Li func (s *Service) DeletePeer(id uint) error { return s.db.Delete(&model.PeerNode{}, id).Error } func toPeerResponse(p model.PeerNode) dto.PeerResponse { tail := "" if len(p.SharedKey) >= 4 { tail = p.SharedKey[len(p.SharedKey)-4:] } return dto.PeerResponse{ ID: p.ID, Name: p.Name, BaseURL: p.BaseURL, Status: p.Status, RemoteNode: p.RemoteNode, KeyTail: tail, LastSyncAt: p.LastSyncAt, LastResult: p.LastResult, CreatedAt: p.CreatedAt, } } // normalizeBaseURL 统一为 https://host/api 形式;本地回环地址保留 http 供联调 // Logan.Li func normalizeBaseURL(raw string) string { base := strings.TrimRight(strings.TrimSpace(raw), "/") if base == "" { return base } isLocal := strings.Contains(base, "//127.0.0.1") || strings.Contains(base, "//localhost") if strings.HasPrefix(base, "http://") && !isLocal { base = "https://" + strings.TrimPrefix(base, "http://") } if strings.HasSuffix(base, "/api/v1") { base = strings.TrimSuffix(base, "/v1") } else if !strings.HasSuffix(base, "/api") { base += "/api" } return base } // ============ HMAC 签名 ============ // buildSignature 计算请求签名:HMAC-SHA256(key, "v1|node|ts|bodySha256") // Logan.Li func buildSignature(key, nodeID string, ts int64, body []byte) string { bodySum := sha256.Sum256(body) mac := hmac.New(sha256.New, []byte(key)) fmt.Fprintf(mac, "v1|%s|%d|%s", nodeID, ts, hex.EncodeToString(bodySum[:])) return hex.EncodeToString(mac.Sum(nil)) } // VerifySignature 校验入站请求签名与时间窗 // Logan.Li func VerifySignature(key string, r *http.Request, body []byte) error { nodeID := r.Header.Get("X-Sync-Node") tsStr := r.Header.Get("X-Sync-Ts") sig := r.Header.Get("X-Sync-Sig") if nodeID == "" || tsStr == "" || sig == "" { return errors.New("缺少同步签名头") } var ts int64 if _, err := fmt.Sscanf(tsStr, "%d", &ts); err != nil { return errors.New("签名时间戳非法") } if absDuration(time.Since(time.Unix(ts, 0))) > sigTimeWindow { return errors.New("签名已过期(时钟偏差过大)") } expect := buildSignature(key, nodeID, ts, body) if !hmac.Equal([]byte(expect), []byte(sig)) { return errors.New("互联密钥校验失败") } return nil } func absDuration(d time.Duration) time.Duration { if d < 0 { return -d } return d } // ============ 握手 ============ // handshakeOut 对端发起握手 // Logan.Li func (s *Service) handshakeOut(peer *model.PeerNode) (*dto.SyncHandshakeResponse, error) { body, _ := json.Marshal(dto.SyncHandshakeRequest{NodeName: s.hostName()}) var resp dto.SyncHandshakeResponse if err := s.signedPost(peer, handshakePath, body, &resp); err != nil { return nil, err } return &resp, nil } // HandshakeIn 入站握手:签名已由 resolvePeer 按互联密钥验证,此处互认并激活 // Logan.Li func (s *Service) HandshakeIn(peer *model.PeerNode, req dto.SyncHandshakeRequest) (*dto.SyncHandshakeResponse, error) { if peer.Status == model.PeerStatusDisabled { return nil, errors.New("本侧已停用该互联") } if peer.Status != model.PeerStatusActive { s.db.Model(peer).Updates(map[string]interface{}{"status": model.PeerStatusActive}) } return &dto.SyncHandshakeResponse{ NodeID: s.nodeID, NodeName: s.hostName(), Version: protocolVersion, ServerTime: time.Now().UTC(), }, nil } // TestHandshake 管理端手动握手:成功则双方激活 // Logan.Li func (s *Service) TestHandshake(id uint) (*dto.PeerResponse, error) { var peer model.PeerNode if err := s.db.First(&peer, id).Error; err != nil { return nil, errors.New("节点不存在") } resp, err := s.handshakeOut(&peer) if err != nil { s.db.Model(&peer).Updates(map[string]interface{}{"last_result": "握手失败: " + truncate(err.Error(), 200)}) return nil, fmt.Errorf("握手失败: %s", err.Error()) } s.db.Model(&peer).Updates(map[string]interface{}{ "status": model.PeerStatusActive, "remote_node": resp.NodeID, "last_result": "握手成功,对端: " + resp.NodeName, }) _ = s.db.First(&peer, id).Error out := toPeerResponse(peer) return &out, nil } // signedPost 携带 HMAC 签名调用对端 // Logan.Li func (s *Service) signedPost(peer *model.PeerNode, path string, body []byte, out interface{}) error { url := strings.TrimRight(peer.BaseURL, "/") + path ts := time.Now().Unix() req, err := http.NewRequest("POST", url, bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Content-Type", "application/json") req.Header.Set("X-Sync-Node", s.nodeID) req.Header.Set("X-Sync-Ts", fmt.Sprintf("%d", ts)) req.Header.Set("X-Sync-Sig", buildSignature(peer.SharedKey, s.nodeID, ts, body)) resp, err := s.client.Do(req) if err != nil { return fmt.Errorf("请求对端失败: %w", err) } defer resp.Body.Close() raw, _ := io.ReadAll(io.LimitReader(resp.Body, 32<<20)) if resp.StatusCode != http.StatusOK { return fmt.Errorf("对端返回 %d: %s", resp.StatusCode, truncate(string(raw), 120)) } if err := json.Unmarshal(raw, out); err != nil { return fmt.Errorf("对端响应解析失败: %w", err) } return nil } // hostName 对端展示用本机名(无 os/env 依赖,用节点 ID 前缀) // Logan.Li func (s *Service) hostName() string { return "node-" + s.nodeID[:8] } func truncate(s string, n int) string { if len(s) <= n { return s } return s[:n] }