From 29582f9124380d44a6e9af6069d966038b938d75 Mon Sep 17 00:00:00 2001 From: dushixiang Date: Mon, 24 Aug 2026 00:14:07 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=20Telegram=20?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E5=92=8C=E8=87=AA=E5=AE=9A=E4=B9=89=E5=8F=8D?= =?UTF-8?q?=E4=BB=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go.mod | 2 +- internal/service/notifier.go | 88 +++++++++++- internal/service/notifier_telegram_test.go | 125 ++++++++++++++++++ .../pages/Settings/NotificationChannels.tsx | 102 +++++++++++--- 4 files changed, 290 insertions(+), 27 deletions(-) create mode 100644 internal/service/notifier_telegram_test.go diff --git a/go.mod b/go.mod index 54432a3..35d1deb 100644 --- a/go.mod +++ b/go.mod @@ -37,6 +37,7 @@ require ( github.com/valyala/fasttemplate v1.2.2 go.uber.org/zap v1.28.0 golang.org/x/crypto v0.54.0 + golang.org/x/net v0.57.0 golang.org/x/oauth2 v0.36.0 golang.org/x/text v0.40.0 gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df @@ -85,7 +86,6 @@ require ( github.com/yusufpapurcu/wmi v1.2.4 // indirect go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect - golang.org/x/net v0.57.0 // indirect golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/time v0.15.0 // indirect diff --git a/internal/service/notifier.go b/internal/service/notifier.go index 6520f4c..bec2cf5 100644 --- a/internal/service/notifier.go +++ b/internal/service/notifier.go @@ -10,7 +10,9 @@ import ( "errors" "fmt" "io" + "net" "net/http" + "net/url" "strconv" "strings" "sync" @@ -22,6 +24,7 @@ import ( "github.com/pika-monitor/pika/internal/utils" "github.com/valyala/fasttemplate" "go.uber.org/zap" + "golang.org/x/net/proxy" "gopkg.in/gomail.v2" ) @@ -540,8 +543,62 @@ func (n *Notifier) sendFeishu(ctx context.Context, webhook, signSecret, message return nil } +func buildTelegramWebhookURL(botToken, apiBaseURL string) (string, error) { + if apiBaseURL == "" { + apiBaseURL = "https://api.telegram.org" + } + + parsedURL, err := url.Parse(apiBaseURL) + if err != nil || parsedURL.Host == "" || (parsedURL.Scheme != "http" && parsedURL.Scheme != "https") { + return "", fmt.Errorf("Telegram 自定义反代地址必须是有效的 HTTP/HTTPS 地址") + } + if parsedURL.RawQuery != "" || parsedURL.Fragment != "" { + return "", fmt.Errorf("Telegram 自定义反代地址不能包含查询参数或片段") + } + + parsedURL.Path = strings.TrimRight(parsedURL.Path, "/") + "/bot" + botToken + "/sendMessage" + return parsedURL.String(), nil +} + +func newTelegramHTTPClient(proxyURL string) (*http.Client, error) { + if proxyURL == "" { + return sharedHTTPClient, nil + } + + parsedURL, err := url.Parse(proxyURL) + if err != nil || parsedURL.Host == "" { + return nil, fmt.Errorf("Telegram 代理地址无效") + } + + transport := &http.Transport{ + MaxIdleConns: 10, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + } + + switch parsedURL.Scheme { + case "http": + transport.Proxy = http.ProxyURL(parsedURL) + case "socks5": + dialer, err := proxy.FromURL(parsedURL, proxy.Direct) + if err != nil { + return nil, fmt.Errorf("创建 Telegram SOCKS5 代理失败: %w", err) + } + transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + if contextDialer, ok := dialer.(proxy.ContextDialer); ok { + return contextDialer.DialContext(ctx, network, address) + } + return dialer.Dial(network, address) + } + default: + return nil, fmt.Errorf("Telegram 代理仅支持 http:// 或 socks5://") + } + + return &http.Client{Timeout: 10 * time.Second, Transport: transport}, nil +} + // sendTelegram 发送 Telegram 通知 -func (n *Notifier) sendTelegram(ctx context.Context, botToken, chatID, message string) error { +func (n *Notifier) sendTelegram(ctx context.Context, botToken, chatID, message, proxyURL, apiBaseURL string) error { if utf8.RuneCountInString(message) > 4096 { n.logger.Warn("Telegram消息过长，进行截断", zap.Int("originalLen", utf8.RuneCountInString(message)), @@ -551,14 +608,22 @@ func (n *Notifier) sendTelegram(ctx context.Context, botToken, chatID, message s message = string(runes[:4093]) + "..." } - webhookURL := fmt.Sprintf("https://api.telegram.org/bot%s/sendMessage", botToken) + webhookURL, err := buildTelegramWebhookURL(botToken, apiBaseURL) + if err != nil { + return err + } + + httpClient, err := newTelegramHTTPClient(proxyURL) + if err != nil { + return err + } body := map[string]interface{}{ "chat_id": chatID, "text": message, } - _, err := n.sendJSONRequest(ctx, webhookURL, body) + _, err = n.sendJSONRequestWithClient(ctx, httpClient, webhookURL, body) if err != nil { return err } @@ -719,6 +784,10 @@ func (n *Notifier) buildCustomBody(agent *models.Agent, record *models.AlertReco // retryDoWithBackoff 带重试和退避的 HTTP 请求执行 func (n *Notifier) retryDoWithBackoff(ctx context.Context, createReq func() (*http.Request, error)) (*http.Response, error) { + return n.retryDoWithClientBackoff(ctx, sharedHTTPClient, createReq) +} + +func (n *Notifier) retryDoWithClientBackoff(ctx context.Context, httpClient *http.Client, createReq func() (*http.Request, error)) (*http.Response, error) { var lastErr error for i := 0; i < maxRetries; i++ { if i > 0 { @@ -739,7 +808,7 @@ func (n *Notifier) retryDoWithBackoff(ctx context.Context, createReq func() (*ht return nil, fmt.Errorf("创建请求失败: %w", err) } - resp, err := sharedHTTPClient.Do(req) + resp, err := httpClient.Do(req) if err == nil { return resp, nil } @@ -815,6 +884,10 @@ func (n *Notifier) sendCustomWebhook(ctx context.Context, config map[string]inte } func (n *Notifier) sendJSONRequest(ctx context.Context, url string, body interface{}) ([]byte, error) { + return n.sendJSONRequestWithClient(ctx, sharedHTTPClient, url, body) +} + +func (n *Notifier) sendJSONRequestWithClient(ctx context.Context, httpClient *http.Client, url string, body interface{}) ([]byte, error) { data, err := json.Marshal(body) if err != nil { return nil, fmt.Errorf("序列化请求体失败: %w", err) @@ -829,7 +902,7 @@ func (n *Notifier) sendJSONRequest(ctx context.Context, url string, body interfa return req, nil } - resp, err := n.retryDoWithBackoff(ctx, createReq) + resp, err := n.retryDoWithClientBackoff(ctx, httpClient, createReq) if err != nil { return nil, fmt.Errorf("发送请求失败: %w", err) } @@ -932,7 +1005,10 @@ func (n *Notifier) sendTelegramByConfig(ctx context.Context, config map[string]i return fmt.Errorf("Telegram 配置缺少 chatID") } - return n.sendTelegram(ctx, botToken, chatID, message) + proxyURL, _ := config["proxyURL"].(string) + apiBaseURL, _ := config["apiBaseURL"].(string) + + return n.sendTelegram(ctx, botToken, chatID, message, proxyURL, apiBaseURL) } // sendEmailByConfig 根据配置发送邮件通知 diff --git a/internal/service/notifier_telegram_test.go b/internal/service/notifier_telegram_test.go new file mode 100644 index 0000000..ed6090c --- /dev/null +++ b/internal/service/notifier_telegram_test.go @@ -0,0 +1,125 @@ +package service + +import ( + "context" + "encoding/base64" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "testing" + + "go.uber.org/zap" +) + +func TestBuildTelegramWebhookURL(t *testing.T) { + tests := []struct { + name string + apiBaseURL string + want string + wantErr bool + }{ + {name: "default", want: "https://api.telegram.org/bot123:abc/sendMessage"}, + {name: "custom host", apiBaseURL: "https://telegram.example.com/", want: "https://telegram.example.com/bot123:abc/sendMessage"}, + {name: "custom path", apiBaseURL: "http://telegram.example.com/api", want: "http://telegram.example.com/api/bot123:abc/sendMessage"}, + {name: "unsupported scheme", apiBaseURL: "ftp://telegram.example.com", wantErr: true}, + {name: "query is rejected", apiBaseURL: "https://telegram.example.com/?key=value", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := buildTelegramWebhookURL("123:abc", tt.apiBaseURL) + if tt.wantErr { + if err == nil { + t.Fatal("expected an error") + } + return + } + if err != nil { + t.Fatalf("buildTelegramWebhookURL() error = %v", err) + } + if got != tt.want { + t.Fatalf("buildTelegramWebhookURL() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestNewTelegramHTTPClient(t *testing.T) { + for _, proxyURL := range []string{ + "http://user:pass@127.0.0.1:7890", + "socks5://user:pass@127.0.0.1:1080", + } { + t.Run(proxyURL, func(t *testing.T) { + client, err := newTelegramHTTPClient(proxyURL) + if err != nil { + t.Fatalf("newTelegramHTTPClient() error = %v", err) + } + if client == sharedHTTPClient { + t.Fatal("proxied Telegram client unexpectedly uses the shared client") + } + }) + } + + if _, err := newTelegramHTTPClient("https://127.0.0.1:7890"); err == nil { + t.Fatal("expected HTTPS proxy scheme to be rejected") + } +} + +func TestSendTelegramWithCustomAPIBaseURL(t *testing.T) { + var gotBody map[string]interface{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/telegram/bot123:abc/sendMessage" { + t.Errorf("request path = %q", r.URL.Path) + } + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + t.Errorf("decode request body: %v", err) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer server.Close() + + notifier := NewNotifier(zap.NewNop()) + err := notifier.sendTelegram(context.Background(), "123:abc", "456", "hello", "", server.URL+"/telegram/") + if err != nil { + t.Fatalf("sendTelegram() error = %v", err) + } + if gotBody["chat_id"] != "456" || gotBody["text"] != "hello" { + t.Fatalf("request body = %#v", gotBody) + } +} + +func TestSendTelegramWithHTTPProxy(t *testing.T) { + proxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Scheme != "http" || r.URL.Host != "telegram.invalid" { + t.Errorf("proxy request URL = %q", r.URL.String()) + } + wantProxyAuthorization := "Basic " + base64.StdEncoding.EncodeToString([]byte("proxy-user:proxy-password")) + if got := r.Header.Get("Proxy-Authorization"); got != wantProxyAuthorization { + t.Errorf("Proxy-Authorization = %q, want %q", got, wantProxyAuthorization) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer proxyServer.Close() + + parsedProxyURL, err := url.Parse(proxyServer.URL) + if err != nil { + t.Fatal(err) + } + parsedProxyURL.User = url.UserPassword("proxy-user", "proxy-password") + + notifier := NewNotifier(zap.NewNop()) + err = notifier.sendTelegram( + context.Background(), + "123:abc", + "456", + "hello", + parsedProxyURL.String(), + "http://telegram.invalid", + ) + if err != nil { + t.Fatalf("sendTelegram() error = %v", err) + } +} diff --git a/web/src/pages/Settings/NotificationChannels.tsx b/web/src/pages/Settings/NotificationChannels.tsx index e2dcca9..f6bc021 100644 --- a/web/src/pages/Settings/NotificationChannels.tsx +++ b/web/src/pages/Settings/NotificationChannels.tsx @@ -102,6 +102,43 @@ const NotificationChannels = () => { return Promise.resolve(); }; + const validateTelegramProxyURL = (_: any, value?: string) => { + if (!value) { + return Promise.resolve(); + } + + try { + const proxyURL = new URL(value); + if (!['http:', 'socks5:'].includes(proxyURL.protocol) || !proxyURL.hostname) { + throw new Error(); + } + return Promise.resolve(); + } catch { + return Promise.reject(new Error('请输入 http:// 或 socks5:// 开头的有效代理地址')); + } + }; + + const validateTelegramAPIBaseURL = (_: any, value?: string) => { + if (!value) { + return Promise.resolve(); + } + + try { + const apiBaseURL = new URL(value); + if ( + !['http:', 'https:'].includes(apiBaseURL.protocol) + || !apiBaseURL.hostname + || apiBaseURL.search + || apiBaseURL.hash + ) { + throw new Error(); + } + return Promise.resolve(); + } catch { + return Promise.reject(new Error('请输入 http:// 或 https:// 开头的有效反代地址')); + } + }; + // 监听各个通知渠道的启用状态 const dingtalkEnabled = Form.useWatch('dingtalkEnabled', form); const wecomEnabled = Form.useWatch('wecomEnabled', form); @@ -179,6 +216,8 @@ const NotificationChannels = () => { formValues.telegramEnabled = channel.enabled; formValues.telegramBotToken = channel.config?.botToken || ''; formValues.telegramChatID = channel.config?.chatID || ''; + formValues.telegramProxyURL = channel.config?.proxyURL || ''; + formValues.telegramAPIBaseURL = channel.config?.apiBaseURL || ''; } else if (channel.type === 'email') { formValues.emailEnabled = channel.enabled; formValues.emailSmtpHost = channel.config?.smtpHost || ''; @@ -268,6 +307,8 @@ const NotificationChannels = () => { config: { botToken: values.telegramBotToken || '', chatID: values.telegramChatID || '', + proxyURL: values.telegramProxyURL || '', + apiBaseURL: values.telegramAPIBaseURL || '', }, }); } @@ -457,26 +498,47 @@ const NotificationChannels = () => { ); case 'telegram': return ( -
- - - - - - +
+
+ + + + + + + + + + + + +
); case 'email':