zkcoi/Meshray-Manager

Fork 0

Files

T

zkcoi 15dab96872 Initial commit

2026-06-30 15:14:37 +08:00

24 KiB

Raw Permalink Blame History

MeshRay 关键技术详解

1. 雪花算法 ID 生成

算法结构

64 位整数：
┌─┬──────────────────────┬────────────┬─────────────┐
│0│     41 bits 时间戳    │ 10 bits    │ 12 bits     │
│ │ (毫秒级，69 年)       │ 节点 ID     │ 序列号      │
└─┴──────────────────────┴────────────┴─────────────┘
   ↑                      ↑           ↑
   符号位                机器标识     单毫秒内自增

Go 实现

package snowflake

import (
    "sync"
    "time"
)

type Snowflake struct {
    mu          sync.Mutex
    lastStamp   int64  // 上次生成 ID 的时间戳
    sequence    int64  // 当前毫秒内的序列号
    machineID   int64  // 机器 ID（0-1023）
}

func NewSnowflake(machineID int64) *Snowflake {
    return &Snowflake{
        machineID: machineID,
        lastStamp: -1,
    }
}

func (s *Snowflake) Generate() int64 {
    s.mu.Lock()
    defer s.mu.Unlock()
    
    stamp := time.Now().UnixMilli()
    if stamp < s.lastStamp {
        panic("时钟回拨")
    }
    
    if stamp == s.lastStamp {
        // 同一毫秒内，序列号自增
        s.sequence = (s.sequence + 1) & 0xFFF
        if s.sequence == 0 {
            // 序列号溢出，等待下一毫秒
            for stamp <= s.lastStamp {
                stamp = time.Now().UnixMilli()
            }
        }
    } else {
        // 新毫秒，重置序列号
        s.sequence = 0
    }
    
    s.lastStamp = stamp
    
    // 组合 ID
    id := (stamp << 22) | (s.machineID << 12) | s.sequence
    return id
}

前端精度问题与解决方案

问题现象： ``javascript // ❌ JavaScript 无法精确表示 20 位数字 const backendID = 18446744073709551615; console.log(backendID);
// 输出：18446744073709552000（精度丢失！）

Number.MAX_SAFE_INTEGER = 9007199254740991 // 仅 16 位


**解决方案：后端使用 string 传输**

``go
// internal/model/network.go
type Network struct {
    ID        string    `gorm:"primaryKey;type:varchar(36)" json:"id"`
    ID_UInt64 uint64    `gorm:"column:id_uint64;not null" json:"-"`
}

// BeforeCreate Hook：uint64 → string
func (n *Network) BeforeCreate(tx *gorm.DB) error {
    n.ID = strconv.FormatUint(n.ID_UInt64, 10)
    return nil
}

// AfterFind Hook：string ← uint64
func (n *Network) AfterFind(tx *gorm.DB) error {
    n.ID_UInt64, _ = strconv.ParseUint(n.ID, 10, 64)
    return nil
}

前端使用： ``typescript // 前端始终处理字符串 interface Network { id: string // "18446744073709551615" name: string }

// 不需要任何特殊处理 const network = await api.getNetwork('18446744073709551615') console.log(network.id) // "18446744073709551615" ✅


---

## 2. Conn.Bind 模型与 9 层传输

### Bind 接口定义

WireGuard 官方库提供的网络层抽象接口：

// golang.zx2c4.com/wireguard/conn/bind.go type Bind interface { Open(port uint16) (fns []ReceiveFunc, err error) Close() error SetMark(mark uint32) error Send(buffs [][]byte, ep Endpoint) error ParseEndpoint(addr string) (Endpoint, error) }


**接口契约说明**：

| 方法 | 调用时机 | 返回值含义 |
|------|---------|------------|
| `Open` | WG 设备启动时 | `fns` 数组长度 = CPU 核心数（每核一个接收函数） |
| `Close` | WG 设备停止时 | 必须释放所有资源（端口、goroutine） |
| `SetMark` | Linux 特有 | 用于 iptables 路由策略 |
| `Send` | WG 要发送密文包 | 返回 nil 表示已发出，非 nil 触发重试 |
| `ParseEndpoint` | 添加 Peer 时 | 将字符串 "host:port" 转为 Endpoint 对象 |

