优化WebRTC信令服务器抗攻击能力的限流熔断技巧

-- lib/ratelimit.lua
local _M = { _VERSION = '1.0.2' }
local redis = require "resty.redis"
local cjson = require "cjson.safe"

-- 连接池复用，避免频繁握手
local function get_redis()
    local red = redis:new()
    red:set_timeouts(50, 200, 200) -- connect, read, send timeout (ms)
    local ok, err = red:connect("unix:/tmp/redis.sock") -- 或 TCP 地址
    if not ok then return nil, err end
    return red
end

-- 通用令牌桶原子脚本 (KEYS[1]=key, ARGV[1]=rate, ARGV[2]=burst, ARGV[3]=cost, ARGV[4]=ttl)
local TOKEN_BUCKET_SCRIPT = [[
    local key, rate, burst, cost, ttl = KEYS[1], tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3]), tonumber(ARGV[4])
    local now = tonumber(ARGV[5])
    local last_tokens, last_refill = redis.call('HMGET', key, 'tokens', 'refill')
    last_tokens = tonumber(last_tokens) or burst
    last_refill = tonumber(last_refill) or now
    -- 补充令牌
    local delta = math.floor((now - last_refill) * rate)
    local tokens = math.min(burst, last_tokens + delta)
    if tokens >= cost then
        tokens = tokens - cost
        redis.call('HMSET', key, 'tokens', tokens, 'refill', now)
        redis.call('EXPIRE', key, ttl)
        return {1, tokens} -- 允许
    else
        redis.call('HMSET', key, 'tokens', tokens, 'refill', last_refill)
        redis.call('EXPIRE', key, ttl)
        return {0, tokens} -- 拒绝
    end
]]
local token_bucket_sha

function _M.check_limit(red, key, rate, burst, cost, ttl)
    cost = cost or 1
    ttl = ttl or math.ceil(burst / rate) + 10
    if not token_bucket_sha then
        token_bucket_sha = red:script("LOAD", TOKEN_BUCKET_SCRIPT)
    end
    local ok, err = red:evalsha(token_bucket_sha, 1, key, rate, burst, cost, ttl, ngx.time())
    if not ok then
        ngx.log(ngx.ERR, "ratelimit evalsha failed: ", err, " fallback to allow")
        return true, 999 -- Redis挂了默认放行(降级策略)，防止单点拖垮网关
    end
    return ok[1] == 1, ok[2]
end

return _M

# 1. 共享内存区：存储 IP 级连接计数、握手频次
lua_shared_dict signaling_conn_store 100m;
lua_shared_dict signaling_handshake_store 50m;

# 2. 上游信令集群 (保持长连接)
upstream signaling_backend {
    least_conn; # 最少连接调度，配合熔断更平滑
    server 10.0.1.10:8080 max_fails=2 fail_timeout=10s;
    server 10.0.1.11:8080 max_fails=2 fail_timeout=10s;
    keepalive 500; # 关键：复用后端连接，减少 TIME_WAIT
    keepalive_requests 1000;
    keepalive_timeout 60s;
}

