ntfy-alertmanager/main.go

// A bridge between ntfy and Alertmanager
package main

import (
	"crypto/sha512"
	"crypto/subtle"
	_ "embed"
	"encoding/base64"
	"encoding/json"
	"flag"
	"fmt"
	"net/http"
	"os"
	"strings"
	"time"

	"git.xenrox.net/~xenrox/go-log"
	"golang.org/x/text/cases"
	"golang.org/x/text/language"
)

var version = "dev"

type receiver struct {
	cfg    *config
	logger *log.Logger
	cache  *cache
}

type payload struct {
	Status            string            `json:"status"`
	Alerts            []alert           `json:"alerts"`
	GroupLabels       map[string]string `json:"groupLabels"`
	CommonLabels      map[string]string `json:"commonLabels"`
	CommonAnnotations map[string]string `json:"commonAnnotations"`
	ExternalURL       string            `json:"externalURL"`
}

type alert struct {
	Status      string            `json:"status"`
	Labels      map[string]string `json:"labels"`
	Annotations map[string]string `json:"annotations"`
	Fingerprint fingerprint       `json:"fingerprint"`
}

type notification struct {
	title       string
	body        string
	priority    string
	tags        string
	silenceBody string
	fingerprint fingerprint
	status      string
}

func (rcv *receiver) singleAlertNotifications(p *payload) []*notification {
	var notifications []*notification
	for _, alert := range p.Alerts {
		if rcv.cache.contains(alert.Fingerprint, status(alert.Status)) {
			rcv.logger.Debugf("Alert %s skipped: Still in cache", alert.Fingerprint)
			continue
		}

		n := new(notification)
		n.fingerprint = alert.Fingerprint
		n.status = alert.Status

		// create title
		n.title = fmt.Sprintf("[%s]", strings.ToUpper(alert.Status))
		if name, ok := alert.Labels["alertname"]; ok {
			n.title = fmt.Sprintf("%s %s", n.title, name)
		}

		for _, value := range p.GroupLabels {
			n.title = fmt.Sprintf("%s %s", n.title, value)
		}

		// create body
		n.body = "Labels:\n"
		for key, value := range alert.Labels {
			n.body = fmt.Sprintf("%s%s = %s\n", n.body, key, value)
		}

		n.body += "\nAnnotations:\n"
		for key, value := range alert.Annotations {
			n.body = fmt.Sprintf("%s%s = %s\n", n.body, key, value)
		}

		var tags []string
		if alert.Status == "resolved" {
			tags = append(tags, rcv.cfg.resolved.Tags...)
		}

		for _, labelName := range rcv.cfg.labels.Order {
			val, ok := alert.Labels[labelName]
			if !ok {
				continue
			}

			labelConfig, ok := rcv.cfg.labels.Label[fmt.Sprintf("%s:%s", labelName, val)]
			if !ok {
				continue
			}

			if n.priority == "" {
				n.priority = labelConfig.Priority
			}

			for _, val := range labelConfig.Tags {
				if !sliceContains(tags, val) {
					tags = append(tags, val)
				}
			}
		}

		n.tags = strings.Join(tags, ",")

		if rcv.cfg.am.SilenceDuration != 0 && alert.Status == "firing" {
			if rcv.cfg.BaseURL == "" {
				rcv.logger.Error("Failed to create silence action: No base-url set")
			} else {
				// I could not convince ntfy to accept an Action with a body which contains
				// a json with more than one key. Instead the json will be base64 encoded
				// and sent to the ntfy-alertmanager silences endpoint, that operates as
				// a proxy and will do the Alertmanager API request.
				s := &silenceBody{AlertManagerURL: p.ExternalURL, Labels: alert.Labels}
				b, err := json.Marshal(s)
				if err != nil {
					rcv.logger.Errorf("Failed to create silence action: %v", err)
				}

				n.silenceBody = base64.StdEncoding.EncodeToString(b)
			}
		}

		notifications = append(notifications, n)
	}

	return notifications
}

