mirror of
https://github.com/chenhg5/cc-connect.git
synced 2026-10-25 17:36:43 +08:00
fix: add concurrency safety, error logging, and Windows path hardening
P0 — Feishu platform concurrency safety (platform/feishu/feishu.go): - Add sync.RWMutex to Platform struct - Protect botOpenID, handler, cancel, server fields with getter helpers - Fix TOCTOU race in Stop() by capturing cancel/server under lock P2 — Log silently discarded errors (5 locations): - core/engine.go: cron EnableJob/DisableJob, outgoing rate limit wait - platform/telegram/telegram.go: invalid thread ID parsing - agent/claudecode/session.go: stdin close, SIGTERM, force kill failures - agent/claudecode/claude_usage.go: ptmx close, process kill failures P4 — Windows path hardening (2 locations): - core/message.go: normalize backslashes with filepath.ToSlash + strings.ReplaceAll for cross-platform defense in depth - core/dir_history.go: tighten MkdirAll permissions from 0755 to 0700 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
1 parent
a2cc0ab117
commit
65ff498174
7 files changed
+91
-29
No files matched your search
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"os/exec"
|
||||
"regexp"
|
||||
@@ -119,7 +120,9 @@ func (a *Agent) runClaudeUsageProbe(ctx context.Context) (string, error) {
|
||||
}()
|
||||
|
||||
defer func() {
|
||||
_ = ptmx.Close()
|
||||
if err := ptmx.Close(); err != nil {
|
||||
slog.Warn("claudeSession: close ptmx", "error", err)
|
||||
}
|
||||
cancel()
|
||||
// Wait for reader goroutine to finish so it is never leaked.
|
||||
<-readDone
|
||||
@@ -127,7 +130,9 @@ func (a *Agent) runClaudeUsageProbe(ctx context.Context) (string, error) {
|
||||
case <-processDone:
|
||||
case <-time.After(2 * time.Second):
|
||||
if cmd.Process != nil {
|
||||
_ = cmd.Process.Kill()
|
||||
if err := cmd.Process.Kill(); err != nil {
|
||||
slog.Warn("claudeSession: kill process", "error", err)
|
||||
}
|
||||
}
|
||||
<-processDone
|
||||
}
|
||||
|
||||
@@ -706,7 +706,9 @@ func (cs *claudeSession) Close() error {
|
||||
// Phase 1: Close stdin to signal EOF. Claude Code exits cleanly on
|
||||
// stdin close, running Stop hooks (e.g. claude-mem session summary).
|
||||
cs.stdinMu.Lock()
|
||||
_ = cs.stdin.Close()
|
||||
if err := cs.stdin.Close(); err != nil {
|
||||
slog.Warn("claudeSession: close stdin", "error", err)
|
||||
}
|
||||
cs.stdinMu.Unlock()
|
||||
|
||||
graceful := cs.gracefulStopTimeout
|
||||
@@ -726,7 +728,9 @@ func (cs *claudeSession) Close() error {
|
||||
// Phase 2: SIGTERM the whole process group — gives the process and its
|
||||
// descendants (e.g. MCP server bridges) a second chance to run cleanup
|
||||
// handlers that respond to signals but not stdin EOF.
|
||||
_ = signalProcessGroup(cs.cmd, syscall.SIGTERM)
|
||||
if err := signalProcessGroup(cs.cmd, syscall.SIGTERM); err != nil {
|
||||
slog.Warn("claudeSession: signal SIGTERM", "error", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-cs.done:
|
||||
@@ -741,7 +745,9 @@ func (cs *claudeSession) Close() error {
|
||||
// such as the Telegram bridge) are reaped along with the direct child;
|
||||
// otherwise they can survive as orphans and spin at 100% CPU.
|
||||
cs.cancel()
|
||||
_ = forceKillCmd(cs.cmd)
|
||||
if err := forceKillCmd(cs.cmd); err != nil {
|
||||
slog.Warn("claudeSession: force kill", "error", err)
|
||||
}
|
||||
<-cs.done
|
||||
return nil
|
||||
}
|
||||
|
||||
+1
-1
@@ -157,7 +157,7 @@ func (dh *DirHistory) saveLocked() {
|
||||
return
|
||||
}
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(dh.storePath), 0755); err != nil {
|
||||
if err := os.MkdirAll(filepath.Dir(dh.storePath), 0700); err != nil {
|
||||
slog.Error("dir_history: failed to create dir", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
+9
-3
@@ -6134,7 +6134,9 @@ func (e *Engine) cmdDiff(p Platform, msg *Message, raw string) {
|
||||
htmlData, err := e.diff2html(ctx, diffOutput, workDir, title)
|
||||
if err == nil {
|
||||
fileName := fmt.Sprintf("%s-%s.html", currentBranch, commitID)
|
||||
_ = e.waitOutgoing(p)
|
||||
if err := e.waitOutgoing(p); err != nil {
|
||||
slog.Warn("outgoing rate limit", "platform", p.Name(), "error", err)
|
||||
}
|
||||
if err := fileSender.SendFile(e.ctx, msg.ReplyCtx, FileAttachment{
|
||||
MimeType: "text/html", Data: htmlData, FileName: fileName,
|
||||
}); err == nil {
|
||||
@@ -9737,9 +9739,13 @@ func (e *Engine) executeCardAction(cmd, args, sessionKey string) {
|
||||
sub, id := subArgs[0], subArgs[1]
|
||||
switch sub {
|
||||
case "enable":
|
||||
_ = e.cronScheduler.EnableJob(id)
|
||||
if err := e.cronScheduler.EnableJob(id); err != nil {
|
||||
slog.Error("cron: enable job failed", "id", id, "error", err)
|
||||
}
|
||||
case "disable":
|
||||
_ = e.cronScheduler.DisableJob(id)
|
||||
if err := e.cronScheduler.DisableJob(id); err != nil {
|
||||
slog.Error("cron: disable job failed", "id", id, "error", err)
|
||||
}
|
||||
case "delete":
|
||||
e.cronScheduler.RemoveJob(id)
|
||||
case "mute":
|
||||
|
||||
+2
-1
@@ -120,8 +120,9 @@ func SaveFilesToDisk(workDir string, files []FileAttachment) []string {
|
||||
// Returns "" when the input cannot produce a safe basename, so callers can
|
||||
// fall back to a generated name.
|
||||
func sanitizeAttachmentFileName(name string) string {
|
||||
// Normalize backslashes to forward slashes so filepath.Base on Linux
|
||||
// Normalize backslashes to forward slashes so filepath.Base on any OS
|
||||
// strips Windows-style separators in attacker-supplied paths too.
|
||||
name = filepath.ToSlash(name)
|
||||
name = strings.ReplaceAll(name, "\\", "/")
|
||||
name = filepath.Base(name)
|
||||
if name == "" || name == "." || name == ".." {
|
||||
|
||||
+55
-17
@@ -112,6 +112,7 @@ type replyContext struct {
|
||||
}
|
||||
|
||||
type Platform struct {
|
||||
mu sync.RWMutex
|
||||
platformName string
|
||||
domain string
|
||||
appID string
|
||||
@@ -305,12 +306,38 @@ func (p *Platform) dispatchPlatform() core.Platform {
|
||||
return p
|
||||
}
|
||||
|
||||
func (p *Platform) getHandler() core.MessageHandler {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
return p.handler
|
||||
}
|
||||
|
||||
func (p *Platform) getCancel() context.CancelFunc {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
return p.cancel
|
||||
}
|
||||
|
||||
func (p *Platform) getServer() *http.Server {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
return p.server
|
||||
}
|
||||
|
||||
func (p *Platform) getBotOpenID() string {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
return p.botOpenID
|
||||
}
|
||||
|
||||
func (p *Platform) KeepPreviewOnFinish() bool {
|
||||
return p.useInteractiveCard
|
||||
}
|
||||
|
||||
func (p *Platform) Start(handler core.MessageHandler) error {
|
||||
p.mu.Lock()
|
||||
p.handler = handler
|
||||
p.mu.Unlock()
|
||||
|
||||
// In webhook mode (private/self-hosted Feishu/Lark), startup must not depend
|
||||
// on a successful bot-info API call. Older private deployments may not support
|
||||
@@ -321,7 +348,9 @@ func (p *Platform) Start(handler core.MessageHandler) error {
|
||||
if openID, err := p.fetchBotOpenID(); err != nil {
|
||||
slog.Warn(p.platformName+": failed to get bot open_id, group chat filtering disabled", "error", err)
|
||||
} else {
|
||||
p.mu.Lock()
|
||||
p.botOpenID = openID
|
||||
p.mu.Unlock()
|
||||
slog.Info(p.platformName+": bot identified", "open_id", openID)
|
||||
}
|
||||
}
|
||||
@@ -426,7 +455,9 @@ func (p *Platform) startWebSocketMode() error {
|
||||
p.wsClient = larkws.NewClient(p.appID, p.appSecret, wsOpts...)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
p.mu.Lock()
|
||||
p.cancel = cancel
|
||||
p.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
if err := p.wsClient.Start(ctx); err != nil {
|
||||
@@ -442,6 +473,7 @@ func (p *Platform) startWebhookMode() error {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc(p.callbackPath, p.webhookHandler)
|
||||
|
||||
p.mu.Lock()
|
||||
p.server = &http.Server{
|
||||
Addr: ":" + p.port,
|
||||
Handler: mux,
|
||||
@@ -449,6 +481,7 @@ func (p *Platform) startWebhookMode() error {
|
||||
|
||||
_, cancel := context.WithCancel(context.Background())
|
||||
p.cancel = cancel
|
||||
p.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
slog.Info(p.tag()+": webhook server listening", "port", p.port, "path", p.callbackPath)
|
||||
@@ -619,7 +652,8 @@ func (p *Platform) onCardAction(event *callback.CardActionTriggerEvent) (*callba
|
||||
}
|
||||
|
||||
rctx := replyContext{messageID: messageID, chatID: chatID, sessionKey: sessionKey}
|
||||
go p.handler(p.dispatchPlatform(), &core.Message{
|
||||
h := p.getHandler()
|
||||
go h(p.dispatchPlatform(), &core.Message{
|
||||
SessionKey: sessionKey,
|
||||
Platform: p.platformName,
|
||||
UserID: userID,
|
||||
@@ -650,7 +684,8 @@ func (p *Platform) onCardAction(event *callback.CardActionTriggerEvent) (*callba
|
||||
// askq: — AskUserQuestion option selected, forward as user message
|
||||
if strings.HasPrefix(actionVal, "askq:") {
|
||||
rctx := replyContext{messageID: messageID, chatID: chatID, sessionKey: sessionKey}
|
||||
go p.handler(p.dispatchPlatform(), &core.Message{
|
||||
h := p.getHandler()
|
||||
go h(p.dispatchPlatform(), &core.Message{
|
||||
SessionKey: sessionKey,
|
||||
Platform: p.platformName,
|
||||
UserID: userID,
|
||||
@@ -685,7 +720,8 @@ func (p *Platform) onCardAction(event *callback.CardActionTriggerEvent) (*callba
|
||||
|
||||
slog.Info(p.tag()+": card action dispatched as command", "cmd", cmdText, "user", userID)
|
||||
|
||||
go p.handler(p.dispatchPlatform(), &core.Message{
|
||||
h := p.getHandler()
|
||||
go h(p.dispatchPlatform(), &core.Message{
|
||||
SessionKey: sessionKey,
|
||||
Platform: p.platformName,
|
||||
UserID: userID,
|
||||
@@ -895,14 +931,15 @@ func isMessageWithdrawnError(err error) bool {
|
||||
}
|
||||
|
||||
func (p *Platform) dispatchCoreMessage(msg *core.Message) {
|
||||
if msg == nil || p.handler == nil {
|
||||
h := p.getHandler()
|
||||
if msg == nil || h == nil {
|
||||
return
|
||||
}
|
||||
if p.isMessageRecalled(msg.MessageID) {
|
||||
slog.Debug(p.tag()+": recalled message dispatch dropped", "message_id", msg.MessageID)
|
||||
return
|
||||
}
|
||||
p.handler(p.dispatchPlatform(), msg)
|
||||
h(p.dispatchPlatform(), msg)
|
||||
}
|
||||
|
||||
func (p *Platform) onMessageRecalled(_ context.Context, event *larkim.P2MessageRecalledV1) error {
|
||||
@@ -928,10 +965,11 @@ func (p *Platform) onMessageRecalled(_ context.Context, event *larkim.P2MessageR
|
||||
"recall_type", stringValue(event.Event.RecallType),
|
||||
)
|
||||
|
||||
if p.handler == nil {
|
||||
h := p.getHandler()
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
p.handler(p.dispatchPlatform(), &core.Message{
|
||||
h(p.dispatchPlatform(), &core.Message{
|
||||
Platform: p.platformName,
|
||||
MessageID: messageID,
|
||||
Recalled: true,
|
||||
@@ -999,8 +1037,8 @@ func (p *Platform) onMessage(ctx context.Context, event *larkim.P2MessageReceive
|
||||
"thread_isolation", p.threadIsolation,
|
||||
)
|
||||
|
||||
if chatType == "group" && !p.groupReplyAll && p.botOpenID != "" {
|
||||
if !isBotMentioned(msg.Mentions, p.botOpenID) {
|
||||
if chatType == "group" && !p.groupReplyAll && p.getBotOpenID() != "" {
|
||||
if !isBotMentioned(msg.Mentions, p.getBotOpenID()) {
|
||||
// Feishu @all sends {"text":"@_all"} with 0 mentions.
|
||||
if p.respondToAtEveryoneAndHere && msg.Content != nil && strings.Contains(*msg.Content, "@_all") {
|
||||
slog.Debug(p.tag()+": responding to @all message", "chat_id", chatID)
|
||||
@@ -1090,7 +1128,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
|
||||
slog.Error(p.tag()+": failed to parse text content", "error", err)
|
||||
return
|
||||
}
|
||||
text := stripMentions(textBody.Text, mentions, p.botOpenID)
|
||||
text := stripMentions(textBody.Text, mentions, p.getBotOpenID())
|
||||
if text == "" {
|
||||
slog.Debug(p.tag()+": dropping empty text after mention stripping",
|
||||
"message_id", messageID,
|
||||
@@ -1163,7 +1201,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
|
||||
|
||||
case "post":
|
||||
textParts, images := p.parsePostContent(messageID, content)
|
||||
text := stripMentions(strings.Join(textParts, "\n"), mentions, p.botOpenID)
|
||||
text := stripMentions(strings.Join(textParts, "\n"), mentions, p.getBotOpenID())
|
||||
if text == "" && len(images) == 0 {
|
||||
return
|
||||
}
|
||||
@@ -3644,17 +3682,17 @@ func (p *Platform) Stop() error {
|
||||
slog.Warn(p.tag()+": primary shutting down, secondary platforms will lose event source",
|
||||
"remaining", remaining)
|
||||
}
|
||||
if p.cancel != nil {
|
||||
p.cancel()
|
||||
if cancel := p.getCancel(); cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
} else {
|
||||
unregisterSharedWS(p)
|
||||
}
|
||||
// Stop webhook server if running (Lark international version)
|
||||
if p.server != nil {
|
||||
if svr := p.getServer(); svr != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := p.server.Shutdown(ctx); err != nil {
|
||||
if err := svr.Shutdown(ctx); err != nil {
|
||||
slog.Error(p.tag()+": webhook server shutdown error", "error", err)
|
||||
}
|
||||
}
|
||||
@@ -3809,7 +3847,7 @@ func (p *Platform) extractPostParts(messageID string, post *postLang) ([]string,
|
||||
textParts = append(textParts, elem.Text)
|
||||
}
|
||||
case "at":
|
||||
if p.botOpenID != "" && elem.UserId == p.botOpenID {
|
||||
if p.getBotOpenID() != "" && elem.UserId == p.getBotOpenID() {
|
||||
continue
|
||||
}
|
||||
switch {
|
||||
@@ -3873,7 +3911,7 @@ func (p *Platform) onBotMenu(event *larkapplication.P2BotMenuV6) error {
|
||||
userName := p.resolveUserName(userID)
|
||||
sessionKey := p.platformName + ":" + userID + ":" + userID
|
||||
|
||||
p.handler(p.dispatchPlatform(), &core.Message{
|
||||
p.getHandler()(p.dispatchPlatform(), &core.Message{
|
||||
SessionKey: sessionKey,
|
||||
Platform: p.platformName,
|
||||
Content: content,
|
||||
|
||||
@@ -1275,12 +1275,18 @@ func (p *Platform) ReconstructReplyCtx(sessionKey string) (any, error) {
|
||||
case 3:
|
||||
if p.shareSessionInChannel {
|
||||
// telegram:{chatID}:{threadID}
|
||||
threadID, _ = strconv.Atoi(parts[2])
|
||||
threadID, err = strconv.Atoi(parts[2])
|
||||
if err != nil {
|
||||
slog.Warn("telegram: invalid thread ID", "raw", parts[2], "error", err)
|
||||
}
|
||||
}
|
||||
// else: telegram:{chatID}:{userID} — no threadID
|
||||
case 4:
|
||||
// telegram:{chatID}:{threadID}:{userID}
|
||||
threadID, _ = strconv.Atoi(parts[2])
|
||||
threadID, err = strconv.Atoi(parts[2])
|
||||
if err != nil {
|
||||
slog.Warn("telegram: invalid thread ID", "raw", parts[2], "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
return replyContext{chatID: chatID, threadID: threadID}, nil
|
||||
|
||||
Reference in new issue
Block a user