server {
    listen 443 ssl http2;
    server_name signaling.example.com;

    # --- 阶段 A: TCP/握手层限流 (access_by_lua_block 早期执行) ---
    location /signaling/ws {
        # A1: 单 IP 并发 WebSocket 连接数硬限制 (基于共享内存，极快)
        access_by_lua_block {
            local conn_store = ngx.shared.signaling_conn_store
            local ip = ngx.var.binary_remote_addr
            local max_conns = 200 -- 单 IP 最大并发连接数
            local count = conn_store:incr(ip, 1, 0)
            if count > max_conns then
                conn_store:decr(ip, 1)
                ngx.log(ngx.WARN, "IP conn limit exceeded: ", ngx.var.remote_addr)
                return ngx.exit(429) -- Too Many Requests
            end
            -- 连接关闭时递减 (需在 log_by_lua 或 upstream 响应阶段处理，见下文)
            ngx.ctx.conn_limited = true
            ngx.ctx.limit_key = ip
        }

        # A2: 握手频率限流 (Token Bucket + Redis，应对建连风暴)
        access_by_lua_block {
            local rl = require "lib.ratelimit"
            local red = rl.get_redis()
            if red then
                local ip = ngx.var.binary_remote_addr
                local allowed, tokens = rl.check_limit(red, "hs:" .. ip, 20, 50, 1, 60) -- 20r/s, burst 50
                red:set_keepalive(10000, 100)
                if not allowed then
                    ngx.header["X-RateLimit-Remaining"] = tokens
                    return ngx.exit(429)
                end
                ngx.header["X-RateLimit-Remaining"] = tokens
            end
        }

        # --- 代理转发配置 ---
        proxy_pass http://signaling_backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header X-Real-IP $remote_addr;
        proxy_read_timeout 3600s; # 长连接超时
        proxy_send_timeout 3600s;
        
        # 关键：WebSocket 升级成功后，记录连接建立；失败/关闭时释放计数
        proxy_set_header X-Nginx-Conn-Key $binary_remote_addr; # 传给后端或日志用
    }

    # --- 阶段 B: 业务语义层限流 (基于 WebSocket Frame 内容，需 ngx_websocket 模块或后端协同) ---
    # 注意：Nginx 原生难以解析 WS Frame 内容，建议此层下沉到 Sidecar 或业务框架中间件 (见第二节)
    # 此处仅做 HTTP 短轮询/REST API 的语义限流示例
    location /api/signaling/offer {
        access_by_lua_block {
            local rl = require "lib.ratelimit"
            local red = rl.get_redis()
            if red then
                -- 基于 UserID (从 JWT 解析) 或 SessionID 限流
                local user_id = ngx.var.http_x_user_id -- 假设网关已解析 JWT 透传 Header
                if not user_id or user_id == "" then user_id = "anon:" .. ngx.var.binary_remote_addr end
                -- Offer 操作：单用户 5次/分钟，突发 10
                local allowed, tokens = rl.check_limit(red, "offer:" .. user_id, 5/60, 10, 1, 120)
                red:set_keepalive(10000, 100)
                if not allowed then
                    ngx.header["Content-Type"] = "application/json"
                    ngx.header["Retry-After"] = 60
                    return ngx.exit(ngx.HTTP_TOO_MANY_REQUESTS)
                end
            end
        }
        proxy_pass http://signaling_backend;
    }

    # --- 连接释放钩子 (log_by_lua_block 保证最后执行) ---
    location /signaling/ws {
        log_by_lua_block {
            if ngx.ctx.conn_limited then
                local conn_store = ngx.shared.signaling_conn_store
                conn_store:decr(ngx.ctx.limit_key, 1)
            end
        }
    }
}

package main

import (
    "context"
    "encoding/json"
    "sync"
    "time"
    "github.com/gin-gonic/gin"
    "github.com/gorilla/websocket"
    "github.com/pion/sdp/v3" // 标准 SDP 解析库
    "github.com/redis/go-redis/v9"
    "golang.org/x/time/rate"
)

type SignalingMsg struct {
    Type    string          `json:"type"` // offer, answer, candidate, renegotiate
    Payload json.RawMessage `json:"payload"`
    SessionID string        `json:"session_id"`
    UserID    string        `json:"user_id"`
}

var (
    redisCli *redis.Client
    // 本地内存限流器：针对高频 Candidate，避免每次走 Redis
    localLimiters = sync.Map{} // key: sessionID -> *rate.Limiter
)

func main() {
    // ... 初始化 Redis, Gin ...
    r := gin.Default()
    r.GET("/ws", wsHandler)
    r.Run(":8081")
}

func wsHandler(c *gin.Context) {
    conn, _ := upgrader.Upgrade(c.Writer, c.Request, nil)
    defer conn.Close()

    // 连接建立即从 Header/Query 获取身份
    userID := c.GetHeader("X-User-ID")
    sessionID := c.Query("session_id")

    for {
        _, msgBytes, err := conn.ReadMessage()
        if err != nil { break }
        
        var msg SignalingMsg
        if json.Unmarshal(msgBytes, &msg) != nil { continue } // 非标准协议直接丢弃/断开

        // --- 语义感知限流决策树 ---
        action := checkSemanticLimit(userID, sessionID, &msg)
        switch action {
        case ActionAllow:
            forwardToBackend(conn, msgBytes) // 透传给后端真实信令服务
        case ActionDropSilent:
            // 静默丢弃 Candidate 风暴，不回包，不断连，攻击者无感知
            continue 
        case ActionRejectWithCode:
            conn.WriteJSON(map[string]string{"error": "rate_limit", "type": msg.Type})
        case ActionCloseConn:
            conn.WriteJSON(map[string]string{"error": "sdp_malformed"})
            return // 主动断开
        }
    }
}