---

### TransportFactory 接口定义

每种传输方式实现 `TransportFactory` 接口：

// core/connect/factory.go type TransportFactory interface { // Layer 返回所属传输层（用于优先级排序） Layer() Layer

// Name 返回名称（用于日志和监控）
Name() string

// Dial 拨号建立连接（核心方法）
Dial(ctx context.Context, config *DialConfig) (net.Conn, error)

}

// DialConfig 拨号配置（所有传输方式共享） type DialConfig struct { LocalAddr string // 本地地址 RemoteAddr string // 对端地址 PeerID string // 对端设备 ID Timeout time.Duration // 超时时间 STUNServers []string // STUN 服务器列表（P2P 需要） TURNConfig *TURNConfig // TURN 配置（TURN 需要） }


**接口契约说明**：

| 方法 | 作用 | 实现要求 |
|------|------|---------|
| `Layer()` | 返回所属层级 | 决定在 9 层中的位置（LayerDirectUDP/LayerTURN_UDP 等） |
| `Name()` | 人类可读的名称 | 用于日志输出："Direct-UDP" / "TURN-TCP" / "WS" |
| `Dial()` | 建立连接 | 返回 net.Conn 接口，封装底层协议细节 |

---

### CoreBind 实现

// core/core_bind.go type CoreBind struct { conn.Bind routes map[RouteID]net.Conn // Route ID → 实际链路 mu sync.RWMutex }

// Send 截获 WG 密文包，选择链路发送 func (cb *CoreBind) Send(buffs [][]byte, ep conn.Endpoint) error { cb.mu.RLock() defer cb.mu.RUnlock()

packet := buffs[0]

// 数据包分类
msgType := packet[0]
if msgType == 1 || msgType == 2 || msgType == 3 {
    // 控制包（握手/Cookie）：按已建链路透传
    return cb.sendControlPacket(packet, ep)
}

if msgType == 4 {
    // 数据包：提取 Route ID 并查表
    routeID := binary.LittleEndian.Uint32(packet[4:8])
    link := cb.routes[routeID]
    
    if link == nil {
        return fmt.Errorf("no route for %d", routeID)
    }
    
    // 通过选定链路发送
    _, err := link.Write(packet)
    return err
}

return nil

}


### 9 层传输工厂模式

每种传输方式实现 `TransportFactory` 接口：

// core/connect/factory.go type TransportFactory interface { Layer() Layer Name() string Dial(ctx context.Context, config *DialConfig) (net.Conn, error) }


#### 示例 1：Direct-UDP（STUN P2P）

// core/connect/p2p_factory.go type P2PFactory struct { stunServers []string logger *zap.Logger }

func (f *P2PFactory) Layer() Layer { return LayerDirectUDP }

func (f *P2PFactory) Name() string { return "Direct-UDP" }

func (f *P2PFactory) Dial(ctx context.Context, config *DialConfig) (net.Conn, error) { // 1. STUN 打洞获取公网地址 publicAddr, err := stun.GetPublicAddr(f.stunServers...)

// 2. 交换候选地址（通过 ctr 信令中转）
localCandidate := Candidate{Addr: publicAddr, Priority: 100}
remoteCandidate := f.exchangeCandidates(config.PeerID, localCandidate)

// 3. 建立 P2P 连接
conn, err := net.DialUDP("udp", nil, &remoteCandidate.Addr)

// 4. 包装为 P2PConn（带保活、NAT 刷新）
return NewP2PConn(conn, f.logger), nil

}


#### 示例 2：TURN-UDP

// core/connect/turn_factory.go type TURNFactory struct { serverConfig *TURNConfig logger *zap.Logger }

func (f *TURNFactory) Dial(ctx context.Context, config *DialConfig) (net.Conn, error) { // 1. 连接 TURN 服务器 turnConn, err := turnclient.New(config.ServerAddr, config.Username, config.Password)

// 2. 创建 Permission（允许对端 IP）
turnConn.CreatePermission(config.RemoteAddr)

// 3. 返回 TURNConn（封装了 Allocate/Send/Refresh）
return NewTURNConn(turnConn, f.logger), nil

}


