Live: channel ID validation (#34215)

pkg/services/live/features/dashboard.go

@@ -102,27 +102,27 @@ func (h *DashboardHandler) OnPublish(ctx context.Context, user *models.SignedInU
 		event := dashboardEvent{}
 		err := json.Unmarshal(e.Data, &event)
 		if err != nil || event.UID != parts[1] {
-			return models.PublishReply{}, backend.SubscribeStreamStatusNotFound, fmt.Errorf("bad request")
+			return models.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("bad request")
 		}
 		if event.Action != EDITING_STARTED {
 			// just ignore the event
-			return models.PublishReply{}, backend.SubscribeStreamStatusNotFound, fmt.Errorf("ignore???")
+			return models.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("ignore???")
 		}
 		query := models.GetDashboardQuery{Uid: parts[1], OrgId: user.OrgId}
 		if err := bus.Dispatch(&query); err != nil {
 			logger.Error("Unknown dashboard", "query", query)
-			return models.PublishReply{}, backend.SubscribeStreamStatusNotFound, nil
+			return models.PublishReply{}, backend.PublishStreamStatusNotFound, nil
 		}
 
 		guardian := guardian.New(query.Result.Id, user.OrgId, user)
 		canEdit, err := guardian.CanEdit()
 		if err != nil {
-			return models.PublishReply{}, backend.SubscribeStreamStatusNotFound, fmt.Errorf("internal error")
+			return models.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("internal error")
 		}
 
 		// Ignore edit events if the user can not edit
 		if !canEdit {
-			return models.PublishReply{}, backend.SubscribeStreamStatusNotFound, nil // NOOP
+			return models.PublishReply{}, backend.PublishStreamStatusNotFound, nil // NOOP
 		}
 
 		// Tell everyone who is editing
@@ -130,12 +130,12 @@ func (h *DashboardHandler) OnPublish(ctx context.Context, user *models.SignedInU
 
 		msg, err := json.Marshal(event)
 		if err != nil {
-			return models.PublishReply{}, backend.SubscribeStreamStatusNotFound, fmt.Errorf("internal error")
+			return models.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("internal error")
 		}
 		return models.PublishReply{Data: msg}, backend.PublishStreamStatusOK, nil
 	}
 
-	return models.PublishReply{}, backend.SubscribeStreamStatusNotFound, nil
+	return models.PublishReply{}, backend.PublishStreamStatusNotFound, nil
 }
 
 // DashboardSaved should broadcast to the appropriate stream

pkg/services/live/live.go

@@ -2,6 +2,7 @@ package live
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -326,6 +327,10 @@ func (g *GrafanaLive) handleOnSubscribe(client *centrifuge.Client, e centrifuge.
 
 	handler, addr, err := g.GetChannelHandler(user, channel)
 	if err != nil {
+		if errors.Is(err, live.ErrInvalidChannelID) {
+			logger.Info("Invalid channel ID", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
+			return centrifuge.SubscribeReply{}, &centrifuge.Error{Code: uint32(http.StatusBadRequest), Message: "invalid channel ID"}
+		}
 		logger.Error("Error getting channel handler", "user", client.UserID(), "client", client.ID(), "channel", e.Channel, "error", err)
 		return centrifuge.SubscribeReply{}, centrifuge.ErrorInternal
 	}
@@ -377,6 +382,10 @@ func (g *GrafanaLive) handleOnPublish(client *centrifuge.Client, e centrifuge.Pu
 
 	handler, addr, err := g.GetChannelHandler(user, channel)
 	if err != nil {
+		if errors.Is(err, live.ErrInvalidChannelID) {
+			logger.Info("Invalid channel ID", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
+			return centrifuge.PublishReply{}, &centrifuge.Error{Code: uint32(http.StatusBadRequest), Message: "invalid channel ID"}
+		}
 		logger.Error("Error getting channel handler", "user", client.UserID(), "client", client.ID(), "channel", e.Channel, "error", err)
 		return centrifuge.PublishReply{}, centrifuge.ErrorInternal
 	}
@@ -442,9 +451,9 @@ func publishStatusToHTTPError(status backend.PublishStreamStatus) (int, string)
 // GetChannelHandler gives thread-safe access to the channel.
 func (g *GrafanaLive) GetChannelHandler(user *models.SignedInUser, channel string) (models.ChannelHandler, live.Channel, error) {
 	// Parse the identifier ${scope}/${namespace}/${path}
-	addr := live.ParseChannel(channel)
-	if !addr.IsValid() {
-		return nil, live.Channel{}, fmt.Errorf("invalid channel: %q", channel)
+	addr, err := live.ParseChannel(channel)
+	if err != nil {
+		return nil, live.Channel{}, err
 	}
 
 	g.channelsMu.RLock()
@@ -569,9 +578,9 @@ func (g *GrafanaLive) ClientCount(orgID int64, channel string) (int, error) {
 }
 
 func (g *GrafanaLive) HandleHTTPPublish(ctx *models.ReqContext, cmd dtos.LivePublishCmd) response.Response {
-	addr := live.ParseChannel(cmd.Channel)
-	if !addr.IsValid() {
-		return response.Error(http.StatusBadRequest, "Bad channel address", nil)
+	addr, err := live.ParseChannel(cmd.Channel)
+	if err != nil {
+		return response.Error(http.StatusBadRequest, "invalid channel ID", nil)
 	}
 
 	logger.Debug("Publish API cmd", "user", ctx.SignedInUser.UserId, "channel", cmd.Channel)