Live: pipeline subscription authorization based on user role (#39587)

pkg/services/live/live.go

@@ -895,15 +895,19 @@ func (g *GrafanaLive) HandlePipelineEntitiesListHTTP(_ *models.ReqContext) respo
 		"subscribers": []configInfo{
 			{
 				Type:        pipeline.SubscriberTypeBuiltin,
-				Description: "list the fields that should be removed",
-				// Example:     pipeline.Bu{},
+				Description: "apply builtin feature subscribe logic",
 			},
 			{
 				Type:        pipeline.SubscriberTypeManagedStream,
-				Description: "list the fields that should be removed",
+				Description: "apply managed stream subscribe logic",
 			},
 			{
-				Type: pipeline.SubscriberTypeMultiple,
+				Type:        pipeline.SubscriberTypeMultiple,
+				Description: "apply multiple subscribers",
+			},
+			{
+				Type:        pipeline.SubscriberTypeAuthorizeRole,
+				Description: "authorize user role",
 			},
 		},
 		"outputs": []configInfo{
@@ -964,7 +968,7 @@ func (g *GrafanaLive) HandlePipelineEntitiesListHTTP(_ *models.ReqContext) respo
 			},
 			{
 				Type:        pipeline.ProcessorTypeMultiple,
-				Description: "apply multiplie processors",
+				Description: "apply multiple processors",
 				Example:     pipeline.MultipleProcessorConfig{},
 			},
 		},

pkg/services/live/pipeline/config.go

@@ -61,8 +61,9 @@ type MultipleSubscriberConfig struct {
 }
 
 type SubscriberConfig struct {
-	Type                     string                    `json:"type"`
-	MultipleSubscriberConfig *MultipleSubscriberConfig `json:"multiple,omitempty"`
+	Type                          string                         `json:"type"`
+	MultipleSubscriberConfig      *MultipleSubscriberConfig      `json:"multiple,omitempty"`
+	AuthorizeRoleSubscriberConfig *AuthorizeRoleSubscriberConfig `json:"authorizeRole,omitempty"`
 }
 
 type ChannelRuleSettings struct {
@@ -132,6 +133,11 @@ func (f *StorageRuleBuilder) extractSubscriber(config *SubscriberConfig) (Subscr
 		return NewBuiltinSubscriber(f.ChannelHandlerGetter), nil
 	case SubscriberTypeManagedStream:
 		return NewManagedStreamSubscriber(f.ManagedStream), nil
+	case SubscriberTypeAuthorizeRole:
+		if config.AuthorizeRoleSubscriberConfig == nil {
+			return nil, missingConfiguration
+		}
+		return NewAuthorizeRoleSubscriber(*config.AuthorizeRoleSubscriberConfig), nil
 	case SubscriberTypeMultiple:
 		if config.MultipleSubscriberConfig == nil {
 			return nil, missingConfiguration

pkg/services/live/pipeline/subscribe_authorize_role.go

@@ -0,0 +1,39 @@
+package pipeline
+
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/services/live/livecontext"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/models"
+)
+
+type AuthorizeRoleSubscriberConfig struct {
+	Role models.RoleType `json:"role,omitempty"`
+}
+
+type AuthorizeRoleSubscriber struct {
+	config AuthorizeRoleSubscriberConfig
+}
+
+func NewAuthorizeRoleSubscriber(config AuthorizeRoleSubscriberConfig) *AuthorizeRoleSubscriber {
+	return &AuthorizeRoleSubscriber{config: config}
+}
+
+const SubscriberTypeAuthorizeRole = "authorizeRole"
+
+func (s *AuthorizeRoleSubscriber) Type() string {
+	return SubscriberTypeAuthorizeRole
+}
+
+func (s *AuthorizeRoleSubscriber) Subscribe(ctx context.Context, _ Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
+	u, ok := livecontext.GetContextSignedUser(ctx)
+	if !ok {
+		return models.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil
+	}
+	if u.HasRole(s.config.Role) {
+		return models.SubscribeReply{}, backend.SubscribeStreamStatusOK, nil
+	}
+	return models.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil
+}

pkg/services/live/pipeline/subscribe_authorize_role_test.go

@@ -0,0 +1,34 @@
+package pipeline
+
+import (
+	"context"
+	"testing"
+
+	"github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/services/live/livecontext"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAuthorizeRoleSubscriber_Subscribe_PermissionDenied(t *testing.T) {
+	ctx := context.Background()
+	ctx = livecontext.SetContextSignedUser(ctx, &models.SignedInUser{OrgRole: models.ROLE_EDITOR})
+	s := NewAuthorizeRoleSubscriber(AuthorizeRoleSubscriberConfig{
+		Role: models.ROLE_ADMIN,
+	})
+	_, status, err := s.Subscribe(ctx, Vars{})
+	require.NoError(t, err)
+	require.Equal(t, backend.SubscribeStreamStatusPermissionDenied, status)
+}
+
+func TestAuthorizeRoleSubscriber_Subscribe_OK(t *testing.T) {
+	ctx := context.Background()
+	ctx = livecontext.SetContextSignedUser(ctx, &models.SignedInUser{OrgRole: models.ROLE_ADMIN})
+	s := NewAuthorizeRoleSubscriber(AuthorizeRoleSubscriberConfig{
+		Role: models.ROLE_ADMIN,
+	})
+	_, status, err := s.Subscribe(ctx, Vars{})
+	require.NoError(t, err)
+	require.Equal(t, backend.SubscribeStreamStatusOK, status)
+}