#### 示例 3：FakeTCP（UDP 伪装 TCP）

// core/connect/fake_tcp_factory.go type FakeTCPFactory struct { logger *zap.Logger }

func (f *FakeTCPFactory) Dial(ctx context.Context, config *DialConfig) (net.Conn, error) { // 1. 建立 TCP 连接 tcpConn, err := net.DialTCP("tcp", nil, config.RemoteAddr)

// 2. 将 UDP 包封装为 TCP 流（带长度前缀）
//    [4 bytes length][UDP packet...]
return NewFakeTCPConn(tcpConn, f.logger), nil

}


### 策略调度器

// core/connect/strategy.go type StrategyScheduler struct { factories map[Layer]TransportFactory priority []Layer // 优先级顺序 }

// TryAll 按优先级尝试所有层，直到成功 func (s *StrategyScheduler) TryAll(ctx context.Context, config *DialConfig) (net.Conn, error) { var lastErr error

for _, layer := range s.priority {
    factory := s.factories[layer]
    if factory == nil {
        continue
    }
    
    conn, err := factory.Dial(ctx, config)
    if err == nil {
        // 成功，记录日志并返回
        s.logger.Info("connected via " + factory.Name())
        return conn, nil
    }
    
    lastErr = err
    s.logger.Debug(layer.String() + " failed: " + err.Error())
}

return nil, fmt.Errorf("all layers failed: %w", lastErr)

}


### 传输状态机

初始状态 ──→ Layer 1 (Direct-UDP) ├─ 成功 → 保持（持续监控） ├─ 超时 (500ms) → Layer 2 (FakeTCP) └─ 丢包率 > 10% → Layer 2

Layer 2 (FakeTCP) ──→ Layer 3 (RealTCP) ├─ 成功 → 保持 └─ 失败 → Layer 4

... 依次降级 ...

Layer 8 (WS/WSS) ──→ 降级完成 → 启动恢复探测 │ ├─ 每 30 秒探测 Layer 1 ├─ 连续 2 次成功 → 切回 Layer 1 └─ 失败 → 继续等待下一轮探测


**状态转换规则**：

| 当前状态 | 触发条件 | 目标状态 |
|----------|---------|---------|
| Layer N | 握手成功 | 保持 Layer N |
| Layer N | 单包超时 > 500ms | Layer N+1 |
| Layer N | 丢包率 > 10% (10s 窗口) | Layer N+1 |
| Layer N+1 | 建立成功 | 保持 Layer N+1 |
| Layer 9 (WS) | 恢复探测成功×2 | Layer 1 |
| Layer 9 (WS) | 恢复探测失败 | 保持 Layer 9 |

### 自动切换机制

// 降级触发条件 type FallbackPolicy struct { PacketTimeout time.Duration // 单包超时：500ms LossRateWindow time.Duration // 丢包率窗口：10s LossRateThreshold float64 // 丢包率阈值：10% }

// 恢复探测 func (s *StrategyScheduler) StartRecoveryProbe(currentLayer Layer) { ticker := time.NewTicker(30 * time.Second)

go func() {
    for range ticker.C {
        // 每隔 30 秒向 Layer 1 发送探测包
        success := s.probeLayer(LayerDirectUDP)
        
        if success {
            // 连续 2 次成功 → 直接切回 Layer 1
            s.switchTo(LayerDirectUDP)
            break
        }
    }
}()

}


---

## 3. Mesh 中继架构

### 核心设计原则

**Mesh 中继不是 Core 的功能，而是 WG 设备层的静态路由拓扑配置。**

### 数据流向

A (10.0.0.2) ──→ C (10.0.0.1) ──→ B (10.0.0.3) ↑ 中继节点

WG 视角： A 的 WG 配置：Peer C, AllowedIPs = 10.0.0.0/24 B 的 WG 配置：Peer C, AllowedIPs = 10.0.0.0/24 C 的 WG 配置：Peer A (AllowedIPs=10.0.0.2/32), Peer B (AllowedIPs=10.0.0.3/32)

