Sync Source and Authenticate Listener

The source table was extended to hold a hashed password,
listener_password_hash, which is now synchronized into a new Source
type, being hold in the RuntimeConfig.

This value is now being used in the Listener to enforce authenticated
API requests.

README.md

@@ -25,8 +25,12 @@ It is required that you have created a new database and imported the [schema](sc
 
 Additionally, it also requires you to manually insert items into the **source** table before starting the daemon.
 ```sql
-INSERT INTO source (id, type, name) VALUES (1, 'icinga2', 'Icinga 2')
+INSERT INTO source (id, type, name, listener_password_hash)
+VALUES (1, 'icinga2', 'Icinga 2', '$2y$10$QU8bJ7cpW1SmoVQ/RndX5O2J5L1PJF7NZ2dlIW7Rv3zUEcbUFg3z2');
 ```
+The `listener_password_hash` is a [PHP `password_hash`](https://www.php.net/manual/en/function.password-hash.php) with the `PASSWORD_DEFAULT` algorithm, currently bcrypt.
+In the example above, this is "correct horse battery staple".
+This mimics Icinga Web 2's behavior, as stated in [its documentation](https://icinga.com/docs/icinga-web/latest/doc/20-Advanced-Topics/#manual-user-creation-for-database-authentication-backend).
 
 Then, you can launch the daemon with the following command.
 ```go

go.mod

@@ -29,8 +29,9 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/ssgreg/journald v1.0.0 // indirect
 	go.uber.org/multierr v1.10.0 // indirect
+	golang.org/x/crypto v0.16.0 // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.6.0 // indirect
+	golang.org/x/sys v0.15.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -59,9 +59,12 @@ go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN8
 go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo=
 go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so=
 golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A=
+golang.org/x/crypto v0.16.0 h1:mMMrFzRSCF0GvB7Ne27XVtVAaXLrPmgPC7/v0tkwHaY=
+golang.org/x/crypto v0.16.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
 golang.org/x/exp v0.0.0-20220613132600-b0d781184e0d h1:vtUKgx8dahOomfFzLREU8nSv25YHnTgLBn4rDnWZdU0=
 golang.org/x/exp v0.0.0-20220613132600-b0d781184e0d/go.mod h1:Kr81I6Kryrl9sr8s2FK3vxD90NdsKWRuOIl2O4CvYbA=
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 h1:DzZ89McO9/gWPsQXS/FVKAlG02ZjaQ6AlZRBimEYOd0=
+golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -71,7 +74,10 @@ golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
+golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 h1:H2TDz8ibqkAF6YGhCdN3jS9O0/s90v0rJh3X/OLHEUk=
 golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=

icinga2.conf

@@ -5,9 +5,12 @@ if (!globals.contains("IcingaNotificationsIcingaWebUrl")) {
 	const IcingaNotificationsIcingaWebUrl = "http://localhost/icingaweb2"
 }
 if (!globals.contains("IcingaNotificationsEventSourceId")) {
-	// INSERT INTO source (id, type, name) VALUES (1, 'icinga2', 'Icinga 2')
+	// INSERT INTO source (id, type, name, listener_password_hash) VALUES (1, 'icinga2', 'Icinga 2', '$2y$10$QU8bJ7cpW1SmoVQ/RndX5O2J5L1PJF7NZ2dlIW7Rv3zUEcbUFg3z2')
 	const IcingaNotificationsEventSourceId = 1
 }
+if (!globals.contains("IcingaNotificationsAuth")) {
+	const IcingaNotificationsAuth = "source-1:correct horse battery staple"
+}
 
 // urlencode a string loosely based on RFC 3986.
 //
@@ -55,6 +58,7 @@ var baseBody = {
 			(len(macro("$event_severity$")) > 0 || len(macro("$event_type$")) > 0) ? "curl" : "true"
 		}}
 	}
+	"--user" = { value = IcingaNotificationsAuth }
 	"--fail" = { set_if = true }
 	"--silent" = { set_if = true }
 	"--show-error" = { set_if = true }

internal/config/runtime.go

@@ -3,6 +3,7 @@ package config
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"github.com/icinga/icinga-notifications/internal/channel"
 	"github.com/icinga/icinga-notifications/internal/recipient"
 	"github.com/icinga/icinga-notifications/internal/rule"
@@ -11,6 +12,9 @@ import (
 	"github.com/icinga/icingadb/pkg/logging"
 	"github.com/jmoiron/sqlx"
 	"go.uber.org/zap"
+	"golang.org/x/crypto/bcrypt"
+	"regexp"
+	"strconv"
 	"sync"
 	"time"
 )
@@ -44,6 +48,7 @@ type ConfigSet struct {
 	TimePeriods      map[int64]*timeperiod.TimePeriod
 	Schedules        map[int64]*recipient.Schedule
 	Rules            map[int64]*rule.Rule
+	Sources          map[int64]*Source
 }
 
 func (r *RuntimeConfig) UpdateFromDatabase(ctx context.Context) error {
@@ -137,6 +142,51 @@ func (r *RuntimeConfig) GetContact(username string) *recipient.Contact {
 	return nil
 }
 
+// getSourceFromCredentialsUsernameRegexp is used in GetSourceFromCredentials to parse the HTTP basic auth username.
+//
+// Even when Source.ListenerPasswordHash is of type int64, the optional leading minus must not be parsed as the data
+// type within PostgreSQL, bigserial, is positive.
+var getSourceFromCredentialsUsernameRegexp = regexp.MustCompile(`source-(\d+)`)
+
+// GetSourceFromCredentials verifies a credential pair against known Sources.
+//
+// This method returns either a *Source or a nil pointer and logs the cause to the given logger. This is in almost all
+// cases a debug logging message, except when something server-side is wrong, e.g., the hash is invalid.
+func (r *RuntimeConfig) GetSourceFromCredentials(user, pass string, logger *logging.Logger) *Source {
+	r.RLock()
+	defer r.RUnlock()
+
+	usernameMatches := getSourceFromCredentialsUsernameRegexp.FindStringSubmatch(user)
+	if usernameMatches == nil {
+		logger.Debugw("Cannot extract source ID from HTTP basic auth username", zap.String("user-input", user))
+		return nil
+	}
+	sourceId, err := strconv.ParseInt(usernameMatches[1], 10, 64)
+	if err != nil {
+		logger.Debugw("Cannot convert extracted source Id to int", zap.String("user-input", user), zap.Error(err))
+		return nil
+	}
+
+	source, ok := r.Sources[sourceId]
+	if !ok {
+		logger.Debugw("Cannot check credentials for unknown source ID", zap.Int64("id", sourceId))
+		return nil
+	}
+
+	// If either PHP's PASSWORD_DEFAULT changes or Icinga Web 2 starts using something else, e.g., Argon2id, this will
+	// return a descriptive error as the identifier does no longer match the bcrypt "$2y$".
+	err = bcrypt.CompareHashAndPassword([]byte(source.ListenerPasswordHash), []byte(pass))
+	if errors.Is(err, bcrypt.ErrMismatchedHashAndPassword) {
+		logger.Debugw("Invalid password for this source", zap.Int64("id", sourceId))
+		return nil
+	} else if err != nil {
+		logger.Errorw("Failed to verify password for this source", zap.Int64("id", sourceId), zap.Error(err))
+		return nil
+	}
+
+	return source
+}
+
 func (r *RuntimeConfig) fetchFromDatabase(ctx context.Context) error {
 	r.logger.Debug("fetching configuration from database")
 	start := time.Now()
@@ -162,6 +212,7 @@ func (r *RuntimeConfig) fetchFromDatabase(ctx context.Context) error {
 		r.fetchTimePeriods,
 		r.fetchSchedules,
 		r.fetchRules,
+		r.fetchSources,
 	}
 	for _, f := range updateFuncs {
 		if err := f(ctx, tx); err != nil {
@@ -188,6 +239,7 @@ func (r *RuntimeConfig) applyPending() {
 	r.applyPendingTimePeriods()
 	r.applyPendingSchedules()
 	r.applyPendingRules()
+	r.applyPendingSources()
 
 	r.logger.Debugw("applied pending configuration", zap.Duration("took", time.Since(start)))
 }

internal/config/source.go

@@ -0,0 +1,78 @@
+package config
+
+import (
+	"context"
+	"github.com/jmoiron/sqlx"
+	"go.uber.org/zap"
+)
+
+// Source entry within the ConfigSet to describe a source.
+type Source struct {
+	ID   int64  `db:"id"`
+	Type string `db:"type"`
+	Name string `db:"name"`
+
+	ListenerPasswordHash string `db:"listener_password_hash"`
+}
+
+func (r *RuntimeConfig) fetchSources(ctx context.Context, tx *sqlx.Tx) error {
+	var sourcePtr *Source
+	stmt := r.db.BuildSelectStmt(sourcePtr, sourcePtr)
+	r.logger.Debugf("Executing query %q", stmt)
+
+	var sources []*Source
+	if err := tx.SelectContext(ctx, &sources, stmt); err != nil {
+		r.logger.Errorln(err)
+		return err
+	}
+
+	sourcesById := make(map[int64]*Source)
+	for _, s := range sources {
+		sourceLogger := r.logger.With(
+			zap.Int64("id", s.ID),
+			zap.String("name", s.Name),
+			zap.String("type", s.Type),
+		)
+		if sourcesById[s.ID] != nil {
+			sourceLogger.Warnw("ignoring duplicate config for source ID")
+		} else {
+			sourcesById[s.ID] = s
+
+			sourceLogger.Debugw("loaded source config")
+		}
+	}
+
+	if r.Sources != nil {
+		// mark no longer existing sources for deletion
+		for id := range r.Sources {
+			if _, ok := sourcesById[id]; !ok {
+				sourcesById[id] = nil
+			}
+		}
+	}
+
+	r.pending.Sources = sourcesById
+
+	return nil
+}
+
+func (r *RuntimeConfig) applyPendingSources() {
+	if r.Sources == nil {
+		r.Sources = make(map[int64]*Source)
+	}
+
+	for id, pendingSource := range r.pending.Sources {
+		if pendingSource == nil {
+			r.logger.Infow("Source has been removed",
+				zap.Int64("id", r.Sources[id].ID),
+				zap.String("name", r.Sources[id].Name),
+				zap.String("type", r.Sources[id].Type))
+
+			delete(r.Sources, id)
+		} else {
+			r.Sources[id] = pendingSource
+		}
+	}
+
+	r.pending.Sources = nil
+}

internal/listener/listener.go

@@ -79,42 +79,69 @@ func (l *Listener) Run(ctx context.Context) error {
 }
 
 func (l *Listener) ProcessEvent(w http.ResponseWriter, req *http.Request) {
+	// abort the current connection by sending the status code and an error both to the log and back to the client.
+	abort := func(statusCode int, ev *event.Event, format string, a ...any) {
+		msg := format
+		if len(a) > 0 {
+			msg = fmt.Sprintf(format, a...)
+		}
+
+		logger := l.logger.With(zap.Int("status-code", statusCode), zap.String("msg", msg))
+		if ev != nil {
+			logger = logger.With(zap.Stringer("event", ev))
+		}
+
+		http.Error(w, msg, statusCode)
+		logger.Debugw("Abort listener submitted event processing")
+	}
+
 	if req.Method != http.MethodPost {
-		w.WriteHeader(http.StatusMethodNotAllowed)
-		_, _ = fmt.Fprintln(w, "POST required")
+		abort(http.StatusMethodNotAllowed, nil, "POST required")
+		return
+	}
+
+	var source *config.Source
+	if authUser, authPass, authOk := req.BasicAuth(); authOk {
+		source = l.runtimeConfig.GetSourceFromCredentials(authUser, authPass, l.logger)
+	}
+	if source == nil {
+		w.Header().Set("WWW-Authenticate", `Basic realm="icinga-notifications"`)
+		abort(http.StatusUnauthorized, nil, "HTTP authorization required")
 		return
 	}
 
 	var ev event.Event
 	err := json.NewDecoder(req.Body).Decode(&ev)
 	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		_, _ = fmt.Fprintf(w, "cannot parse JSON body: %v\n", err)
+		abort(http.StatusBadRequest, nil, "cannot parse JSON body: %v", err)
 		return
 	}
 	ev.Time = time.Now()
 
+	if ev.SourceId != source.ID {
+		abort(http.StatusBadRequest, &ev, "authenticated for source %d with an event for %d", source.ID, ev.SourceId)
+		return
+	}
+
 	if ev.Severity == event.SeverityNone && ev.Type == "" {
-		w.WriteHeader(http.StatusBadRequest)
-		_, _ = fmt.Fprintln(w, "ignoring invalid event: must set 'type' or 'severity'")
+		abort(http.StatusBadRequest, &ev, "ignoring invalid event: must set 'type' or 'severity'")
 		return
 	}
 
 	if ev.Severity != event.SeverityNone {
 		if ev.Type == "" {
 			ev.Type = event.TypeState
 		} else if ev.Type != event.TypeState {
-			w.WriteHeader(http.StatusBadRequest)
-			_, _ = fmt.Fprintf(w, "ignoring invalid event: if 'severity' is set, 'type' must not be set or set to %q\n", event.TypeState)
+			abort(http.StatusBadRequest, &ev,
+				"ignoring invalid event: if 'severity' is set, 'type' must not be set or set to %q", event.TypeState)
 			return
 		}
 	}
 
 	if ev.Severity == event.SeverityNone {
 		if ev.Type != event.TypeAcknowledgement {
 			// It's neither a state nor an acknowledgement event.
-			w.WriteHeader(http.StatusBadRequest)
-			_, _ = fmt.Fprintf(w, "received not a state/acknowledgement event, ignoring\n")
+			abort(http.StatusBadRequest, &ev, "received not a state/acknowledgement event, ignoring")
 			return
 		}
 	}
@@ -123,17 +150,14 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, req *http.Request) {
 	obj, err := object.FromEvent(ctx, l.db, &ev)
 	if err != nil {
 		l.logger.Errorw("Can't sync object", zap.Error(err))
-
-		w.WriteHeader(http.StatusInternalServerError)
-		_, _ = fmt.Fprintln(w, err.Error())
+		abort(http.StatusInternalServerError, &ev, err.Error())
 		return
 	}
 
 	createIncident := ev.Severity != event.SeverityNone && ev.Severity != event.SeverityOK
 	currentIncident, created, err := incident.GetCurrent(ctx, l.db, obj, l.logs.GetChildLogger("incident"), l.runtimeConfig, createIncident)
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		_, _ = fmt.Fprintln(w, err)
+		abort(http.StatusInternalServerError, &ev, err.Error())
 		return
 	}
 
@@ -159,12 +183,11 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	l.logger.Infof("Processing event")
+	l.logger.Infow("Processing event", zap.String("event", ev.String()))
 
 	err = currentIncident.ProcessEvent(ctx, &ev, created)
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		_, _ = fmt.Fprintln(w, err)
+		abort(http.StatusInternalServerError, &ev, err.Error())
 		return
 	}
 

schema/pgsql/schema.sql

@@ -130,6 +130,14 @@ CREATE TABLE source (
     -- will likely need a distinguishing value for multiple sources of the same type in the future, like for example
     -- the Icinga DB environment ID for Icinga 2 sources
 
+    -- listener_password_hash is required to limit API access for incoming connections to the Listener. The username is
+    -- "source-${id}", allowing an early verification before having to parse the POSTed event.
+    -- The hash is PHP's password_hash with the PASSWORD_DEFAULT algorithm, as already being used in Icinga Web 2.
+    -- https://icinga.com/docs/icinga-web/latest/doc/20-Advanced-Topics/#manual-user-creation-for-database-authentication-backend
+    --
+    -- This behavior might change in the future to become "type"-dependable.
+    listener_password_hash text NOT NULL,
+
     CONSTRAINT pk_source PRIMARY KEY (id)
 );
 

schema/pgsql/upgrades/019.sql

@@ -0,0 +1,4 @@
+ALTER TABLE source ADD COLUMN listener_password_hash text;
+-- php > $listener_password_hash = password_hash("correct horse battery staple", PASSWORD_DEFAULT));
+UPDATE source SET listener_password_hash = '$2y$10$QU8bJ7cpW1SmoVQ/RndX5O2J5L1PJF7NZ2dlIW7Rv3zUEcbUFg3z2';
+ALTER TABLE source ALTER COLUMN listener_password_hash SET NOT NULL;