func (rcv *receiver) multiAlertNotification(p *payload) *notification {
	n := new(notification)

	// create title
	count := len(p.Alerts)
	title := fmt.Sprintf("[%s", strings.ToUpper(p.Status))
	if p.Status == "firing" {
		title = fmt.Sprintf("%s:%d", title, count)
	}

	title += "]"
	for _, value := range p.GroupLabels {
		title = fmt.Sprintf("%s %s", title, value)
	}
	n.title = title

	// create body
	var body string
	c := cases.Title(language.English)

	for _, alert := range p.Alerts {
		alertBody := fmt.Sprintf("%s\nLabels:\n", c.String(alert.Status))
		for key, value := range alert.Labels {
			alertBody = fmt.Sprintf("%s%s = %s\n", alertBody, key, value)
		}

		alertBody += "Annotations:\n"
		for key, value := range alert.Annotations {
			alertBody = fmt.Sprintf("%s%s = %s\n", alertBody, key, value)
		}

		alertBody += "\n"

		body += alertBody
	}
	n.body = body

	var priority string
	var tags []string
	if p.Status == "resolved" {
		tags = append(tags, rcv.cfg.resolved.Tags...)
	}

	for _, labelName := range rcv.cfg.labels.Order {
		val, ok := p.CommonLabels[labelName]
		if !ok {
			continue
		}

		labelConfig, ok := rcv.cfg.labels.Label[fmt.Sprintf("%s:%s", labelName, val)]
		if !ok {
			continue
		}

		if priority == "" {
			priority = labelConfig.Priority
		}

		for _, val := range labelConfig.Tags {
			if !sliceContains(tags, val) {
				tags = append(tags, val)
			}
		}
	}

	tagString := strings.Join(tags, ",")
	n.priority = priority
	n.tags = tagString

	if rcv.cfg.am.SilenceDuration != 0 && p.Status == "firing" {
		if rcv.cfg.BaseURL == "" {
			rcv.logger.Error("Failed to create silence action: No base-url set")
		} else {

			s := &silenceBody{AlertManagerURL: p.ExternalURL, Labels: p.CommonLabels}
			b, err := json.Marshal(s)
			if err != nil {
				rcv.logger.Errorf("Failed to create silence action: %v", err)
			}

			n.silenceBody = base64.StdEncoding.EncodeToString(b)
		}
	}

	return n
}

func (rcv *receiver) publish(n *notification) error {
	client := &http.Client{Timeout: time.Second * 3}
	req, err := http.NewRequest(http.MethodPost, rcv.cfg.ntfy.Topic, strings.NewReader(n.body))
	if err != nil {
		return err
	}

	// Basic auth
	if rcv.cfg.ntfy.Password != "" && rcv.cfg.ntfy.User != "" {
		auth := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s:%s", rcv.cfg.ntfy.User, rcv.cfg.ntfy.Password)))
		req.Header.Set("Authorization", fmt.Sprintf("Basic %s", auth))
	}

	req.Header.Set("X-Title", n.title)

	if n.priority != "" {
		req.Header.Set("X-Priority", n.priority)
	}

	if n.tags != "" {
		req.Header.Set("X-Tags", n.tags)
	}

	if n.silenceBody != "" {
		url := rcv.cfg.BaseURL + "/silences"

		var authString string
		if rcv.cfg.User != "" && rcv.cfg.Password != "" {
			auth := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s:%s", rcv.cfg.User, rcv.cfg.Password)))
			authString = fmt.Sprintf(", headers.Authorization=Basic %s", auth)
		}

		req.Header.Set("Actions", fmt.Sprintf("http, Silence, %s, method=POST, body=%s%s", url, n.silenceBody, authString))
	}

	resp, err := client.Do(req)
	if err != nil {
		return err
	}

	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("ntfy: received status code %d", resp.StatusCode)
	}

	return nil
}

