fix(channel,user): owner 车道只认白名单,堵住群主/伪造 role 冒充主人(裁决 06.4.2/06.4.3/06.4.4)

- 适配器只描述平台身份:qq_obv11 的 owner/admin 一律上报 RoleMember,不再上报 RoleOwner
- session 层唯一判主点只认白名单(owner.yaml Accounts ∪ 每渠道 adminIds),nil 档案 fail-closed 拒绝
- 新增「每渠道管理员层」:channels.json ChannelAccount.adminIds[] → runtimeconfig.AdminIDs → OwnerProfile.SetPlatformAdmins(整体替换语义,热重载后移除的 id 立即失效)
- 组合根接线:启动即播种、面板保存先灌白名单再重载适配器(cmd/cyrene/wire_channels.go + 4 条端到端单测)
- 06.4.3 节流与参与度状态按会话独立计数(ConversationKey 三处统一,空会话 GC);06.4.4 未知平台主动发言保守默认 1 条/分钟
This commit is contained in:
AskaEth committed 2026-10-04 12:07:40 +08:00
1 parent 7e7f53bb41
commit af8591b78a
30 files changed
+1289 -144

No files matched your search

+3 -1
View File
@@ -8,7 +8,9 @@ package adapter
type Role string
const (
// RoleOwner is the bound owner/primary account.
// RoleOwner is the bound owner/primary account. 只有渠道层的白名单判主可以赋这个角色
// (owner.yaml 账号白名单 + 每渠道 adminIds[],裁决 06.4.2):适配器**不得**从平台字段
// (如 OneBot 的 sender.role=owner)直接上报它——群主/群管理员都上报 RoleMember。
RoleOwner Role = "owner"
// RoleMember is a non-owner participant (for example a stranger in a group).
RoleMember Role = "member"
+123 -40
View File
@@ -31,19 +31,26 @@ type GateVerdict struct {
Pruned []UnifiedMessage
}
// FlowGateOptions configures a per-adapter ingress gate. Every limit is
// FlowGateOptions configures a per-account ingress gate. Every limit is
// independent and a zero value disables that limit, so an account that does not
// declare the new fields keeps its previous unlocked behavior (backward
// compatible, design/12 §6 实现要点).
//
// **计数键 = 会话级**(裁决 06.4.3):GroupMsgPerMin 与 MaxMessages 都按
// ConversationKey(`platform+channelID+groupID/peerID`)每个群聊/私聊独立计数,不再在
// 适配器顶层汇总——两个群并发互不挤占。MsgPerMin 是防刷屏的**账号级总量护栏**,按设计
// 口径仍留在适配器顶层,不随会话拆分。
type FlowGateOptions struct {
// GroupMsgPerMin caps group-chat inbound messages per minute (design/12 §6
// 策略三). Zero disables the group limit.
// GroupMsgPerMin caps group-chat inbound messages per minute **per
// conversation** (design/12 §6 策略三 / 裁决 06.4.3). Zero disables the limit.
GroupMsgPerMin int
// MsgPerMin caps total inbound non-owner messages per minute. Zero disables it.
// MsgPerMin caps total inbound non-owner messages per minute across the whole
// account — the anti-flood total guard. Zero disables it.
MsgPerMin int
// MaxMessages is the total pending-message cap. When the pending set nears it
// the Reviewer is invoked to prune noise; with no Reviewer the oldest messages
// are evicted FIFO. Zero disables the cap (design/12 §6 策略四).
// MaxMessages is the **per-conversation** pending-message cap (design/12 §6
// 策略四 / 裁决 06.4.3). When a conversation's pending set nears it the Reviewer
// is invoked to prune noise; with no Reviewer the oldest messages are evicted
// FIFO. Zero disables the cap.
MaxMessages int
// Window is the sliding window over which the per-minute limits are counted.
// Zero uses defaultRateWindow.
@@ -72,9 +79,45 @@ const (
// FlowCapDefault is the protective default total-cap of 50 pending messages
// (design/12 §6 策略四).
FlowCapDefault = 50
// sessionGCInterval is how many admissions pass between idle-conversation sweeps,
// so a long-lived gate that has seen many conversations does not grow unbounded.
sessionGCInterval = 256
)
// FlowGate is a concurrency-safe, per-adapter ingress gate implementing the two
// ConversationKey 返回一条消息所属会话的节流键(裁决 06.4.3):
// `platform+channelID+groupID/peerID`——群聊按 (平台, 群) 独立计数,私聊按 (平台, 对端)
// 独立计数。它同时被入站滑窗(本文件的 FlowGate)与出处参与引擎
// (channelbridge.engageKey) 使用,使「每会话独立计数」只有一处口径。
// 多账号(platform:confId)之间的隔离由每个账号各自的 FlowGate/适配器实例承担。
func ConversationKey(msg UnifiedMessage) string {
conv := msg.GroupID
if msg.IsGroup() {
if conv == "" {
// 群聊缺 group id 时退回 channel id(qq-obv11 里 channelID 就是群号),
// 避免同一个群被拆成多个桶而让限流失效。
conv = msg.ChannelID
}
} else {
conv = msg.SenderID
if conv == "" {
conv = msg.Sender.UserID
}
if conv == "" {
conv = msg.ChannelID
}
}
return msg.Platform + "|" + msg.ChannelID + "|" + conv
}
// flowSession 是一个会话(群聊/私聊)独立的节流状态。FlowGate 按 ConversationKey 为
// 每个会话保存一份,因此两个群并发互不挤占(裁决 06.4.3)。
type flowSession struct {
groupHits []time.Time
buffer []queuedMsg
lastReview time.Time
}
// FlowGate is a concurrency-safe, per-account ingress gate implementing the two
// measures from design/12 §6 策略三/四:
//
// - 策略三: a sliding-window rate limit on group chat (and, optionally, all
@@ -84,6 +127,8 @@ const (
// invoked to prune low-value messages; with no Reviewer the oldest messages are
// evicted FIFO. Either way every dropped message is audited.
//
// 两个策略的计数都按会话独立(裁决 06.4.3),只有 MsgPerMin 总量护栏按账号汇总。
//
// Owner-private messages are not routed through the gate (the caller bypasses it),
// so the owner is never throttled or pruned. Identity is never examined here.
type FlowGate struct {
@@ -96,12 +141,11 @@ type FlowGate struct {
onPruned func(UnifiedMessage, string)
now func() time.Time
mu sync.Mutex
groupHits []time.Time
totalHits []time.Time
buffer []queuedMsg
lastReview time.Time
seq uint64
mu sync.Mutex
totalHits []time.Time
sessions map[string]*flowSession
admitted uint64
seq uint64
}
// queuedMsg wraps a buffer entry with a stable sequence so pruning decisions can
@@ -123,6 +167,7 @@ func NewFlowGate(opts FlowGateOptions) *FlowGate {
review: opts.Review,
onPruned: opts.OnPruned,
now: opts.Now,
sessions: make(map[string]*flowSession),
}
if g.window <= 0 {
g.window = defaultRateWindow
@@ -139,23 +184,30 @@ func NewFlowGate(opts FlowGateOptions) *FlowGate {
return g
}
// Admit applies the rate limit then the total-cap/review gate to one inbound
// message. It returns a verdict telling the caller whether to proceed; messages
// that the gate drops are returned (and audited) so nothing is lost silently.
// Admit applies the rate limits then the per-conversation cap/review gate to one
// inbound message. It returns a verdict telling the caller whether to proceed;
// messages that the gate drops are returned (and audited) so nothing is lost
// silently.
//
// 计数按会话独立(裁决 06.4.3):群聊滑窗与待处理上限都落在 ConversationKey 对应的
// flowSession 上(两个群并发互不挤占),只有 MsgPerMin 这一账号级总量护栏仍按适配器汇总。
func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
now := g.now()
key := ConversationKey(msg)
g.mu.Lock()
defer g.mu.Unlock()
g.maybeGCLocked(now)
// 策略三: sliding-window rate limits. Group chat has its own cap; MsgPerMin is
// an optional total cap across all non-owner messages. Over-limit messages are
// dropped at the door (downgraded) instead of blocking.
// 策略三(会话级): sliding-window rate limits. 群聊上限按每个群独立计数;MsgPerMin
// 是账号级的非主人消息总量护栏。Over-limit messages are dropped at the door
// (downgraded) instead of blocking.
if g.groupPerMin > 0 && msg.ChannelType == ChannelGroup {
g.groupHits = g.trimHits(g.groupHits, now)
if len(g.groupHits) >= g.groupPerMin {
st := g.sessionLocked(key)
st.groupHits = g.trimHits(st.groupHits, now)
if len(st.groupHits) >= g.groupPerMin {
return g.drop("rate_limited", msg)
}
g.groupHits = append(g.groupHits, now)
st.groupHits = append(st.groupHits, now)
}
if g.msgPerMin > 0 {
g.totalHits = g.trimHits(g.totalHits, now)
@@ -169,29 +221,31 @@ func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
return GateVerdict{Proceed: true}
}
// 策略四: add to the pending set, then prune when it nears the cap.
// 策略四(会话级): add to this conversation's pending set, then prune when it
// nears the cap.
st := g.sessionLocked(key)
g.seq++
seq := g.seq
g.buffer = append(g.buffer, queuedMsg{seq: seq, msg: msg})
st.buffer = append(st.buffer, queuedMsg{seq: seq, msg: msg})
var dropped []queuedMsg
if g.review != nil && len(g.buffer) >= g.reviewAt && !g.reviewRecently(now) {
keep, err := g.review(ctx, queueMessages(g.buffer))
if g.review != nil && len(st.buffer) >= g.reviewAt && !g.reviewRecently(st, now) {
keep, err := g.review(ctx, queueMessages(st.buffer))
if err == nil {
g.buffer, dropped = g.applyReview(g.buffer, keep)
g.lastReview = now
st.buffer, dropped = g.applyReview(st.buffer, keep)
st.lastReview = now
g.auditDropped(dropped, "review_pruned")
}
}
if len(g.buffer) > g.maxMessages {
n := len(g.buffer) - g.maxMessages
if len(st.buffer) > g.maxMessages {
n := len(st.buffer) - g.maxMessages
var evicted []queuedMsg
g.buffer, evicted = g.dropFront(g.buffer, n)
st.buffer, evicted = g.dropFront(st.buffer, n)
dropped = append(dropped, evicted...)
g.auditDropped(evicted, "capped")
}
if !g.containsSeq(seq) {
if !bufferHasSeq(st.buffer, seq) {
reason := "capped"
if containsSeq(dropped, seq) {
reason = "review_pruned"
@@ -213,13 +267,41 @@ func (g *FlowGate) drop(reason string, msg UnifiedMessage) GateVerdict {
return v
}
// sessionLocked 返回 key 对应的会话节流状态,首次使用时惰性创建。调用方须持有 g.mu。
func (g *FlowGate) sessionLocked(key string) *flowSession {
if g.sessions == nil {
g.sessions = make(map[string]*flowSession)
}
st := g.sessions[key]
if st == nil {
st = &flowSession{}
g.sessions[key] = st
}
return st
}
// maybeGCLocked 每 sessionGCInterval 次放行清理一次空闲会话,避免会话状态表随历史会话
// 无界增长:滑窗时间戳全部过期且待处理队列为空的会话直接删除。调用方须持有 g.mu。
func (g *FlowGate) maybeGCLocked(now time.Time) {
g.admitted++
if g.admitted%sessionGCInterval != 0 {
return
}
for key, st := range g.sessions {
st.groupHits = g.trimHits(st.groupHits, now)
if len(st.groupHits) == 0 && len(st.buffer) == 0 {
delete(g.sessions, key)
}
}
}
// reviewRecently throttles review so it does not run on every message: it fires at
// most once per window, which is the cost guard for the review model.
func (g *FlowGate) reviewRecently(now time.Time) bool {
if g.lastReview.IsZero() {
// most once per window per conversation, which is the cost guard for the review model.
func (g *FlowGate) reviewRecently(st *flowSession, now time.Time) bool {
if st == nil || st.lastReview.IsZero() {
return false
}
return now.Sub(g.lastReview) < g.window
return now.Sub(st.lastReview) < g.window
}
// trimHits drops event timestamps older than the sliding window.
@@ -265,9 +347,10 @@ func (g *FlowGate) dropFront(buffer []queuedMsg, n int) (kept, dropped []queuedM
return buffer[n:], append([]queuedMsg(nil), buffer[:n]...)
}
// containsSeq reports whether an entry with the given sequence is still buffered.
func (g *FlowGate) containsSeq(seq uint64) bool {
for _, e := range g.buffer {
// bufferHasSeq reports whether an entry with the given sequence is still buffered
// in the conversation's pending set.
func bufferHasSeq(buffer []queuedMsg, seq uint64) bool {
for _, e := range buffer {
if e.seq == seq {
return true
}
@@ -0,0 +1,132 @@
package adapter
// 本文件覆盖裁决 06.4.3:节流(群聊滑窗与待处理上限)的计数键是**会话级**
// (ConversationKey = platform+channelID+groupID/peerID),不在适配器顶层汇总——
// 两个群并发互不挤占。
import (
"context"
"testing"
"time"
)
// groupMsgIn 构造指定群的群聊消息(platform kook),用于验证会话级计数。
func groupMsgIn(groupID string, i int) UnifiedMessage {
return UnifiedMessage{
Platform: "kook",
ChannelID: groupID,
ChannelType: ChannelGroup,
GroupID: groupID,
SenderID: "u" + itoa(i),
Content: "msg " + itoa(i),
ContentType: ContentText,
}
}
// directMsgTo 构造指定对端的私聊消息,用于验证私聊额度独立。
func directMsgTo(peer string, i int) UnifiedMessage {
return UnifiedMessage{
Platform: "telegram",
ChannelID: peer,
ChannelType: ChannelDirect,
SenderID: peer,
Content: "dm " + itoa(i),
ContentType: ContentText,
}
}
// TestConversationKeySessionScoped 固定会话键口径(裁决 06.4.3):群聊按 (平台, 群)
// 独立、私聊按 (平台, 对端) 独立,不同会话的键互不相等。
func TestConversationKeySessionScoped(t *testing.T) {
t.Parallel()
if ConversationKey(groupMsgIn("-100", 0)) != ConversationKey(groupMsgIn("-100", 1)) {
t.Error("同一个群的两条消息应落在同一个会话键")
}
if ConversationKey(groupMsgIn("-100", 0)) == ConversationKey(groupMsgIn("-200", 0)) {
t.Error("两个群共用了同一个会话键(会话级计数失效)")
}
if ConversationKey(directMsgTo("peer-1", 0)) == ConversationKey(directMsgTo("peer-2", 0)) {
t.Error("两个私聊对端共用了同一个会话键")
}
if ConversationKey(directMsgTo("peer-1", 0)) == ConversationKey(groupMsgIn("-100", 0)) {
t.Error("私聊与群聊共用了同一个会话键")
}
// 群聊缺 group id 时退回 channel id:仍按群独立,不得退化成按发送者分桶
// (否则限流会被发言者数量放大而失效)。
bare := UnifiedMessage{Platform: "kook", ChannelType: ChannelGroup, ChannelID: "-100"}
withSender := UnifiedMessage{
Platform: "kook", ChannelType: ChannelGroup, ChannelID: "-100", SenderID: "u9",
}
if ConversationKey(bare) != ConversationKey(withSender) {
t.Error("群聊缺 group id 时退化成按发送者分桶")
}
}
// TestFlowGate_GroupRateLimitIsPerConversation 是裁决 06.4.3 的核心断言:群聊滑窗按
// 每个群独立计数——A 群用尽额度不挤占 B 群(两个群并发互不挤占)。
func TestFlowGate_GroupRateLimitIsPerConversation(t *testing.T) {
t.Parallel()
now, clock := fixedClock(time.Unix(1000, 0))
g := NewFlowGate(FlowGateOptions{GroupMsgPerMin: 2, Window: time.Minute, Now: now})
ctx := context.Background()
for i := 0; i < 2; i++ {
if v := g.Admit(ctx, groupMsgIn("-100", i)); !v.Proceed {
t.Fatalf("A 群第 %d 条被拒,want admitted: %+v", i+1, v)
}
}
if v := g.Admit(ctx, groupMsgIn("-100", 2)); v.Proceed {
t.Fatal("A 群超出自身额度仍被放行")
}
for i := 0; i < 2; i++ {
if v := g.Admit(ctx, groupMsgIn("-200", i)); !v.Proceed {
t.Fatalf("B 群第 %d 条被 A 群额度挤占: %+v", i+1, v)
}
}
if v := g.Admit(ctx, groupMsgIn("-200", 9)); v.Proceed {
t.Fatal("B 群超出自身额度仍被放行")
}
// 私聊有自己的额度,不参与群聊滑窗。
if v := g.Admit(ctx, directMsgTo("peer-1", 0)); !v.Proceed {
t.Fatalf("私聊被群聊额度挤占: %+v", v)
}
// 窗口滚过之后 A 群额度恢复。
*clock = clock.Add(61 * time.Second)
if v := g.Admit(ctx, groupMsgIn("-100", 3)); !v.Proceed {
t.Fatalf("窗口滚过后 A 群应恢复放行: %+v", v)
}
}
// TestFlowGate_CapIsPerConversation 覆盖策略四同样按会话独立:一个群堆满待处理消息
// 触发的是本群的 FIFO 淘汰,不会把另一个群的待处理消息挤掉。
func TestFlowGate_CapIsPerConversation(t *testing.T) {
t.Parallel()
g := NewFlowGate(FlowGateOptions{MaxMessages: 2})
ctx := context.Background()
for i := 0; i < 2; i++ {
if v := g.Admit(ctx, groupMsgIn("-100", i)); !v.Proceed {
t.Fatalf("A 群第 %d 条被拒: %+v", i+1, v)
}
}
v := g.Admit(ctx, groupMsgIn("-100", 2))
if !v.Proceed {
t.Fatalf("A 群新消息在 FIFO 淘汰后应放行: %+v", v)
}
if len(v.Pruned) != 1 || v.Pruned[0].ChannelID != "-100" {
t.Fatalf("A 群淘汰的应是本群消息, got %+v", v.Pruned)
}
for i := 0; i < 2; i++ {
if v := g.Admit(ctx, groupMsgIn("-200", i)); !v.Proceed || len(v.Pruned) != 0 {
t.Fatalf("B 群被 A 群的待处理队列挤占: %+v", v)
}
}
}