func checkSemanticLimit(userID, sessionID string, msg *SignalingMsg) Action {
    switch msg.Type {
    case "offer", "answer":
        // 1. SDP 结构合规性校验 (防御畸形包/注入)
        if !validateSDP(msg.Payload) { return ActionCloseConn }
        // 2. Redis 全局限流：单用户 5次/分钟
        if !redisAllow("sdp:" + userID, 5, time.Minute) { return ActionRejectWithCode }
        return ActionAllow

    case "candidate":
        // 3. 本地令牌桶：单会话 30/s，突发 100 (无需走 Redis，性能最高)
        limiter := getLocalLimiter(sessionID, 30, 100)
        if !limiter.Allow() { return ActionDropSilent } // 静默丢弃最有效
        return ActionAllow

    case "renegotiate":
        // 4. 重协商极低频：单会话 3次/小时
        if !redisAllow("renego:" + sessionID, 3, time.Hour) { return ActionRejectWithCode }
        return ActionAllow
    }
    return ActionAllow
}

// SDP 合规性校验示例
func validateSDP(payload json.RawMessage) bool {
    var desc sdp.SessionDescription
    if err := desc.Unmarshal(payload); err != nil { return false }
    // 规则：媒体流数量 <= 5 (音频+视频+屏幕共享+数据通道冗余)
    if len(desc.MediaDescriptions) > 5 { return false }
    for _, m := range desc.MediaDescriptions {
        // 规则：禁止非标准编解码器 PT 动态分配冲突
        // 规则：fmtp 参数长度限制，防缓冲区溢出尝试
        for _, a := range m.Attributes {
            if a.Key == "fmtp" && len(a.Value) > 512 { return false }
        }
    }
    return true
}

func getLocalLimiter(key string, r, b int) *rate.Limiter {
    val, _ := localLimiters.LoadOrStore(key, rate.NewLimiter(rate.Limit(r), b))
    return val.(*rate.Limiter)
}

// shared/proto/signaling_errors.proto
enum SignalErrorCode {
  UNKNOWN = 0;
  RATE_LIMIT_CONN = 1001;      // 连接数限流，建议指数退避 + 抖动
  RATE_LIMIT_SDP = 1002;       // SDP 限流，建议固定间隔重试
  RATE_LIMIT_CANDIDATE = 1003; // Candidate 限流，建议本地合并/去抖后重发
  SERVER_OVERLOAD = 1004;      // 服务端过载熔断，携带 retry_after_ms 字段
  SDP_MALFORMED = 1005;        // 协议不合规，禁止重试，上报崩溃日志
  MIGRATION_REQUIRED = 1006;   // 服务端发起热迁移，需携带 token 重连
}

message RateLimitInfo {
  int32 retry_after_ms = 1;      // 建议等待时间
  int32 limit = 2;               // 当前限额
  int32 remaining = 3;           // 剩余配额
  string reset_at = 4;           // RFC3339 重置时间
}

// sdk/src/transport/SignalingTransport.ts
export class SignalingTransport {
  private ws: WebSocket | null = null;
  private retryState = new Map<SignalErrorCode, RetryContext>();
  
  // 指数退避 + 抖动 + 最大上限 + 服务端建议时间取最大值
  private calcBackoff(code: SignalErrorCode, serverSuggestMs?: number): number {
    const ctx = this.retryState.get(code) || { attempt: 0, baseDelay: 1000 };
    ctx.attempt++;
    const expDelay = Math.min(ctx.baseDelay * Math.pow(2, ctx.attempt - 1), 30000); // 最大 30s
    const jitter = Math.random() * 0.3 * expDelay; // 0-30% 抖动，防惊群
    const finalDelay = serverSuggestMs ? Math.max(expDelay + jitter, serverSuggestMs) : expDelay + jitter;
    this.retryState.set(code, ctx);
    return finalDelay;
  }

