P2P服务DoS防护：基于内容与频率拒绝连接（Golang方案咨询）

问题描述 #

我运行一个P2P服务（无Web服务器、无具备DoS防护功能的负载均衡等组件），接收直接连接请求。需求为：当相同消息的发送频率超过设定阈值（如var threshold = 10）时，拒绝该连接。

当前实现思路：

1. 保存每条请求的时间戳，数量上限为阈值
2. 允许阈值范围内的所有请求
3. 达到阈值时，检查当前时间与首个时间戳的间隔是否超过maxAllowFrequency，若是则断开对等节点，否则允许连接

暂不考虑状态重置逻辑的实现。

请问该方案性能表现如何？是否存在更优或更优雅的实现方式？

附当前Golang代码示例：

const (
  threshold = 10
  maxAllowFrequency = 1 * time.Second
)

type connections struct {
   timestamps []time.Time
}

var received = map[string]*connections

//the listener
received[newMsg].timestamps = append(received[newMsg].timestamps, time.Now())
if len(received[newMsg]) > threshold && time.Now().Sub(received[newMsg].timestamp[0]) > maxAllowFrequency {
  // ban this peer
}

一、当前方案的性能与问题分析 #

1. 并发安全隐患：代码中的received普通map没有任何并发保护，P2P服务是多连接并发处理场景，直接读写会触发竞态条件，导致panic或数据错乱。
2. 内存开销与泄漏风险：每个消息类型维护一个最多存threshold个time.Time的切片，若消息类型多、请求量大且无过期清理，内存会持续累积。
3. 频率控制不精确：仅在请求数超过阈值时才检查时间差，存在逻辑漏洞——比如前9个请求集中在1秒末尾，第10个在1秒后不久，此时1.1秒内有10个请求不会被拦截，但后续短时间内再进10个请求就会过载。
4. 切片操作的GC压力：高频请求下，切片append频繁触发内存扩容与拷贝，增加垃圾回收负担。

二、更优的实现方式 #

1. 滑动窗口：精确控制单位时间请求数 #

滑动窗口能严格限制maxAllowFrequency时间内的请求数，避免固定数量判断的漏洞：

• 维护按时间排序的时间戳队列，新请求到来时先移除窗口外（超过maxAllowFrequency）的所有时间戳
• 再判断队列长度是否小于阈值，是则允许请求并加入队列，否则拒绝

2. 令牌桶：支持突发流量的轻量化控制 #

令牌桶是流量控制经典算法，适合P2P场景下的动态流量：

• 每个消息类型对应一个令牌桶，每秒生成threshold个令牌
• 每次请求消耗1个令牌，无令牌则拒绝，允许短时间内的合理突发请求

3. 基础优化：解决并发与内存问题 #

• 用sync.Map或带sync.RWMutex的普通map处理并发读写
• 后续可添加定期清理逻辑，移除长时间无请求的消息统计数据，避免内存泄漏

优化后代码示例 #

滑动窗口版 #

import (
    "sync"
    "time"
)

const (
    threshold         = 10
    maxAllowFrequency = 1 * time.Second
)

type connectionStats struct {
    timestamps []time.Time
    mu         sync.Mutex
}

var received = sync.Map{}

func handleRequest(msg string) bool {
    stats, _ := received.LoadOrStore(msg, &connectionStats{})
    cs := stats.(*connectionStats)
    cs.mu.Lock()
    defer cs.mu.Unlock()

    now := time.Now()
    // 清理窗口外的过期时间戳
    cutoff := now.Add(-maxAllowFrequency)
    for len(cs.timestamps) > 0 && cs.timestamps[0].Before(cutoff) {
        cs.timestamps = cs.timestamps[1:]
    }

    if len(cs.timestamps) >= threshold {
        // 拒绝连接/拦截请求
        return false
    }
    cs.timestamps = append(cs.timestamps, now)
    return true
}

令牌桶版 #

import (
    "sync"
    "time"
)

const (
    rateLimit    = 10       // 每秒允许的请求数
    bucketCap    = 10       // 桶的最大令牌容量
    refillInterval = 1 * time.Second // 令牌 refill 周期
)

type tokenBucket struct {
    tokens     int
    lastRefill time.Time
    mu         sync.Mutex
}

var buckets = sync.Map{}

func handleRequest(msg string) bool {
    bucket, _ := buckets.LoadOrStore(msg, &tokenBucket{tokens: bucketCap, lastRefill: time.Now()})
    tb := bucket.(*tokenBucket)
    tb.mu.Lock()
    defer tb.mu.Unlock()

    now := time.Now()
    // 计算需要补充的令牌数量
    elapsed := now.Sub(tb.lastRefill)
    refillCount := int(elapsed / refillInterval * rateLimit)
    if refillCount > 0 {
        tb.tokens = min(tb.tokens+refillCount, bucketCap)
        tb.lastRefill = now
    }

    if tb.tokens <= 0 {
        return false
    }
    tb.tokens--
    return true
}

func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}

内容的提问来源于stack exchange，提问作者unsafe_where_true