diff --git a/internal/apps/admin/push/push_test.go b/internal/apps/admin/push/push_test.go index 83f3688e..9174151f 100644 --- a/internal/apps/admin/push/push_test.go +++ b/internal/apps/admin/push/push_test.go @@ -51,12 +51,12 @@ type mockPusher struct { sentTgt string } -func (m *mockPusher) Send(ctx context.Context, cfg pkgpush.Config, target string, body map[string]any, template string, ext map[string]any) error { +func (m *mockPusher) Send(ctx context.Context, cfg pkgpush.Config, target string, body map[string]any, template string, ext map[string]any) (string, error) { m.mu.Lock() defer m.mu.Unlock() m.sentBody = body m.sentTgt = target - return nil + return "", nil } func (m *mockPusher) ValidateConfig(cfg pkgpush.Config) error { diff --git a/internal/apps/admin/push/routers.go b/internal/apps/admin/push/routers.go index d7e0df0f..7f937c13 100644 --- a/internal/apps/admin/push/routers.go +++ b/internal/apps/admin/push/routers.go @@ -284,7 +284,7 @@ func TestPush(c *gin.Context) { keyContent: "当您收到这条消息,说明当前渠道连通性测试通过。", keyLevel: defaultLevelInfo, } - if err := pusher.Send(c.Request.Context(), req.Config, req.Target, testBody, "", nil); err != nil { + if _, err := pusher.Send(c.Request.Context(), req.Config, req.Target, testBody, "", nil); err != nil { response.AbortBadRequest(c, err.Error()) return } diff --git a/internal/apps/admin/push/tasks.go b/internal/apps/admin/push/tasks.go index 2d4ad7ca..de310e87 100644 --- a/internal/apps/admin/push/tasks.go +++ b/internal/apps/admin/push/tasks.go @@ -93,13 +93,16 @@ func (h *PushHandler) Execute(ctx context.Context, payload []byte) (*task.TaskRe // 执行真正的消息推送,扁平化为原始 json 格式 flatBody := req.Body.Flatten() - err = pusher.Send(ctx, req.Config, req.Target, flatBody, req.Template, nil) + upstreamResp, err := pusher.Send(ctx, req.Config, req.Target, flatBody, req.Template, nil) title := req.Body.Title content := req.Body.Content if err != nil { task.AppendLog(ctx, "消息推送失败 (标题: %s): %v", title, err) + if upstreamResp != "" { + task.AppendLog(ctx, "上游返回: %s", upstreamResp) + } if task.IsFinalAttempt(ctx) { h.recordHistory(ctx, req, "failed", err.Error()) } @@ -107,6 +110,9 @@ func (h *PushHandler) Execute(ctx context.Context, payload []byte) (*task.TaskRe } task.AppendLog(ctx, "消息推送成功 (标题: %s, 内容摘要: %s)", title, content) + if upstreamResp != "" { + task.AppendLog(ctx, "上游返回: %s", upstreamResp) + } h.recordHistory(ctx, req, "success", "") return &task.TaskResult{ diff --git a/pkg/push/custom.go b/pkg/push/custom.go index 6957821d..c8326792 100644 --- a/pkg/push/custom.go +++ b/pkg/push/custom.go @@ -9,6 +9,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "net/http" "strings" @@ -19,13 +20,16 @@ func init() { Register("custom", &CustomPusher{}) } +// maxCustomResponseBytes 限制读取 Webhook 响应体的最大字节数,防止无界读取。 +const maxCustomResponseBytes = 4096 + // CustomPusher 自定义 Webhook 发送实现 type CustomPusher struct{} // Send 发送自定义 webhook -func (p *CustomPusher) Send(ctx context.Context, cfg Config, _ string, body map[string]any, template string, _ map[string]any) error { +func (p *CustomPusher) Send(ctx context.Context, cfg Config, _ string, body map[string]any, template string, _ map[string]any) (string, error) { if cfg.URL == "" { - return errors.New("custom: URL is required") + return "", errors.New("custom: URL is required") } var reqBody []byte @@ -39,13 +43,13 @@ func (p *CustomPusher) Send(ctx context.Context, cfg Config, _ string, body map[ var err error reqBody, err = json.Marshal(body) if err != nil { - return fmt.Errorf("custom: marshal body failed: %w", err) + return "", fmt.Errorf("custom: marshal body failed: %w", err) } } httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.URL, bytes.NewReader(reqBody)) if err != nil { - return fmt.Errorf("custom: create http request failed: %w", err) + return "", fmt.Errorf("custom: create http request failed: %w", err) } httpReq.Header.Set("Content-Type", "application/json") @@ -58,15 +62,28 @@ func (p *CustomPusher) Send(ctx context.Context, cfg Config, _ string, body map[ client := httppool.NewClient(defaultHTTPClientTimeout) resp, err := client.Do(httpReq) if err != nil { - return fmt.Errorf("custom: http request failed: %w", err) + return "", fmt.Errorf("custom: http request failed: %w", err) } defer func() { _ = resp.Body.Close() }() - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("custom: http status %s", resp.Status) + bodyBytes, _ := io.ReadAll(io.LimitReader(resp.Body, maxCustomResponseBytes)) + upstreamResp := strings.TrimSpace(string(bodyBytes)) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { //nolint:mnd + return upstreamResp, fmt.Errorf("custom: http status %s", resp.Status) } - return nil + // 部分 Webhook(如企业微信、钉钉)即使业务失败也返回 HTTP 200, + // 仅当响应体包含非零 errcode 时才判定为发送失败,避免审计记录误报成功。 + var apiResp struct { + ErrCode int `json:"errcode"` + ErrMsg string `json:"errmsg"` + } + if err := json.Unmarshal(bodyBytes, &apiResp); err == nil && apiResp.ErrCode != 0 { + return upstreamResp, fmt.Errorf("custom: webhook rejected: errcode=%d errmsg=%q", apiResp.ErrCode, apiResp.ErrMsg) + } + + return upstreamResp, nil } // ValidateConfig 校验自定义配置 diff --git a/pkg/push/custom_test.go b/pkg/push/custom_test.go new file mode 100644 index 00000000..fb622e65 --- /dev/null +++ b/pkg/push/custom_test.go @@ -0,0 +1,91 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package push + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCustomPusherSend_ResponseBodyErrcode(t *testing.T) { + tests := []struct { + name string + statusCode int + body string + wantErr bool + wantErrMsg string + }{ + { + name: "wechat business error returns HTTP 200 with non-zero errcode", + statusCode: http.StatusOK, + body: `{"errcode":93000,"errmsg":"invalid request data"}`, + wantErr: true, + wantErrMsg: "errcode=93000", + }, + { + name: "wechat success returns errcode 0", + statusCode: http.StatusOK, + body: `{"errcode":0,"errmsg":"ok"}`, + wantErr: false, + }, + { + name: "json response without errcode is tolerated", + statusCode: http.StatusOK, + body: `{"success":true}`, + wantErr: false, + }, + { + name: "non-json response body is tolerated", + statusCode: http.StatusOK, + body: "ok", + wantErr: false, + }, + { + name: "empty response body is tolerated", + statusCode: http.StatusNoContent, + body: "", + wantErr: false, + }, + { + name: "http error status still fails", + statusCode: http.StatusInternalServerError, + body: `{"errcode":0,"errmsg":"ok"}`, + wantErr: true, + wantErrMsg: "http status", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tt.statusCode) + _, _ = w.Write([]byte(tt.body)) + })) + defer srv.Close() + + pusher := &CustomPusher{} + upstreamResp, err := pusher.Send(context.Background(), + Config{Channel: "custom", URL: srv.URL}, + "", + map[string]any{"title": "t", "content": "c"}, + `{"title":"$title","content":"$content"}`, + nil, + ) + if tt.wantErr { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErrMsg) + return + } + assert.NoError(t, err) + if tt.body != "" { + assert.Contains(t, upstreamResp, tt.body) + } + }) + } +} diff --git a/pkg/push/email.go b/pkg/push/email.go index ba8d289b..c42dc80d 100644 --- a/pkg/push/email.go +++ b/pkg/push/email.go @@ -20,12 +20,12 @@ func init() { type EmailPusher struct{} // Send 发送邮件 -func (p *EmailPusher) Send(ctx context.Context, cfg Config, target string, body map[string]any, _ string, ext map[string]any) error { +func (p *EmailPusher) Send(ctx context.Context, cfg Config, target string, body map[string]any, _ string, ext map[string]any) (string, error) { if cfg.URL == "" || cfg.Key == "" || cfg.Secret == "" { - return errors.New("email: SMTP configuration (url, key, secret) is incomplete") + return "", errors.New("email: SMTP configuration (url, key, secret) is incomplete") } if target == "" { - return errors.New("email: target email address is required") + return "", errors.New("email: target email address is required") } title := defaultTitle @@ -84,14 +84,14 @@ func (p *EmailPusher) Send(ctx context.Context, cfg Config, target string, body select { case <-ctx.Done(): - return ctx.Err() + return "", ctx.Err() case err := <-errChan: if err != nil { - return fmt.Errorf("email: send smtp mail failed: %w", err) + return "", fmt.Errorf("email: send smtp mail failed: %w", err) } } - return nil + return "", nil } // ValidateConfig 校验邮件 SMTP 配置 diff --git a/pkg/push/lark.go b/pkg/push/lark.go index a29ea7aa..e97a8b1a 100644 --- a/pkg/push/lark.go +++ b/pkg/push/lark.go @@ -76,9 +76,9 @@ type larkMessageResponse struct { // Send 执行飞书消息发送 // //nolint:nestif,cyclop -func (p *LarkPusher) Send(ctx context.Context, cfg Config, _ string, body map[string]any, template string, _ map[string]any) error { +func (p *LarkPusher) Send(ctx context.Context, cfg Config, _ string, body map[string]any, template string, _ map[string]any) (string, error) { if cfg.URL == "" { - return errors.New("lark: URL is required") + return "", errors.New("lark: URL is required") } var req larkMessageRequest @@ -212,7 +212,7 @@ func (p *LarkPusher) Send(ctx context.Context, cfg Config, _ string, body map[st timestamp := time.Now().Unix() sign, err := larkSign(cfg.Secret, timestamp) if err != nil { - return fmt.Errorf("lark: sign failed: %w", err) + return "", fmt.Errorf("lark: sign failed: %w", err) } req.Timestamp = strconv.FormatInt(timestamp, 10) req.Sign = sign @@ -220,37 +220,37 @@ func (p *LarkPusher) Send(ctx context.Context, cfg Config, _ string, body map[st jsonData, err := json.Marshal(req) if err != nil { - return fmt.Errorf("lark: marshal request failed: %w", err) + return "", fmt.Errorf("lark: marshal request failed: %w", err) } // 4. 发送 POST 请求 httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.URL, bytes.NewBuffer(jsonData)) if err != nil { - return fmt.Errorf("lark: create http request failed: %w", err) + return "", fmt.Errorf("lark: create http request failed: %w", err) } httpReq.Header.Set("Content-Type", "application/json") client := httppool.NewClient(defaultHTTPClientTimeout) resp, err := client.Do(httpReq) if err != nil { - return fmt.Errorf("lark: http request failed: %w", err) + return "", fmt.Errorf("lark: http request failed: %w", err) } defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusOK { - return fmt.Errorf("lark: http status %s", resp.Status) + return "", fmt.Errorf("lark: http status %s", resp.Status) } var res larkMessageResponse if err := json.NewDecoder(resp.Body).Decode(&res); err != nil { - return fmt.Errorf("lark: decode response failed: %w", err) + return "", fmt.Errorf("lark: decode response failed: %w", err) } if res.Code != 0 { - return fmt.Errorf("lark: send message failed, code %d: %s", res.Code, res.Msg) + return "", fmt.Errorf("lark: send message failed, code %d: %s", res.Code, res.Msg) } - return nil + return "", nil } // ValidateConfig 校验飞书配置 diff --git a/pkg/push/push.go b/pkg/push/push.go index cacdbba7..e035d62a 100644 --- a/pkg/push/push.go +++ b/pkg/push/push.go @@ -33,7 +33,8 @@ type Pusher interface { // body: 消息体数据 (含默认字段如 title, content, level) // template: 消息卡片/模板 JSON (可选) // ext: 预留的单次发送拓展数据 - Send(ctx context.Context, cfg Config, target string, body map[string]any, template string, ext map[string]any) error + // 返回 upstreamResp: 上游服务返回的响应内容(如 Webhook 响应体),用于任务日志审计;无响应时为空字符串 + Send(ctx context.Context, cfg Config, target string, body map[string]any, template string, ext map[string]any) (upstreamResp string, err error) // ValidateConfig 校验渠道配置合法性 ValidateConfig(cfg Config) error diff --git a/pkg/push/telegram.go b/pkg/push/telegram.go index a3956c3f..54f49e76 100644 --- a/pkg/push/telegram.go +++ b/pkg/push/telegram.go @@ -37,9 +37,9 @@ type telegramErrorResponse struct { // Send 执行 Telegram 消息发送 // //nolint:cyclop -func (p *TelegramPusher) Send(ctx context.Context, cfg Config, target string, body map[string]any, template string, _ map[string]any) error { +func (p *TelegramPusher) Send(ctx context.Context, cfg Config, target string, body map[string]any, template string, _ map[string]any) (string, error) { if cfg.Secret == "" { - return errors.New("telegram: Bot Token (Secret) is required") + return "", errors.New("telegram: Bot Token (Secret) is required") } chatID := target @@ -47,7 +47,7 @@ func (p *TelegramPusher) Send(ctx context.Context, cfg Config, target string, bo chatID = cfg.Key // Use default chat ID (Key) if target is blank } if chatID == "" { - return errors.New("telegram: chat_id (target or default Key) is required") + return "", errors.New("telegram: chat_id (target or default Key) is required") } baseURL := cfg.URL @@ -92,11 +92,11 @@ func (p *TelegramPusher) Send(ctx context.Context, cfg Config, target string, bo } fallbackErr := p.sendMessage(ctx, baseURL, cfg.Secret, chatID, plainText, "") if fallbackErr != nil { - return fmt.Errorf("telegram: send message failed (fallback also failed): %w (original HTML error: %v)", fallbackErr, err) + return "", fmt.Errorf("telegram: send message failed (fallback also failed): %w (original HTML error: %v)", fallbackErr, err) } } - return nil + return "", nil } // ValidateConfig 校验 Telegram 配置 diff --git a/pkg/push/telegram_test.go b/pkg/push/telegram_test.go index f018cadd..74fcf728 100644 --- a/pkg/push/telegram_test.go +++ b/pkg/push/telegram_test.go @@ -41,7 +41,7 @@ func TestTelegramPusher_Send(t *testing.T) { "content": "Host down", "level": "CRITICAL", } - err := pusher.Send(context.Background(), cfg, "123456", body, "", nil) + _, err := pusher.Send(context.Background(), cfg, "123456", body, "", nil) require.NoError(t, err) assert.Equal(t, "123456", receivedReq.ChatID) @@ -79,7 +79,7 @@ func TestTelegramPusher_Send(t *testing.T) { "content": "A < B comparison", "level": "INFO", } - err := pusher.Send(context.Background(), cfg, "123456", body, "", nil) + _, err := pusher.Send(context.Background(), cfg, "123456", body, "", nil) require.NoError(t, err) require.Len(t, requests, 2)