Core 视角： A.Core: Direct-UDP → C.Core B.Core: Direct-UDP → C.Core C.Core: 无感知，只是转发密文包


### ctr 配置中继节点

// internal/ctr/wg.go func (c *Ctr) ConfigureRelay(relayDevice *model.Device, peers []*model.Device) error { wgConfig := wgtypes.Config{}

// 为中继节点配置所有普通节点作为 Peer
for _, peer := range peers {
    wgConfig.Peers = append(wgConfig.Peers, wgtypes.PeerConfig{
        PublicKey: peer.PublicKey,
        AllowedIPs: []net.IPNet{
            peer.VirtualIP,      // 10.0.0.3/32
            peer.LocalSubnet,    // 192.168.1.0/24（可选）
        },
    })
}

// 调用 wgctrl 应用配置
return c.wgClient.ConfigureDevice(relayDevice.WGInterface, wgConfig)

}


### 系统要求（重要）

**启用 Mesh 中继前，必须在操作系统层面开启 IP 转发**：

#### Linux 系统

1. 临时开启 IPv4 转发

sudo sysctl -w net.ipv4.ip_forward=1

2. 永久开启

echo "net.ipv4.ip_forward = 1" >> /etc/sysctl.conf sysctl -p

3. 配置 iptables NAT 转发（必需！）

sudo iptables -t nat -A POSTROUTING -o wg0 -j MASQUERADE sudo iptables -A FORWARD -i wg0 -j ACCEPT sudo iptables -A FORWARD -o wg0 -j ACCEPT

4. 保存 iptables 规则

sudo apt install iptables-persistent sudo netfilter-persistent save


#### Docker 容器

docker run
--privileged \ # 必需：特权模式 --sysctl net.ipv4.ip_forward=1 \ # 必需：开启 IP 转发 --cap-add=NET_ADMIN \ # 推荐：网络管理权限 meshray-node:latest


---

## 4. ExternalService 服务市场

### 架构设计理念

**为什么需要三层架构？**

| 问题 | 单层方案 | 三层方案 |
|------|---------|----------|
| **配置千差万别** | ❌ 需要 N 张表 | ✅ JSON 一张表 |
| **如何防止乱填** | ❌ 手动校验每个字段 | ✅ Schema 自动验证 |
| **前端如何知道填什么** | ❌ 硬编码表单 | ✅ 动态渲染 |
| **后端类型安全** | ❌ map[string]interface{} | ✅ Go Struct |

### 三层架构详解

#### 第一层：JSON 存储

