Live: more cleanup (#115144)

2025-12-16 08:22:19 +03:00
parent b8a5a516b5
commit 9fb61bd9f6
10 changed files with 95 additions and 151 deletions
@@ -112,17 +112,15 @@ func TestGetHomeDashboard(t *testing.T) {
 }

 func newTestLive(t *testing.T) *live.GrafanaLive {
-	features := featuremgmt.WithFeatures()
 	cfg := setting.NewCfg()
 	cfg.AppURL = "http://localhost:3000/"
-	gLive, err := live.ProvideService(nil, cfg,
+	gLive, err := live.ProvideService(cfg,
 		routing.NewRouteRegister(),
 		nil, nil, nil, nil,
-		nil,
 		&usagestats.UsageStatsMock{T: t},
-		features, acimpl.ProvideAccessControl(features),
-		&dashboards.FakeDashboardService{},
-		nil, nil)
+		featuremgmt.WithFeatures(),
+		&dashboards.FakeDashboardService{}, nil)
+
 	require.NoError(t, err)
 	return gLive
 }
@@ -349,6 +349,7 @@ var wireBasicSet = wire.NewSet(
 	dashboardservice.ProvideDashboardService,
 	dashboardservice.ProvideDashboardProvisioningService,
 	dashboardservice.ProvideDashboardPluginService,
+	dashboardservice.ProvideDashboardAccessService,
 	dashboardstore.ProvideDashboardStore,
 	folderimpl.ProvideService,
 	wire.Bind(new(folder.Service), new(*folderimpl.Service)),
@@ -44,6 +44,11 @@ type DashboardService interface {
 	GetDashboardsByLibraryPanelUID(ctx context.Context, libraryPanelUID string, orgID int64) ([]*DashboardRef, error)
 }

+type DashboardAccessService interface {
+	// The user as access to {VERB} the requested dashboard
+	HasDashboardAccess(ctx context.Context, user identity.Requester, verb string, namespace string, name string) (bool, error)
+}
+
 type PermissionsRegistrationService interface {
 	RegisterDashboardPermissions(service accesscontrol.DashboardPermissionsService)

@@ -5,9 +5,10 @@ package dashboards
 import (
 	context "context"

-	identity "github.com/grafana/grafana/pkg/apimachinery/identity"
 	mock "github.com/stretchr/testify/mock"

+	identity "github.com/grafana/grafana/pkg/apimachinery/identity"
+
 	model "github.com/grafana/grafana/pkg/services/search/model"

 	unstructured "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -529,6 +530,11 @@ func (_m *FakeDashboardService) ValidateDashboardRefreshInterval(minRefreshInter
 	return r0
 }

+// CanViewDashboard uses the access control service to check if the requested user can see a dashboard
+func (_m *FakeDashboardService) HasDashboardAccess(ctx context.Context, user identity.Requester, verb string, namespace string, name string) (bool, error) {
+	return true, nil
+}
+
 // NewFakeDashboardService creates a new instance of FakeDashboardService. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewFakeDashboardService(t interface {
@@ -67,6 +67,7 @@ var (
 	_ dashboards.DashboardService             = (*DashboardServiceImpl)(nil)
 	_ dashboards.DashboardProvisioningService = (*DashboardServiceImpl)(nil)
 	_ dashboards.PluginService                = (*DashboardServiceImpl)(nil)
+	_ dashboards.DashboardAccessService       = (*DashboardServiceImpl)(nil)

 	daysInTrash = 24 * 30 * time.Hour
 	tracer      = otel.Tracer("github.com/grafana/grafana/pkg/services/dashboards/service")
@@ -100,6 +101,38 @@ type DashboardServiceImpl struct {
 	dashboardPermissionsReady chan struct{}
 }

+// CanViewDashboard uses the access control service to check if the requested user can see a dashboard
+func (dr *DashboardServiceImpl) HasDashboardAccess(ctx context.Context, user identity.Requester, verb string, namespace string, name string) (bool, error) {
+	ns, err := claims.ParseNamespace(namespace)
+	if err != nil {
+		return false, err
+	}
+	dash, err := dr.GetDashboard(ctx, &dashboards.GetDashboardQuery{
+		UID:   name,
+		OrgID: ns.OrgID,
+	})
+	if err != nil || dash == nil {
+		return false, nil
+	}
+	var action string
+	switch verb {
+	case utils.VerbGet:
+		action = dashboards.ActionDashboardsRead
+	case utils.VerbUpdate:
+		action = dashboards.ActionDashboardsWrite
+	default:
+		return false, fmt.Errorf("unsupported verb")
+	}
+
+	evaluator := accesscontrol.EvalPermission(action,
+		dashboards.ScopeDashboardsProvider.GetResourceScopeUID(name))
+	canView, err := dr.ac.Evaluate(ctx, user, evaluator)
+	if err != nil || !canView {
+		return false, nil
+	}
+	return true, nil
+}
+
 func (dr *DashboardServiceImpl) startK8sDeletedDashboardsCleanupJob(ctx context.Context) chan struct{} {
 	done := make(chan struct{})
 	go func() {
@@ -23,3 +23,9 @@ func ProvideDashboardPluginService(
 ) dashboards.PluginService {
 	return orig
 }
+
+func ProvideDashboardAccessService(
+	features featuremgmt.FeatureToggles, orig *DashboardServiceImpl,
+) dashboards.DashboardAccessService {
+	return orig
+}
@@ -6,10 +6,11 @@ import (
 	"fmt"
 	"strings"

+	"github.com/grafana/authlib/types"
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
 	"github.com/grafana/grafana/pkg/cmd/grafana-cli/logger"
-	"github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/dashboards"
 	"github.com/grafana/grafana/pkg/services/live/model"
 )
@@ -32,10 +33,9 @@ type dashboardEvent struct {

 // DashboardHandler manages all the `grafana/dashboard/*` channels
 type DashboardHandler struct {
-	Publisher        model.ChannelPublisher
-	ClientCount      model.ChannelClientCount
-	DashboardService dashboards.DashboardService
-	AccessControl    accesscontrol.AccessControl
+	Publisher     model.ChannelPublisher
+	ClientCount   model.ChannelClientCount
+	AccessControl dashboards.DashboardAccessService
 }

 // GetHandlerForPath called on init
@@ -49,23 +49,15 @@ func (h *DashboardHandler) OnSubscribe(ctx context.Context, user identity.Reques

 	// make sure can view this dashboard
 	if len(parts) == 2 && parts[0] == "uid" {
-		query := dashboards.GetDashboardQuery{UID: parts[1], OrgID: user.GetOrgID()}
-		_, err := h.DashboardService.GetDashboard(ctx, &query)
-		if err != nil {
-			logger.Error("Error getting dashboard", "query", query, "error", err)
-			return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, nil
+		ns := types.OrgNamespaceFormatter(user.GetOrgID())
+		ok, err := h.AccessControl.HasDashboardAccess(ctx, user, utils.VerbGet, ns, parts[1])
+		if ok && err == nil {
+			return model.SubscribeReply{
+				Presence:  true,
+				JoinLeave: true,
+			}, backend.SubscribeStreamStatusOK, nil
 		}
-
-		evaluator := accesscontrol.EvalPermission(dashboards.ActionDashboardsRead, dashboards.ScopeDashboardsProvider.GetResourceScopeUID(parts[1]))
-		canView, err := h.AccessControl.Evaluate(ctx, user, evaluator)
-		if err != nil || !canView {
-			return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, err
-		}
-
-		return model.SubscribeReply{
-			Presence:  true,
-			JoinLeave: true,
-		}, backend.SubscribeStreamStatusOK, nil
+		return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, err
 	}

 	// Unknown path
@@ -88,29 +80,16 @@ func (h *DashboardHandler) OnPublish(ctx context.Context, requester identity.Req
 			// just ignore the event
 			return model.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("ignore???")
 		}
-		query := dashboards.GetDashboardQuery{UID: parts[1], OrgID: requester.GetOrgID()}
-		_, err = h.DashboardService.GetDashboard(ctx, &query)
-		if err != nil {
-			logger.Error("Unknown dashboard", "query", query)
-			return model.PublishReply{}, backend.PublishStreamStatusNotFound, nil
-		}

-		evaluator := accesscontrol.EvalPermission(dashboards.ActionDashboardsWrite, dashboards.ScopeDashboardsProvider.GetResourceScopeUID(parts[1]))
-		canEdit, err := h.AccessControl.Evaluate(ctx, requester, evaluator)
-		if err != nil {
-			return model.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("internal error")
+		ns := types.OrgNamespaceFormatter(requester.GetOrgID())
+		ok, err := h.AccessControl.HasDashboardAccess(ctx, requester, utils.VerbUpdate, ns, parts[1])
+		if ok && err == nil {
+			msg, err := json.Marshal(event)
+			if err != nil {
+				return model.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("internal error")
+			}
+			return model.PublishReply{Data: msg}, backend.PublishStreamStatusOK, nil
 		}
-
-		// Ignore edit events if the user can not edit
-		if !canEdit {
-			return model.PublishReply{}, backend.PublishStreamStatusNotFound, nil // NOOP
-		}
-
-		msg, err := json.Marshal(event)
-		if err != nil {
-			return model.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("internal error")
-		}
-		return model.PublishReply{Data: msg}, backend.PublishStreamStatusOK, nil
 	}

 	return model.PublishReply{}, backend.PublishStreamStatusNotFound, nil
@@ -27,13 +27,11 @@ import (
 	"github.com/grafana/grafana/pkg/api/response"
 	"github.com/grafana/grafana/pkg/api/routing"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
-	"github.com/grafana/grafana/pkg/infra/localcache"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/usagestats"
 	"github.com/grafana/grafana/pkg/middleware"
 	"github.com/grafana/grafana/pkg/middleware/requestmeta"
 	"github.com/grafana/grafana/pkg/plugins"
-	"github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/apiserver"
 	contextmodel "github.com/grafana/grafana/pkg/services/contexthandler/model"
 	"github.com/grafana/grafana/pkg/services/dashboards"
@@ -52,7 +50,6 @@ import (
 	"github.com/grafana/grafana/pkg/services/org"
 	"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
 	"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
-	"github.com/grafana/grafana/pkg/services/secrets"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/util"
 	"github.com/grafana/grafana/pkg/web"
@@ -72,28 +69,23 @@ type CoreGrafanaScope struct {
 	Dashboards DashboardActivityChannel
 }

-func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, routeRegister routing.RouteRegister,
-	pluginStore pluginstore.Store, pluginClient plugins.Client, cacheService *localcache.CacheService,
-	dataSourceCache datasources.CacheService, secretsService secrets.Service,
+func ProvideService(cfg *setting.Cfg, routeRegister routing.RouteRegister, plugCtxProvider *plugincontext.Provider,
+	pluginStore pluginstore.Store, pluginClient plugins.Client, dataSourceCache datasources.CacheService,
 	usageStatsService usagestats.Service, toggles featuremgmt.FeatureToggles,
-	accessControl accesscontrol.AccessControl, dashboardService dashboards.DashboardService,
-	orgService org.Service, configProvider apiserver.RestConfigProvider) (*GrafanaLive, error) {
+	dashboardService dashboards.DashboardAccessService,
+	configProvider apiserver.RestConfigProvider) (*GrafanaLive, error) {
 	g := &GrafanaLive{
 		Cfg:                   cfg,
 		Features:              toggles,
 		PluginContextProvider: plugCtxProvider,
-		RouteRegister:         routeRegister,
 		pluginStore:           pluginStore,
 		pluginClient:          pluginClient,
-		CacheService:          cacheService,
 		DataSourceCache:       dataSourceCache,
-		SecretsService:        secretsService,
 		channels:              make(map[string]model.ChannelHandler),
 		GrafanaScope: CoreGrafanaScope{
 			Features: make(map[string]model.ChannelHandlerFactory),
 		},
 		usageStatsService: usageStatsService,
-		orgService:        orgService,
 		keyPrefix:         "gf_live",
 	}

@@ -176,19 +168,13 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r

 	// Initialize the main features
 	dash := &features.DashboardHandler{
-		Publisher:        g.Publish,
-		ClientCount:      g.ClientCount,
-		DashboardService: dashboardService,
-		AccessControl:    accessControl,
+		Publisher:     g.Publish,
+		ClientCount:   g.ClientCount,
+		AccessControl: dashboardService,
 	}
 	g.GrafanaScope.Dashboards = dash
 	g.GrafanaScope.Features["dashboard"] = dash
-
-	// Testing watch with just the provisioning support -- this will be removed when it is well validated
-	//nolint:staticcheck // not yet migrated to OpenFeature
-	if toggles.IsEnabledGlobally(featuremgmt.FlagProvisioning) {
-		g.GrafanaScope.Features["watch"] = features.NewWatchRunner(g.Publish, configProvider)
-	}
+	g.GrafanaScope.Features["watch"] = features.NewWatchRunner(g.Publish, configProvider)

 	g.surveyCaller = survey.NewCaller(managedStreamRunner, node)
 	err = g.surveyCaller.SetupHandlers()
@@ -398,11 +384,11 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
 		pushPipelineWSHandler.ServeHTTP(ctx.Resp, r)
 	}

-	g.RouteRegister.Group("/api/live", func(group routing.RouteRegister) {
+	routeRegister.Group("/api/live", func(group routing.RouteRegister) {
 		group.Get("/ws", g.websocketHandler)
 	}, middleware.ReqSignedIn, requestmeta.SetSLOGroup(requestmeta.SLOGroupNone))

-	g.RouteRegister.Group("/api/live", func(group routing.RouteRegister) {
+	routeRegister.Group("/api/live", func(group routing.RouteRegister) {
 		group.Get("/push/:streamId", g.pushWebsocketHandler)
 		group.Get("/pipeline/push/*", g.pushPipelineWebsocketHandler)
 	}, middleware.ReqOrgAdmin, requestmeta.SetSLOGroup(requestmeta.SLOGroupNone))
@@ -461,13 +447,9 @@ type GrafanaLive struct {
 	PluginContextProvider *plugincontext.Provider
 	Cfg                   *setting.Cfg
 	Features              featuremgmt.FeatureToggles
-	RouteRegister         routing.RouteRegister
-	CacheService          *localcache.CacheService
 	DataSourceCache       datasources.CacheService
-	SecretsService        secrets.Service
 	pluginStore           pluginstore.Store
 	pluginClient          plugins.Client
-	orgService            org.Service

 	keyPrefix string // HA prefix for grafana cloud (since the org is always 1)

@@ -1356,71 +1338,6 @@ func (g *GrafanaLive) HandleWriteConfigsPostHTTP(c *contextmodel.ReqContext) res
 	})
 }

-// HandleWriteConfigsPutHTTP ...
-func (g *GrafanaLive) HandleWriteConfigsPutHTTP(c *contextmodel.ReqContext) response.Response {
-	body, err := io.ReadAll(c.Req.Body)
-	if err != nil {
-		return response.Error(http.StatusInternalServerError, "Error reading body", err)
-	}
-	var cmd pipeline.WriteConfigUpdateCmd
-	err = json.Unmarshal(body, &cmd)
-	if err != nil {
-		return response.Error(http.StatusBadRequest, "Error decoding write config update command", err)
-	}
-	if cmd.UID == "" {
-		return response.Error(http.StatusBadRequest, "UID required", nil)
-	}
-	existingBackend, ok, err := g.pipelineStorage.GetWriteConfig(c.Req.Context(), c.GetOrgID(), pipeline.WriteConfigGetCmd{
-		UID: cmd.UID,
-	})
-	if err != nil {
-		return response.Error(http.StatusInternalServerError, "Failed to get write config", err)
-	}
-	if ok {
-		if cmd.SecureSettings == nil {
-			cmd.SecureSettings = map[string]string{}
-		}
-		secureJSONData, err := g.SecretsService.DecryptJsonData(c.Req.Context(), existingBackend.SecureSettings)
-		if err != nil {
-			logger.Error("Error decrypting secure settings", "error", err)
-			return response.Error(http.StatusInternalServerError, "Error decrypting secure settings", err)
-		}
-		for k, v := range secureJSONData {
-			if _, ok := cmd.SecureSettings[k]; !ok {
-				cmd.SecureSettings[k] = v
-			}
-		}
-	}
-	result, err := g.pipelineStorage.UpdateWriteConfig(c.Req.Context(), c.GetOrgID(), cmd)
-	if err != nil {
-		return response.Error(http.StatusInternalServerError, "Failed to update write config", err)
-	}
-	return response.JSON(http.StatusOK, util.DynMap{
-		"writeConfig": pipeline.WriteConfigToDto(result),
-	})
-}
-
-// HandleWriteConfigsDeleteHTTP ...
-func (g *GrafanaLive) HandleWriteConfigsDeleteHTTP(c *contextmodel.ReqContext) response.Response {
-	body, err := io.ReadAll(c.Req.Body)
-	if err != nil {
-		return response.Error(http.StatusInternalServerError, "Error reading body", err)
-	}
-	var cmd pipeline.WriteConfigDeleteCmd
-	err = json.Unmarshal(body, &cmd)
-	if err != nil {
-		return response.Error(http.StatusBadRequest, "Error decoding write config delete command", err)
-	}
-	if cmd.UID == "" {
-		return response.Error(http.StatusBadRequest, "UID required", nil)
-	}
-	err = g.pipelineStorage.DeleteWriteConfig(c.Req.Context(), c.GetOrgID(), cmd)
-	if err != nil {
-		return response.Error(http.StatusInternalServerError, "Failed to delete write config", err)
-	}
-	return response.JSON(http.StatusOK, util.DynMap{})
-}
-
 // Write to the standard log15 logger
 func handleLog(msg centrifuge.LogEntry) {
 	arr := make([]interface{}, 0)
@@ -19,7 +19,6 @@ import (
 	"github.com/grafana/grafana/pkg/api/routing"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
 	"github.com/grafana/grafana/pkg/infra/usagestats"
-	"github.com/grafana/grafana/pkg/services/accesscontrol/acimpl"
 	"github.com/grafana/grafana/pkg/services/dashboards"
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/setting"
@@ -340,16 +339,14 @@ func setupLiveService(cfg *setting.Cfg, t *testing.T) (*GrafanaLive, error) {
 		cfg = setting.NewCfg()
 	}

-	return ProvideService(nil,
-		cfg,
+	return ProvideService(cfg,
 		routing.NewRouteRegister(),
-		nil, nil, nil, nil,
+		nil, nil, nil,
 		nil,
 		&usagestats.UsageStatsMock{T: t},
 		featuremgmt.WithFeatures(),
-		acimpl.ProvideAccessControl(featuremgmt.WithFeatures()),
 		&dashboards.FakeDashboardService{},
-		nil, nil)
+		nil)
 }

 type dummyTransport struct {