func (rcv *receiver) handleWebhooks(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()

	if r.Method != http.MethodPost {
		http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed)
		rcv.logger.Debugf("illegal HTTP method: expected %q, got %q", "POST", r.Method)
		return
	}

	contentType := r.Header.Get("Content-Type")
	if contentType != "application/json" {
		http.Error(w, "Only application/json allowed", http.StatusUnsupportedMediaType)
		rcv.logger.Debugf("illegal content type: %s", contentType)
		return
	}

	var event payload
	if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
		rcv.logger.Debug(err)
		return
	}

	if rcv.logger.Level() == log.Debug {
		rcv.logger.Debugf("Received alert %+v", event)
	}

	if rcv.cfg.alertMode == single {
		notifications := rcv.singleAlertNotifications(&event)
		for _, n := range notifications {
			err := rcv.publish(n)
			if err != nil {
				rcv.logger.Errorf("Failed to publish notification: %v", err)
			} else {
				rcv.cache.set(n.fingerprint, status(n.status))
			}
		}
	} else {
		notification := rcv.multiAlertNotification(&event)
		err := rcv.publish(notification)
		if err != nil {
			rcv.logger.Errorf("Failed to publish notification: %v", err)
		}
	}
}

func (rcv *receiver) basicAuthMiddleware(handler http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		user, pass, ok := r.BasicAuth()
		if !ok {
			rcv.logger.Debug("basic auth failure")
			return
		}

		inputUserHash := sha512.Sum512([]byte(user))
		inputPassHash := sha512.Sum512([]byte(pass))
		configUserHash := sha512.Sum512([]byte(rcv.cfg.User))
		configPassHash := sha512.Sum512([]byte(rcv.cfg.Password))

		validUser := subtle.ConstantTimeCompare(inputUserHash[:], configUserHash[:])
		validPass := subtle.ConstantTimeCompare(inputPassHash[:], configPassHash[:])

		if validUser != 1 || validPass != 1 {
			http.Error(w, "Unauthorized", http.StatusUnauthorized)
			rcv.logger.Debug("basic auth: wrong user or password")
			return
		}

		handler(w, r)
	}
}

func (rcv *receiver) runCleanup() {
	for {
		time.Sleep(rcv.cfg.cache.CleanupInterval)
		rcv.logger.Info("Pruning cache")
		rcv.cache.cleanup()
	}
}

func main() {
	var configPath string
	flag.StringVar(&configPath, "config", "/etc/ntfy-alertmanager/config", "config file path")
	var showVersion bool
	flag.BoolVar(&showVersion, "version", false, "Show version and exit")
	flag.Parse()

	if showVersion {
		fmt.Println(version)
		os.Exit(0)
	}

	logger := log.NewDefaultLogger()

	cfg, err := readConfig(configPath)
	if err != nil {
		logger.Fatalf("config: %v", err)
	}

	if err := logger.SetLevelFromString(cfg.LogLevel); err != nil {
		logger.Errorf("config: %v", err)
	}

	receiver := &receiver{cfg: cfg, logger: logger, cache: newCache(cfg.cache.Duration)}

	logger.Infof("Listening on %s, ntfy-alertmanager %s", cfg.HTTPAddress, version)

	if cfg.User != "" && cfg.Password != "" {
		logger.Info("Enabling HTTP Basic Authentication")
		http.HandleFunc("/", receiver.basicAuthMiddleware(receiver.handleWebhooks))
		http.HandleFunc("/silences", receiver.basicAuthMiddleware(receiver.handleSilences))
	} else {
		http.HandleFunc("/", receiver.handleWebhooks)
		http.HandleFunc("/silences", receiver.handleSilences)
	}

	go receiver.runCleanup()
	logger.Fatal(http.ListenAndServe(cfg.HTTPAddress, nil))
}