Merge pull request #2741 from manuelsc/BIDS-2811/machine_notifications

(BIDS-2811) machine filter fix

handlers/user.go

@@ -25,6 +25,7 @@ import (
 	"github.com/gorilla/csrf"
 	"github.com/gorilla/mux"
 	"github.com/lib/pq"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/crypto/bcrypt"
 )
@@ -1727,12 +1728,18 @@ func internUserNotificationsSubscribe(event, filter string, threshold float64, w
 
 	errFields["event_name"] = eventName
 
-	if !isValidSubscriptionFilter(eventName, filter) {
-		utils.LogError(nil, "error invalid filter: not pubkey or client for subscription", 0, errFields)
+	valid, err := isValidSubscriptionFilter(user.UserID, eventName, filter)
+	if err != nil {
+		utils.LogError(err, "error validating filter", 0, errFields)
 		ErrorOrJSONResponse(w, r, "Internal server error", http.StatusInternalServerError)
 		return false
 	}
 
+	if !valid {
+		ErrorOrJSONResponse(w, r, "Invalid filter, only pubkey, client or machine name is valid.", http.StatusBadRequest)
+		return false
+	}
+
 	userPremium := getUserPremium(r)
 
 	filterWatchlist := db.WatchlistFilter{
@@ -1916,13 +1923,19 @@ func internUserNotificationsUnsubscribe(event, filter string, w http.ResponseWri
 	}
 
 	errFields["event_name"] = eventName
+	valid, err := isValidSubscriptionFilter(user.UserID, eventName, filter)
 
-	if !isValidSubscriptionFilter(eventName, filter) {
-		utils.LogError(nil, "error invalid filter: not pubkey or client for unsubscription", 0, errFields)
+	if err != nil {
+		utils.LogError(err, "error validating filter", 0, errFields)
 		ErrorOrJSONResponse(w, r, "Internal server error", http.StatusInternalServerError)
 		return false
 	}
 
+	if !valid {
+		ErrorOrJSONResponse(w, r, "Invalid filter, only pubkey, client or machine name is valid.", http.StatusBadRequest)
+		return false
+	}
+
 	filterWatchlist := db.WatchlistFilter{
 		UserId:         user.UserID,
 		Validators:     nil,
@@ -2000,17 +2013,23 @@ func UserNotificationsUnsubscribe(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if !isValidSubscriptionFilter(eventName, filter) {
-		errMsg := fmt.Errorf("error invalid filter, not pubkey or client")
+	valid, err := isValidSubscriptionFilter(user.UserID, eventName, filter)
+	if err != nil {
+		errMsg := fmt.Errorf("error validating filter")
 		errFields := map[string]interface{}{
 			"filter":     filter,
 			"filter_len": len(filter)}
-		utils.LogError(nil, errMsg, 0, errFields)
+		utils.LogError(err, errMsg, 0, errFields)
 
 		ErrorOrJSONResponse(w, r, "Internal server error", http.StatusInternalServerError)
 		return
 	}
 
+	if !valid {
+		ErrorOrJSONResponse(w, r, "Invalid filter, only pubkey, client or machine name is valid.", http.StatusBadRequest)
+		return
+	}
+
 	filterLen := len(filter)
 	if filterLen == 0 && !types.IsUserIndexed(eventName) { // no filter = add all my watched validators
 
@@ -2060,7 +2079,7 @@ func UserNotificationsUnsubscribe(w http.ResponseWriter, r *http.Request) {
 	OKResponse(w, r)
 }
 
-func isValidSubscriptionFilter(eventName types.EventName, filter string) bool {
+func isValidSubscriptionFilter(userID uint64, eventName types.EventName, filter string) (bool, error) {
 	ethClients := []string{"geth", "nethermind", "besu", "erigon", "teku", "prysm", "nimbus", "lighthouse", "lodestar", "rocketpool", "mev-boost"}
 
 	isPkey := searchPubkeyExactRE.MatchString(filter)
@@ -2078,7 +2097,43 @@ func isValidSubscriptionFilter(eventName types.EventName, filter string) bool {
 		isClient = true
 	}
 
-	return len(filter) == 0 || isPkey || isClient
+	isValidMachine := false
+	if types.IsMachineNotification(eventName) {
+		machines, err := db.BigtableClient.GetMachineMetricsMachineNames(userID)
+		if err != nil {
+			return false, errors.Wrap(err, "can not get users machines from bigtable for validation")
+		}
+		for _, userMachineName := range machines {
+			if userMachineName == filter {
+				isValidMachine = true
+				break
+			}
+		}
+
+		// While the above works fine for active machines (adding a new notification to an active machine)
+		// It does not work for a machine that is offline and where the user wants to subscribe/unsubscribe from this machine.
+		// So check the db for any machine names as well
+		if !isValidMachine {
+			machines := make([]string, 0)
+			err = db.FrontendWriterDB.Select(&machines, `
+				select event_filter
+				from users_subscriptions 
+				where user_id = $1 AND event_name = ANY($2)
+			`, userID, pq.Array(types.MachineEvents))
+			if err != nil {
+				return false, errors.Wrap(err, "can not get event_filters from db for validation")
+			}
+
+			for _, machineName := range machines {
+				if machineName == filter {
+					isValidMachine = true
+					break
+				}
+			}
+		}
+	}
+
+	return len(filter) == 0 || isPkey || isClient || isValidMachine, nil
 }
 
 func UserNotificationsUnsubscribeByHash(w http.ResponseWriter, r *http.Request) {

types/frontend.go

@@ -51,6 +51,16 @@ const (
 	SyncCommitteeSoon                                EventName = "validator_synccommittee_soon"
 )
 
+var MachineEvents = []EventName{
+	MonitoringMachineCpuLoadEventName,
+	MonitoringMachineOfflineEventName,
+	MonitoringMachineDiskAlmostFullEventName,
+	MonitoringMachineCpuLoadEventName,
+	MonitoringMachineMemoryUsageEventName,
+	MonitoringMachineSwitchedToETH2FallbackEventName,
+	MonitoringMachineSwitchedToETH1FallbackEventName,
+}
+
 var UserIndexEvents = []EventName{
 	EthClientUpdateEventName,
 	MonitoringMachineCpuLoadEventName,
@@ -103,6 +113,15 @@ func IsUserIndexed(event EventName) bool {
 	return false
 }
 
+func IsMachineNotification(event EventName) bool {
+	for _, ev := range MachineEvents {
+		if ev == event {
+			return true
+		}
+	}
+	return false
+}
+
 var EventNames = []EventName{
 	ValidatorBalanceDecreasedEventName,
 	ValidatorExecutedProposalEventName,