type ExternalService struct { ID string gorm:"primaryKey;type:varchar(36)" json:"id" Category string gorm:"type:varchar(32);index" json:"category" ServiceType string gorm:"type:varchar(64);index" json:"serviceType" Name string gorm:"type:varchar(64)" json:"name" Tags string gorm:"type:varchar(255)" json:"tags" // JSON 数组 Config string gorm:"type:text;not null" json:"config" // ← JSON 字符串 Enabled bool gorm:"default:true;index" json:"enabled" }


**示例数据**：

{ "category": "networking", "serviceType": "frp_server", "name": "我的 FRP 服务器", "config": "{"server_addr":"frp.example.com","token":"xxx"}" }


#### 第二层：Schema 验证

**FRP 服务器配置 Schema**：

func (p *FRPServerProvider) ConfigSchema() string { return { "type": "object", "required": ["server_addr", "token"], "properties": { "server_addr": { "type": "string", "title": "FRP 服务器地址" }, "server_port": { "type": "integer", "default": 7000, "minimum": 1, "maximum": 65535 }, "token": { "type": "string", "format": "password", "minLength": 8 }, "protocol": { "type": "string", "enum": ["tcp", "kcp", "quic"], "default": "tcp" } } } }


#### 第三层：Struct 类型转换

type FRPConfig struct { ServerAddr string json:"server_addr" ServerPort int json:"server_port,omitempty" Token string json:"token" Protocol string json:"protocol,omitempty" }

func (p *FRPServerProvider) ValidateConfig(configJSON string) error { var cfg FRPConfig json.Unmarshal([]byte(configJSON), &cfg)

if cfg.ServerAddr == "" {
    return errors.New("服务器地址不能为空")
}
if cfg.Token == "" {
    return errors.New("认证令牌不能为空")
}
if len(cfg.Token) < 8 {
    return errors.New("认证令牌长度至少 8 位")
}

return nil

}

func (p *FRPServerProvider) BuildConfig(configJSON string) (*FRPConfig, error) { var cfg FRPConfig json.Unmarshal([]byte(configJSON), &cfg)

// 设置默认值
if cfg.ServerPort == 0 {
    cfg.ServerPort = 7000
}
if cfg.Protocol == "" {
    cfg.Protocol = "tcp"
}

return &cfg, nil

}


### 复杂场景支持

#### TURN 服务器多种认证方式

type TURNConfig struct { ServerAddr string json:"server_addr" Realm string json:"realm,omitempty"

// 认证方式（互斥）
AuthType  string         `json:"auth_type"`  // long_term | short_term | auth_secret
LongTerm  *LongTermAuth  `json:"long_term,omitempty"`
ShortTerm *ShortTermAuth `json:"short_term,omitempty"`

}

type LongTermAuth struct { Username string json:"username" Password string json:"password" }

type ShortTermAuth struct { Username string json:"username" AuthSecret string json:"auth_secret" ExpiresIn int json:"expires_in,omitempty" }

// GetCredentials 动态获取凭证 func (p *TURNServerProvider) GetCredentials(configJSON string) (Credentials, error) { var cfg TURNConfig json.Unmarshal([]byte(configJSON), &cfg)

switch cfg.AuthType {
case "long_term":
    return &LongTermCredentials{
        Username: cfg.LongTerm.Username,
        Password: cfg.LongTerm.Password,
    }, nil
    
case "short_term":
    // 动态生成短期凭证（HMAC-SHA1）
    return p.generateShortTermCredentials(cfg.ShortTerm)
    
case "auth_secret":
    return p.generateAuthSecretCredentials(cfg.AuthSecret)
    
default:
    return nil, errors.New("不支持的认证方式")
}

}


#### WS 隧道多种部署方式

type WSTunnelConfig struct { // 服务器类型（仅用于 UI 分类） ServerType string json:"server_type" // coturn | nginx | standalone

// 统一的 WebSocket 端点
EndpointType string `json:"endpoint_type"`  // ws | wss
Endpoint     string `json:"endpoint"`       // ws://example.com:80/ws

// 是否关联 STUN（如果是 coturn）
UseCoturnSTUN bool   `json:"use_coturn_stun"`
STUNEndpoint  string `json:"stun_endpoint,omitempty"`

}

func (p *WSTunnelProvider) BuildTransportConfig(configJSON string) (*TransportConfig, error) { var cfg WSTunnelConfig json.Unmarshal([]byte(configJSON), &cfg)

u, _ := url.Parse(cfg.Endpoint)

return &TransportConfig{
    Type:     "ws_tunnel",
    Protocol: u.Scheme,
    Host:     u.Host,
    Path:     u.Path,
    
    // 如果是 coturn 且启用了 STUN，自动注入 STUN 配置
    STUNServers: func() []string {
        if cfg.UseCoturnSTUN && cfg.STUNEndpoint != "" {
            return []string{cfg.STUNEndpoint}
        }
        return nil
    }(),
}, nil

}


---

## 5. MeshSeed 安全机制

### 双层安全设计

┌─────────────────────────────────────┐ │ 外层：AES-256-GCM 加密 │ │ - 保护机密性 │ │ - 防止偷看 DNS 记录 │ │ - 密钥派生：SHA256("meshray-ddns" + NetworkSecret) │ └─────────────────────────────────────┘ ↓ 包裹着 ┌─────────────────────────────────────┐ │ 内层：Ed25519 签名 │ │ - 保护完整性与来源 │ │ - 防止中间人篡改 │ │ - 公钥从 seed.IssuerNodeID 解析 │ └─────────────────────────────────────┘


### 加密流程

// pkg/meshseed/crypto.go

// 1. 派生 AES-256-GCM 密钥 func deriveKey(networkSecret string) []byte { hash := sha256.Sum256([]byte("meshray-ddns" + networkSecret)) return hash[:] }

