okooooooo/pika
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 支持 Telegram 代理和自定义反代

Browse Source
This commit is contained in:
dushixiang committed 2026-08-24 00:14:07 +08:00
1 parent 1634225e31
commit 29582f9124
4 files changed
+290 -27
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

go.mod
+1 -1
View File
@@ -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
internal/service/notifier.go
+82 -6
View File
@@ -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 根据配置发送邮件通知
internal/service/notifier_telegram_test.go
+125
View File
@@ -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)
}
}
web/src/pages/Settings/NotificationChannels.tsx
+82 -20
View File
@@ -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 (
<div className={fieldGridClass}>
<Form.Item
label="Bot Token"
name="telegramBotToken"
rules={[
{required: true, message: '请输入 Bot Token'},
{validator: validateToken}
]}
tooltip="通过 @BotFather 创建机器人后获得的 token"
>
<Input.Password placeholder="输入 Bot Token"/>
</Form.Item>
<Form.Item
label="Chat ID"
name="telegramChatID"
rules={[{required: true, message: '请输入 Chat ID'}]}
tooltip="可以是用户 ID、群组 ID 或频道 ID，通过 @userinfobot 等机器人获取"
>
<Input placeholder="输入 Chat ID，例如：123456789"/>
</Form.Item>
<div className="space-y-0">
<div className={fieldGridClass}>
<Form.Item
label="Bot Token"
name="telegramBotToken"
rules={[
{required: true, message: '请输入 Bot Token'},
{validator: validateToken}
]}
tooltip="通过 @BotFather 创建机器人后获得的 token"
>
<Input.Password placeholder="输入 Bot Token"/>
</Form.Item>
<Form.Item
label="Chat ID"
name="telegramChatID"
rules={[{required: true, message: '请输入 Chat ID'}]}
tooltip="可以是用户 ID、群组 ID 或频道 ID，通过 @userinfobot 等机器人获取"
>
<Input placeholder="输入 Chat ID，例如：123456789"/>
</Form.Item>
<Form.Item
label="代理地址（HTTP/SOCKS5，可选）"
name="telegramProxyURL"
rules={[{validator: validateTelegramProxyURL}]}
tooltip="支持 HTTP 和 SOCKS5 代理；如需认证，请将账号密码写在代理 URL 中"
>
<Input.Password
autoComplete="new-password"
placeholder="例如：http://user:pass@127.0.0.1:7890"
/>
</Form.Item>
<Form.Item
label="自定义反代地址（可选）"
name="telegramAPIBaseURL"
rules={[{validator: validateTelegramAPIBaseURL}]}
tooltip="Telegram Bot API 的反向代理基址；请求路径会自动拼接 /bot{token}/sendMessage"
>
<Input placeholder="例如：https://telegram.example.com"/>
</Form.Item>
</div>
</div>
);
case 'email':