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)) }