Five operational gaps that bite when a connector-backed bot answers questions in real groups: - Single-instance lock: one connector per robot per machine (pid file, stale-lock takeover). Duplicate Stream connections on one clientId get messages load-balanced between them, so the bot answers intermittently (verified live). - Access policy: --allowed-users / --allowed-groups allowlists plus a per-sender rate limit (--user-rate-limit, default 20/min). Every message is an LLM call; without this anyone who can reach the bot can spend on its behalf. - Per-conversation FIFO queue: same-chat messages run in arrival order (follow-ups need the previous turn's --resume session; parallel CLIs racing one session corrupt it); different chats stay parallel. - Picture messages: resolve the callback downloadCode via robot messageFiles/download to a local temp file and hand the path to the agent CLI — error screenshots are the top Q&A inbound and were previously dropped silently. The claudecode persona now permits reading attached files. - --knowledge-dir: lightweight local retrieval (markdown chunking, latin-token + CJK-bigram scoring) prepends top-k matching chunks to each question while the agent keeps running from the clean scratch dir, so replies stay inside DingTalk's AI-assistant response window. All knobs have env mirrors (DWS_ALLOWED_USERS / DWS_ALLOWED_GROUPS / DWS_USER_RATE_LIMIT / DWS_KNOWLEDGE_DIR) for service-style runs. go build ./... clean; go test ./internal/helpers -race passes with new tests for lock/gate/queue/knowledge/media. test/cli_compat, test/scripts and test/unit fail identically on the unmodified base (pre-existing fixture issues).
116 lines
3.4 KiB
Go
116 lines
3.4 KiB
Go
// Copyright 2026 Alibaba Group
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package helpers
|
|
|
|
import (
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// connectGate enforces the connector's access policy: optional user/group
|
|
// allowlists plus a per-sender rate limit. A long-lived Q&A bot is otherwise
|
|
// drivable (and billable — every message is an LLM call) by anyone who can
|
|
// reach it in a group.
|
|
type connectGate struct {
|
|
allowedUsers map[string]struct{}
|
|
allowedGroups map[string]struct{}
|
|
perMinute int
|
|
|
|
mu sync.Mutex
|
|
hits map[string][]time.Time
|
|
now func() time.Time // test hook
|
|
}
|
|
|
|
// splitCommaList splits a comma-separated flag/env value, trimming blanks.
|
|
func splitCommaList(s string) []string {
|
|
var out []string
|
|
for _, p := range strings.Split(s, ",") {
|
|
if p = strings.TrimSpace(p); p != "" {
|
|
out = append(out, p)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// newConnectGate builds a gate from staffId / openConversationId allowlists
|
|
// and a per-sender messages-per-minute cap (0 = unlimited).
|
|
func newConnectGate(users, groups []string, perMinute int) *connectGate {
|
|
g := &connectGate{perMinute: perMinute, hits: map[string][]time.Time{}, now: time.Now}
|
|
for _, u := range users {
|
|
if u = strings.TrimSpace(u); u != "" {
|
|
if g.allowedUsers == nil {
|
|
g.allowedUsers = map[string]struct{}{}
|
|
}
|
|
g.allowedUsers[u] = struct{}{}
|
|
}
|
|
}
|
|
for _, grp := range groups {
|
|
if grp = strings.TrimSpace(grp); grp != "" {
|
|
if g.allowedGroups == nil {
|
|
g.allowedGroups = map[string]struct{}{}
|
|
}
|
|
g.allowedGroups[grp] = struct{}{}
|
|
}
|
|
}
|
|
return g
|
|
}
|
|
|
|
// enabled reports whether any policy is configured at all.
|
|
func (g *connectGate) enabled() bool {
|
|
return g != nil && (len(g.allowedUsers) > 0 || len(g.allowedGroups) > 0 || g.perMinute > 0)
|
|
}
|
|
|
|
// allow reports whether a message may proceed; on denial, reason names the
|
|
// rule that fired (for the connector log). Group messages must pass the group
|
|
// allowlist AND, when a user allowlist is set, the sender allowlist too.
|
|
func (g *connectGate) allow(staffID, convType, convID string) (bool, string) {
|
|
if g == nil {
|
|
return true, ""
|
|
}
|
|
if convType == "2" && len(g.allowedGroups) > 0 {
|
|
if _, ok := g.allowedGroups[convID]; !ok {
|
|
return false, "group-not-allowed"
|
|
}
|
|
}
|
|
if len(g.allowedUsers) > 0 {
|
|
if _, ok := g.allowedUsers[staffID]; !ok {
|
|
return false, "user-not-allowed"
|
|
}
|
|
}
|
|
if g.perMinute > 0 && staffID != "" {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
now := g.now()
|
|
cutoff := now.Add(-time.Minute)
|
|
kept := g.hits[staffID][:0]
|
|
for _, t := range g.hits[staffID] {
|
|
if t.After(cutoff) {
|
|
kept = append(kept, t)
|
|
}
|
|
}
|
|
if len(kept) >= g.perMinute {
|
|
g.hits[staffID] = kept
|
|
return false, "rate-limited"
|
|
}
|
|
g.hits[staffID] = append(kept, now)
|
|
// Bound memory against sender-id churn: keep only the current sender
|
|
// when the map grows absurdly large.
|
|
if len(g.hits) > 4096 {
|
|
g.hits = map[string][]time.Time{staffID: g.hits[staffID]}
|
|
}
|
|
}
|
|
return true, ""
|
|
}
|