272 lines
8.4 KiB
Go
272 lines
8.4 KiB
Go
//! 节点互联服务：双向授权同步的 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]
|
|
}
|