feat: configurable traffic threshold and Feishu bot alerts

2026-10-23 17:49:19 +08:00
1 parent 3e6e481b79
commit 1273f226fd
@@ -0,0 +1,262 @@
+package alertnotify
+
+import (
+	"bytes"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"gorm.io/gorm"
+	"vpn-platform/backend/internal/model"
+	"vpn-platform/backend/internal/store"
+)
+
+const DefaultThreshold = 50
+
+type Input struct {
+	RateThresholdMbps int    `json:"rate_threshold_mbps"`
+	Enabled           bool   `json:"enabled"`
+	Webhook           string `json:"webhook"`
+	SigningSecret     string `json:"signing_secret"`
+}
+
+type Status struct {
+	RateThresholdMbps int  `json:"rate_threshold_mbps"`
+	Enabled           bool `json:"enabled"`
+	Configured        bool `json:"configured"`
+	Signed            bool `json:"signed"`
+}
+
+func current() (model.AlertSetting, error) {
+	var row model.AlertSetting
+	err := store.DB.First(&row, 1).Error
+	if err == nil {
+		return row, nil
+	}
+	// Missing singleton is normal on older installs; other DB errors must propagate.
+	if !errors.Is(err, gorm.ErrRecordNotFound) {
+		return row, err
+	}
+	return model.AlertSetting{ID: 1, RateThresholdMbps: DefaultThreshold}, nil
+}
+
+func Get() (Status, error) {
+	row, err := current()
+	return Status{RateThresholdMbps: row.RateThresholdMbps, Enabled: row.Enabled,
+		Configured: row.WebhookEncrypted != "", Signed: row.SigningEncrypted != ""}, err
+}
+
+func Save(in Input) (Status, error) {
+	if in.RateThresholdMbps < 1 || in.RateThresholdMbps > 100000 {
+		return Status{}, fmt.Errorf("阈值须为 1–100000 Mbps")
+	}
+	if len(in.Webhook) > 512 || len(in.SigningSecret) > 512 {
+		return Status{}, fmt.Errorf("机器人凭据过长")
+	}
+	row, err := current()
+	if err != nil {
+		return Status{}, err
+	}
+	if in.Webhook != "" {
+		if !validWebhook(in.Webhook) {
+			return Status{}, fmt.Errorf("仅支持飞书/Lark 官方群机器人 HTTPS Webhook")
+		}
+		row.WebhookEncrypted, err = store.EncryptSecret(in.Webhook)
+		if err != nil {
+			return Status{}, err
+		}
+	}
+	if in.SigningSecret != "" {
+		row.SigningEncrypted, err = store.EncryptSecret(in.SigningSecret)
+		if err != nil {
+			return Status{}, err
+		}
+	}
+	if in.Enabled && row.WebhookEncrypted == "" {
+		return Status{}, fmt.Errorf("启用通知前请填写群机器人 Webhook")
+	}
+	row.Enabled = in.Enabled
+	row.RateThresholdMbps = in.RateThresholdMbps
+	if err := store.DB.Save(&row).Error; err != nil {
+		return Status{}, err
+	}
+	return Get()
+}
+
+func Clear() (Status, error) {
+	row, err := current()
+	if err != nil {
+		return Status{}, err
+	}
+	row.Enabled = false
+	row.WebhookEncrypted = ""
+	row.SigningEncrypted = ""
+	if err := store.DB.Save(&row).Error; err != nil {
+		return Status{}, err
+	}
+	return Get()
+}
+
+func validWebhook(raw string) bool {
+	u, err := url.Parse(raw)
+	if err != nil || u.Scheme != "https" || u.User != nil || u.Port() != "" || u.RawQuery != "" || u.Fragment != "" {
+		return false
+	}
+	if u.Host != "open.feishu.cn" && u.Host != "open.larksuite.com" {
+		return false
+	}
+	id := strings.TrimPrefix(u.Path, "/open-apis/bot/v2/hook/")
+	if id == u.Path || id == "" || len(id) > 128 {
+		return false
+	}
+	for _, ch := range id {
+		if !(ch >= 'a' && ch <= 'z' || ch >= 'A' && ch <= 'Z' || ch >= '0' && ch <= '9' || ch == '-') {
+			return false
+		}
+	}
+	return true
+}
+
+var client = &http.Client{Timeout: 6 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
+
+func send(webhook, secret, content string) error {
+	if !validWebhook(webhook) {
+		return errors.New("invalid_webhook")
+	}
+	payload := map[string]interface{}{"msg_type": "text", "content": map[string]string{"text": content}}
+	if secret != "" {
+		ts := strconv.FormatInt(time.Now().Unix(), 10)
+		mac := hmac.New(sha256.New, []byte(ts+"\n"+secret))
+		payload["timestamp"] = ts
+		payload["sign"] = base64.StdEncoding.EncodeToString(mac.Sum(nil))
+	}
+	body, _ := json.Marshal(payload)
+	req, err := http.NewRequest(http.MethodPost, webhook, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := client.Do(req)
+	if err != nil {
+		return errors.New("connection_failed")
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 {
+		return errors.New("http_error")
+	}
+	var result struct {
+		Code       *int `json:"code"`
+		StatusCode *int `json:"StatusCode"`
+	}
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 4096)).Decode(&result); err != nil {
+		return errors.New("invalid_response")
+	}
+	if result.Code != nil && *result.Code == 0 || result.StatusCode != nil && *result.StatusCode == 0 {
+		return nil
+	}
+	return errors.New("provider_rejected")
+}
+
+func Test() error {
+	row, err := current()
+	if err != nil {
+		return err
+	}
+	if !row.Enabled || row.WebhookEncrypted == "" {
+		return errors.New("请先保存并启用飞书通知")
+	}
+	webhook, err := store.DecryptSecret(row.WebhookEncrypted)
+	if err != nil {
+		return err
+	}
+	secret := ""
+	if row.SigningEncrypted != "" {
+		secret, err = store.DecryptSecret(row.SigningEncrypted)
+		if err != nil {
+			return err
+		}
+	}
+	return send(webhook, secret, "OpsX VPN 通知测试：群机器人配置可用")
+}
+
+type event struct {
+	tunnelID     uint
+	kind, detail string
+}
+
+var queue = make(chan event, 64)
+var once sync.Once
+var mu sync.Mutex
+var pending = make(map[string]bool)
+var lastSent = make(map[string]time.Time)
+
+// Notify queues one delivery per tunnel and event. Network calls never block the monitor.
+func Notify(tunnelID uint, kind, detail string) {
+	row, err := current()
+	if err != nil || !row.Enabled || row.WebhookEncrypted == "" {
+		return
+	}
+	key := fmt.Sprintf("%d:%s", tunnelID, kind)
+	mu.Lock()
+	if pending[key] || time.Since(lastSent[key]) < 5*time.Minute {
+		mu.Unlock()
+		return
+	}
+	pending[key] = true
+	mu.Unlock()
+	once.Do(func() { go worker() })
+	select {
+	case queue <- event{tunnelID, kind, detail}:
+	default:
+		mu.Lock()
+		delete(pending, key)
+		mu.Unlock()
+		log.Print("[alertnotify] 通知队列已满")
+	}
+}
+
+func worker() {
+	for ev := range queue {
+		key := fmt.Sprintf("%d:%s", ev.tunnelID, ev.kind)
+		err := deliver(ev)
+		mu.Lock()
+		delete(pending, key)
+		if err == nil {
+			lastSent[key] = time.Now()
+		}
+		mu.Unlock()
+		if err != nil {
+			log.Printf("[alertnotify] 发送失败 (%s): %v", key, err)
+		}
+	}
+}
+
+func deliver(ev event) error {
+	row, err := current()
+	if err != nil || !row.Enabled {
+		return err
+	}
+	webhook, err := store.DecryptSecret(row.WebhookEncrypted)
+	if err != nil {
+		return err
+	}
+	secret := ""
+	if row.SigningEncrypted != "" {
+		secret, err = store.DecryptSecret(row.SigningEncrypted)
+		if err != nil {
+			return err
+		}
+	}
+	return send(webhook, secret, fmt.Sprintf("OpsX VPN 流量告警：专线 #%d / %s\n%s", ev.tunnelID, ev.kind, ev.detail))
+}