// 2. 加密 MeshSeed func Encrypt(seed *MeshSeed, networkSecret string) (string, error) { // 签名（如果还没有） if len(seed.Signature) == 0 { return "", fmt.Errorf("MeshSeed 未签名") }

// 序列化
plaintext, _ := json.Marshal(seed)

// 派生密钥
key := deriveKey(networkSecret)

// AES-GCM 加密
block, _ := aes.NewCipher(key)
gcm, _ := cipher.NewGCM(block)
nonce := make([]byte, gcm.NonceSize())
rand.Read(nonce)

ciphertext := gcm.Seal(nonce, nonce, plaintext, nil)

// Base64URL 编码
return base64.URLEncoding.EncodeToString(ciphertext), nil

}

// 3. 解密 MeshSeed func Decrypt(encrypted string, networkSecret string) (*MeshSeed, error) { // Base64URL 解码 ciphertext, _ := base64.URLEncoding.DecodeString(encrypted)

// 派生密钥
key := deriveKey(networkSecret)

// AES-GCM 解密
block, _ := aes.NewCipher(key)
gcm, _ := cipher.NewGCM(block)
nonceSize := gcm.NonceSize()

nonce, ciphertextBytes := ciphertext[:nonceSize], ciphertext[nonceSize:]
plaintext, _ := gcm.Open(nil, nonce, ciphertextBytes, nil)

// 反序列化
var seed MeshSeed
json.Unmarshal(plaintext, &seed)

return &seed, nil

}


### DDNS TXT 记录格式

DNS TXT 记录

_meshray.home.example.com. IN TXT "meshray-ddns:<base64_url_encoded_data>"

数据结构

base64_url_encoded_data = nonce (12 字节) + ciphertext + tag (16 字节)


### 完整使用流程

#### Publish 发布 MeshSeed

// internal/provider/sync/libdns_adapter.go

func (a *LibdnsAdapter) Publish(configJSON string, seed *meshseed.MeshSeed) error { // 1. 解析配置 var cfg LibdnsConfig json.Unmarshal([]byte(configJSON), &cfg)

// 2. 加密 MeshSeed
encrypted, err := meshseed.Encrypt(seed, seed.NetworkSecret)
if err != nil {
    return fmt.Errorf("加密失败：%w", err)
}

// 3. 构造 TXT 记录值
txtValue := "meshray-ddns:" + encrypted

// 4. 写入 DNS TXT 记录
ctx := context.Background()
_, err = a.appender.AppendRecords(ctx, cfg.Domain, []libdns.Record{
    libdns.RR{
        Type: "TXT",
        Name: cfg.TXTRecordName,
        Data: txtValue,
        TTL:  time.Duration(cfg.TTL) * time.Second,
    },
})

return err

}


#### Fetch 拉取 MeshSeed

func (a *LibdnsAdapter) Fetch(configJSON string, networkID uint64) (*meshseed.MeshSeed, error) { // 1. 解析配置 var cfg LibdnsConfig json.Unmarshal([]byte(configJSON), &cfg)

// 2. 读取 DNS TXT 记录
ctx := context.Background()
records, _ := a.getter.GetRecords(ctx, cfg.Domain)

// 3. 查找匹配的记录
for _, record := range records {
    rr := record.RR()
    if rr.Type != "TXT" || rr.Name != cfg.TXTRecordName {
        continue
    }
    
    // 4. 提取加密数据
    txtValue := rr.Data
    encrypted := strings.TrimPrefix(txtValue, "meshray-ddns:")
    
    // 5. 解密 MeshSeed（需要 NetworkSecret）
    seed, err := meshseed.Decrypt(encrypted, networkSecret)
    if err != nil {
        continue  // 密钥不匹配或数据被篡改
    }
    
    // 6. 验证 NetworkID
    if seed.NetworkID != networkID {
        continue
    }
    
    // 7. Ed25519 验签
    if err := meshseed.Verify(seed, nil); err != nil {
        continue  // 签名无效
    }
    
    return seed, nil
}

return nil, fmt.Errorf("未找到 MeshSeed 记录")

}


---

*最后更新：v2.1.0*

24 KiB Raw Permalink Blame History Unescape Escape