  async sendWithRetry(msg: SignalingMsg): Promise<SignalingMsg> {
    return new Promise((resolve, reject) => {
      const attemptSend = () => {
        if (this.ws?.readyState !== WebSocket.OPEN) {
          this.reconnect().then(attemptSend).catch(reject);
          return;
        }
        this.ws.send(JSON.stringify(msg));
      };

      // 监听服务端错误回包
      const onMessage = (event: MessageEvent) => {
        const resp = JSON.parse(event.data);
        if (resp.error && ERROR_CODE_MAP[resp.error]) {
          const code = ERROR_CODE_MAP[resp.error];
          const info: RateLimitInfo = resp.info || {};
          
          // 关键：根据错误码分策略
          switch (code) {
            case SignalErrorCode.RATE_LIMIT_CANDIDATE:
              // Candidate 丢包不重发，依赖 ICE 重传机制，仅本地去抖合并
              this.mergePendingCandidates(msg.payload); 
              resolve(resp); // 视为“已处理”
              break;
            case SignalErrorCode.SERVER_OVERLOAD:
            case SignalErrorCode.RATE_LIMIT_CONN:
            case SignalErrorCode.RATE_LIMIT_SDP:
              const delay = this.calcBackoff(code, info.retry_after_ms);
              setTimeout(attemptSend, delay);
              break;
            case SignalErrorCode.MIGRATION_REQUIRED:
              this.migrateAndReconnect(resp.migration_token).then(attemptSend);
              break;
            default:
              reject(new Error(resp.error));
          }
          this.ws?.removeEventListener('message', onMessage);
        } else {
          resolve(resp); // 正常响应
        }
      };
      this.ws?.addEventListener('message', onMessage, { once: true });
      attemptSend();
    });
  }
  
  // 弱网下主动合并 Candidate，减少信令交互
  private candidateBuffer: RTCIceCandidate[] = [];
  private mergeTimer: NodeJS.Timeout | null = null;
  private mergePendingCandidates(candidate: RTCIceCandidate) {
    this.candidateBuffer.push(candidate);
    if (this.mergeTimer) return;
    this.mergeTimer = setTimeout(() => {
      // 仅发送最优候选 (优先 host > srflx > relay) 或批量发送
      const best = this.selectBestCandidates(this.candidateBuffer);
      this.send({ type: 'candidate', payload: best });
      this.candidateBuffer = [];
      this.mergeTimer = null;
    }, 50); // 50ms 合并窗口
  }
}

# istio/destinationrule-signaling.yaml
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: signaling-dr
  namespace: rtc-prod
spec:
  host: signaling.rtc-prod.svc.cluster.local
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 50000      # 单 Sidecar 到上游最大连接数
      http:
        h2UpgradePolicy: UPGRADE   # 支持 WebSocket 升级
        http1MaxPendingRequests: 1000
        http2MaxRequests: 10000
    # 熔断策略：连接池/请求/异常检测
    outlierDetection:
      consecutive5xxErrors: 3      # 连续 3 次 5xx 错误
      interval: 10s                # 检测间隔
      baseEjectionTime: 30s        # 最小剔除 30s
      maxEjectionPercent: 50       # 最大剔除 50% 实例
      minHealthPercent: 30         # 健康实例低于 30% 触发全局熔断(保护剩余节点)
      splitExternalLocalOriginErrors: true
      consecutiveGatewayErrors: 5  # 网关错误 (502/503/504) 熔断
    tls:
      mode: ISTIO_MUTUAL           # mTLS 加密信令通道
  subsets:
  - name: v1
    labels:
      version: v1
  - name: canary
    labels:
      version: canary

# istio/envoyfilter-signaling-ratelimit.yaml
apiVersion: networking.istio.io/v1alpha3
kind: EnvoyFilter
metadata:
  name: signaling-semantic-ratelimit
  namespace: istio-system
spec:
  workloadSelector:
    labels:
      app: signaling
  configPatches:
  - applyTo: HTTP_FILTER
    match:
      context: SIDECAR_INBOUND
      listener:
        filterChain:
          filter:
            name: envoy.filters.network.http_connection_manager
    patch:
      operation: INSERT_BEFORE
      value:
        name: envoy.filters.http.lua
        typed_config:
          "@type": type.googleapis.com/udpa.type.v1.TypedStruct
          type_url: type.googleapis.com/envoy.extensions.filters.http.lua.v3.Lua
          value:
            inline_code: |
              -- Envoy Lua 限流示例：基于 Header 中的 x-user-id + 路径
              function envoy_on_request(request_handle)
                local headers = request_handle:headers()
                local path = headers:get(":path")
                local user_id = headers:get("x-user-id") or "anon"
                
                -- 仅拦截信令核心路径
                if not (path == "/api/signaling/offer" or path == "/api/signaling/answer") then
                  return
                end
                
                -- 调用外部 Redis (需配置 Cluster) 或本地缓存
                -- 此处演示调用外部 RateLimit Service (gRPC)
                local ratelimit_svc = "ratelimit.rtc-prod.svc.cluster.local"
                local key = "sig:" .. user_id .. ":" .. path
                
                -- 实际生产建议用 envoy.filters.http.ratelimit 标准滤镜对接 RLS (RateLimit Service)
                -- Lua 仅做复杂逻辑判断 (如 SDP 解析需 Wasm)
                request_handle:logInfo("Semantic RL Check: " .. key)
              end

