package sandbar import ( "bytes" "encoding/json" "errors" "fmt" "log" "net/http" "net/smtp" "net/url" "regexp" "strconv" "strings" "sync" "time" "git.veillax.com/Veillax/reef/netguard" ) // Notify is the ntfy-shaped part: anything can publish a message to a topic with // an HTTP request, and anything can subscribe to it (SSE, newline-delimited JSON, // or the web UI). Alerts and incidents publish here too, so a phone with the ntfy // app subscribed to "alerts" is a working pager with no other setup. type Message struct { ID string `json:"id"` Topic string `json:"topic"` Time int64 `json:"time"` Title string `json:"title,omitempty"` Message string `json:"message"` Priority int `json:"priority"` Tags []string `json:"tags,omitempty"` } var topicRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) // reservedTopics are first path segments that belong to Sandbar itself. var reservedTopics = map[string]bool{"api": true, "status": true, "hb": true, "metrics": true, "static": true, "healthz": true} func validTopic(t string) bool { return topicRe.MatchString(t) && !reservedTopics[t] } // hub fans published messages out to live subscribers. type hub struct { mu sync.Mutex subs map[string]map[chan Message]bool } func newHub() *hub { return &hub{subs: map[string]map[chan Message]bool{}} } func (h *hub) subscribe(topic string) (chan Message, func()) { ch := make(chan Message, 64) h.mu.Lock() if h.subs[topic] == nil { h.subs[topic] = map[chan Message]bool{} } h.subs[topic][ch] = true h.mu.Unlock() return ch, func() { h.mu.Lock() delete(h.subs[topic], ch) h.mu.Unlock() } } func (h *hub) publish(m Message) { h.mu.Lock() defer h.mu.Unlock() for ch := range h.subs[m.Topic] { select { case ch <- m: default: // a stalled subscriber drops messages instead of blocking publishers } } } func parsePriority(s string) int { switch strings.ToLower(strings.TrimSpace(s)) { case "5", "max", "urgent": return 5 case "4", "high": return 4 case "2", "low": return 2 case "1", "min": return 1 } return 3 } func parseTags(s string) []string { var out []string for _, t := range strings.Split(s, ",") { if t = strings.TrimSpace(t); t != "" { out = append(out, t) } } return out } // Publish stores a message and delivers it to live subscribers. func (s *Server) Publish(topic, title, body string, prio int, tags []string) (Message, error) { if !validTopic(topic) { return Message{}, errors.New("a topic is 1-64 letters, digits, - or _") } m := Message{ID: newID(), Topic: topic, Time: s.now().Unix(), Title: title, Message: body, Priority: prio, Tags: tags} tj, _ := json.Marshal(tags) if _, err := s.st.db.Exec(`INSERT INTO messages(id,topic,time,title,body,priority,tags) VALUES(?,?,?,?,?,?,?)`, m.ID, m.Topic, m.Time, m.Title, m.Message, m.Priority, string(tj)); err != nil { return m, err } s.hub.publish(m) return m, nil } // MessagesSince returns stored messages for a topic after the given unix time, oldest first. func (s *Server) MessagesSince(topic string, since int64, limit int) []Message { out := []Message{} rows, err := s.st.db.Query(`SELECT id,time,title,body,priority,tags FROM messages WHERE topic=? AND time>=? ORDER BY time, rowid LIMIT ?`, topic, since, limit) if err != nil { return out } defer rows.Close() for rows.Next() { m := Message{Topic: topic} var tj string if rows.Scan(&m.ID, &m.Time, &m.Title, &m.Message, &m.Priority, &tj) == nil { json.Unmarshal([]byte(tj), &m.Tags) out = append(out, m) } } return out } // ---------- outputs: email and webhooks ---------- // Channel is a place broadcast notifications go: an email address list or a webhook. type Channel struct { ID string `json:"id"` Name string `json:"name"` Kind string `json:"kind"` // "email" or "webhook" Target string `json:"target"` // addresses, or a URL (secret: URLs often carry tokens) Format string `json:"format"` // webhook body: json, slack, discord or ntfy AllowPrivate bool `json:"allowPrivate"` Enabled bool `json:"enabled"` } // Note is what every notification is built from. type Note struct { Status string `json:"status"` // firing, resolved, acked, test, declared, update Kind string `json:"kind"` // alert, monitor, incident Title string `json:"title"` Message string `json:"message"` Severity string `json:"severity,omitempty"` Link string `json:"link,omitempty"` Time int64 `json:"time"` } func (n Note) subject() string { switch n.Status { case "resolved": return "[Sandbar] Resolved: " + n.Title case "test": return "[Sandbar] Test notification" case "acked": return "[Sandbar] Acknowledged: " + n.Title } return "[Sandbar] " + strings.ToUpper(n.Kind[:1]) + n.Kind[1:] + ": " + n.Title } func (n Note) text() string { if n.Status == "test" { return "This is a test from Sandbar. If you can read this, the channel works." } t := n.Message if n.Link != "" { t += "\n" + n.Link } return t } func (n Note) priority() int { switch { case n.Status == "resolved" || n.Status == "acked" || n.Status == "test": return 3 case n.Severity == "critical": return 5 } return 4 } func (n Note) tags() []string { switch n.Status { case "resolved": return []string{"white_check_mark"} case "acked": return []string{"eyes"} case "test": return []string{"test_tube"} } return []string{"rotating_light"} } func webhookBody(format string, n Note) (body []byte, ctype string, hdr map[string]string) { text := n.subject() + "\n" + n.text() switch format { case "slack": body, _ = json.Marshal(map[string]string{"text": text}) case "discord": body, _ = json.Marshal(map[string]string{"content": text}) case "ntfy": return []byte(n.text()), "text/plain", map[string]string{ "Title": n.subject(), "Priority": strconv.Itoa(n.priority()), "Tags": strings.Join(n.tags(), ","), } default: body, _ = json.Marshal(n) } return body, "application/json", nil } func (s *Server) sendChannel(ch Channel, n Note) error { switch ch.Kind { case "email": var errs []error for _, to := range strings.Split(s.st.open(ch.Target), ",") { if to = strings.TrimSpace(to); to != "" { if err := s.Mail(to, n.subject(), n.text()); err != nil { errs = append(errs, err) } } } return errors.Join(errs...) case "webhook": body, ctype, hdr := webhookBody(ch.Format, n) req, err := http.NewRequest("POST", s.st.open(ch.Target), bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Content-Type", ctype) for k, v := range hdr { req.Header.Set(k, v) } resp, err := s.httpClient(ch.AllowPrivate).Do(req) if err != nil { return err } resp.Body.Close() if resp.StatusCode >= 300 { return fmt.Errorf("the webhook answered %s", resp.Status) } return nil } return errors.New("unknown channel kind") } func (s *Server) httpClient(allowPrivate bool) *http.Client { if s.opts.HTTPClient != nil { return s.opts.HTTPClient } c := netguard.Client(allowPrivate) c.Timeout = 15 * time.Second return c } // Mail sends one plain-text email through the configured SMTP server. func (s *Server) Mail(to, subject, body string) error { if s.opts.Mailer != nil { return s.opts.Mailer(to, subject, body) } c := s.opts.SMTP if c.Host == "" { return errors.New("no SMTP server configured (SANDBAR_SMTP_HOST)") } if strings.ContainsAny(to+subject, "\r\n") { return errors.New("invalid header") } port := c.Port if port == "" { port = "587" } msg := "From: " + c.From + "\r\nTo: " + to + "\r\nSubject: " + subject + "\r\nDate: " + s.now().Format(time.RFC1123Z) + "\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n" + body + "\r\n" var auth smtp.Auth if c.User != "" { auth = smtp.PlainAuth("", c.User, c.Password, c.Host) } return smtp.SendMail(c.Host+":"+port, auth, c.From, []string{to}, []byte(msg)) } // broadcast sends a note to the topic named by the source and every enabled channel. func (s *Server) broadcast(n Note, topics ...string) { for _, t := range topics { if _, err := s.Publish(t, n.subject(), n.text(), n.priority(), n.tags()); err != nil { log.Printf("publish %s: %v", t, err) } } chans, _ := listOf[Channel](s.st, "channel") for _, ch := range chans { if !ch.Enabled { continue } ch := ch go func() { if err := s.sendChannel(ch, n); err != nil { log.Printf("channel %q: %v", ch.Name, err) } }() } } // validateChannel normalises ch. func validateChannel(ch *Channel) error { ch.Name = strings.TrimSpace(ch.Name) ch.Target = strings.TrimSpace(ch.Target) switch ch.Kind { case "email": for _, a := range strings.Split(ch.Target, ",") { if a = strings.TrimSpace(a); !strings.Contains(a, "@") || strings.ContainsAny(a, " \r\n<>") { return errors.New("enter one or more email addresses, separated by commas") } } ch.Format, ch.AllowPrivate = "", false case "webhook": u, err := url.Parse(ch.Target) if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" { return errors.New("the webhook must be an http:// or https:// URL") } switch ch.Format { case "json", "slack", "discord", "ntfy": case "": ch.Format = "json" default: return errors.New("format must be json, slack, discord or ntfy") } default: return errors.New("a channel is an email or a webhook") } if ch.Name == "" { ch.Name = ch.Kind } return nil }