# RLS 配置 (config.yaml)
domain: signaling
descriptors:
- key: USER_ID
  rate_limit:
    unit: MINUTE
    requests_per_unit: 20
    # 动态配额：从 Redis 读取用户信誉分调整 requests_per_unit
- key: REMOTE_ADDRESS
  rate_limit:
    unit: SECOND
    requests_per_unit: 100

## 信令节点容量规划表 (版本 v1.2)

| 规格 | CPU | Mem | Max Conns | Max QPS (Offer) | Max QPS (Candidate) | 限流阈值建议 | 熔断触发线 |
| :--- | :--- | :--- | :--- | :--- | :--- | :--- | :--- |
| **标准型** | 4C | 8G | 30,000 | 800 | 5,000 | Conn: 25k / IP: 150 | CPU>75% / Mem>80% / FD>85% |
| **高性能型** | 8C | 16G | 60,000 | 2,000 | 15,000 | Conn: 50k / IP: 300 | CPU>70% / Mem>75% / FD>80% |

* 修订记录：2024-05-20 根据双十一压测数据修正 Candidate QPS 上限 (原 10k -> 15k，优化了内存池)
* 下次复核：2024-08-20 或架构变更后

infra/signaling-resilience/
├── dashboards/           # Grafana JSON (导入即用)
│   ├── signaling-overview.json
│   ├── ratelimit-detail.json
│   └── circuitbreaker-status.json
├── alerts/               # PrometheusRule (kubectl apply -f)
│   ├── signaling-ratelimit.yaml
│   ├── signaling-circuitbreaker.yaml
│   └── signaling-capacity.yaml
├── runbooks/             # Markdown SOP (关联 Alertmanager annotations)
│   ├── HIGH_SIGNALING_RATELIMIT_REJECT.md
│   ├── SIGNALING_NODE_EJECTED.md
│   └── SIGNALING_REDIS_UNAVAILABLE.md
├── chaos/                # ChaosMesh / Litmus 实验定义
│   ├── pod-cpu-stress.yaml
│   ├── network-partition-az.yaml
│   └── redis-failover.yaml
└── configs/              # 限流/熔断参数 ConfigMap
    ├── nginx-ratelimit.lua (ConfigMap)
    ├── sidecar-config.yaml (ConfigMap)
    └── ratelimit-service-config.yaml (ConfigMap)

# 告警：信令限流拒绝率 > 5% 持续 2 分钟

## 1. 现象判断
- Grafana 面板 `signaling-ratelimit-detail` 观察 Top 10 被限流 Key (IP / UserID / SessionID)
- 检查是否为**单一大客户**合规大促，或**分布式攻击** (IP 分散、UA 一致)

## 2. 处置动作 (按顺序执行)
1. [ ] **确认业务影响**：查看 `signaling_session_success_rate` 是否跌破 99%。
2. [ ] **白名单放行 (若为大客户)**：向 Redis `whitelist:signaling` 集合添加 UserID，TTL 2h。
   `redis-cli SADD whitelist:signaling "uid_12345" EX 7200`
3. [ ] **紧急扩容**：手动触发 HPA `kubectl scale deployment signaling --replicas=+3`。
4. [ ] **调整阈值 (谨慎)**：修改 `configs/nginx-ratelimit.lua` 中 `max_conns` 或 `rate`，滚动重载 Nginx `nginx -s reload`。
5. [ ] **溯源分析**：导出被限流 IP 样本，交安全团队研判是否为爬虫/刷单。

## 3. 复盘产出
- 事后 24h 内产出《信令限流误杀/漏放复盘报告》，更新阈值规划表。