From 55aae797441d2beadc911e390f14c04bb3f101da Mon Sep 17 00:00:00 2001 From: Alexander Emelin Date: Tue, 27 Sep 2022 00:25:34 +0400 Subject: [PATCH] GRPC Server: Add gRPC server service (#47849) Co-authored-by: Todd Treece Co-authored-by: Ryan McKinley --- .../src/types/featureToggles.gen.ts | 1 + .../backgroundsvcs/background_services.go | 4 + pkg/server/wire.go | 5 +- pkg/services/featuremgmt/registry.go | 5 + pkg/services/featuremgmt/toggles_gen.go | 4 + pkg/services/grpcserver/auth.go | 129 ++++++++++++++++++ pkg/services/grpcserver/auth_test.go | 108 +++++++++++++++ pkg/services/grpcserver/health.go | 37 +++++ pkg/services/grpcserver/service.go | 98 +++++++++++++ pkg/setting/setting.go | 72 ++++++++++ 10 files changed, 462 insertions(+), 1 deletion(-) create mode 100644 pkg/services/grpcserver/auth.go create mode 100644 pkg/services/grpcserver/auth_test.go create mode 100644 pkg/services/grpcserver/health.go create mode 100644 pkg/services/grpcserver/service.go diff --git a/packages/grafana-data/src/types/featureToggles.gen.ts b/packages/grafana-data/src/types/featureToggles.gen.ts index 3a8f8306097..16ea1a9b613 100644 --- a/packages/grafana-data/src/types/featureToggles.gen.ts +++ b/packages/grafana-data/src/types/featureToggles.gen.ts @@ -64,6 +64,7 @@ export interface FeatureToggles { internationalization?: boolean; topnav?: boolean; customBranding?: boolean; + grpcServer?: boolean; traceqlEditor?: boolean; redshiftAsyncQueryDataSupport?: boolean; athenaAsyncQueryDataSupport?: boolean; diff --git a/pkg/server/backgroundsvcs/background_services.go b/pkg/server/backgroundsvcs/background_services.go index 55a7ea3da3d..5862715661e 100644 --- a/pkg/server/backgroundsvcs/background_services.go +++ b/pkg/server/backgroundsvcs/background_services.go @@ -13,6 +13,7 @@ import ( "github.com/grafana/grafana/pkg/services/alerting" "github.com/grafana/grafana/pkg/services/cleanup" "github.com/grafana/grafana/pkg/services/dashboardsnapshots" + "github.com/grafana/grafana/pkg/services/grpcserver" "github.com/grafana/grafana/pkg/services/guardian" "github.com/grafana/grafana/pkg/services/live" "github.com/grafana/grafana/pkg/services/live/pushhttp" @@ -43,11 +44,13 @@ func ProvideBackgroundServiceRegistry( secretsService *secretsManager.SecretsService, remoteCache *remotecache.RemoteCache, thumbnailsService thumbs.Service, StorageService store.StorageService, searchService searchV2.SearchService, entityEventsService store.EntityEventsService, saService *samanager.ServiceAccountsService, authInfoService *authinfoservice.Implementation, + grpcServerProvider grpcserver.Provider, secretMigrationProvider secretsMigrations.SecretMigrationProvider, // Need to make sure these are initialized, is there a better place to put them? _ dashboardsnapshots.Service, _ *alerting.AlertNotificationService, _ serviceaccounts.Service, _ *guardian.Provider, _ *plugindashboardsservice.DashboardUpdater, _ *sanitizer.Provider, + _ *grpcserver.HealthService, ) *BackgroundServiceRegistry { return NewBackgroundServiceRegistry( httpServer, @@ -72,6 +75,7 @@ func ProvideBackgroundServiceRegistry( thumbnailsService, searchService, entityEventsService, + grpcServerProvider, saService, authInfoService, processManager, diff --git a/pkg/server/wire.go b/pkg/server/wire.go index 5811dee4419..a982bcaae81 100644 --- a/pkg/server/wire.go +++ b/pkg/server/wire.go @@ -75,6 +75,7 @@ import ( encryptionservice "github.com/grafana/grafana/pkg/services/encryption/service" "github.com/grafana/grafana/pkg/services/export" "github.com/grafana/grafana/pkg/services/featuremgmt" + "github.com/grafana/grafana/pkg/services/grpcserver" "github.com/grafana/grafana/pkg/services/guardian" "github.com/grafana/grafana/pkg/services/hooks" "github.com/grafana/grafana/pkg/services/libraryelements" @@ -342,16 +343,18 @@ var wireBasicSet = wire.NewSet( publicdashboardsApi.ProvideApi, userimpl.ProvideService, orgimpl.ProvideService, + grpcserver.ProvideService, + grpcserver.ProvideHealthService, teamimpl.ProvideService, tempuserimpl.ProvideService, dashboardthumbsimpl.ProvideService, loginattemptimpl.ProvideService, + userauthimpl.ProvideService, secretsMigrations.ProvideDataSourceMigrationService, secretsMigrations.ProvideMigrateToPluginService, secretsMigrations.ProvideMigrateFromPluginService, secretsMigrations.ProvideSecretMigrationProvider, wire.Bind(new(secretsMigrations.SecretMigrationProvider), new(*secretsMigrations.SecretMigrationProviderImpl)), - userauthimpl.ProvideService, acimpl.ProvideAccessControl, navtreeimpl.ProvideService, wire.Bind(new(accesscontrol.AccessControl), new(*acimpl.AccessControl)), diff --git a/pkg/services/featuremgmt/registry.go b/pkg/services/featuremgmt/registry.go index a475957440c..321204d99cf 100644 --- a/pkg/services/featuremgmt/registry.go +++ b/pkg/services/featuremgmt/registry.go @@ -268,6 +268,11 @@ var ( State: FeatureStateAlpha, }, { + Name: "grpcServer", + Description: "Run GRPC server", + State: FeatureStateAlpha, + RequiresDevMode: true, + }, { Name: "traceqlEditor", Description: "Show the TraceQL editor in the explore page", State: FeatureStateAlpha, diff --git a/pkg/services/featuremgmt/toggles_gen.go b/pkg/services/featuremgmt/toggles_gen.go index c98d7cdee2f..c280a2a41a1 100644 --- a/pkg/services/featuremgmt/toggles_gen.go +++ b/pkg/services/featuremgmt/toggles_gen.go @@ -199,6 +199,10 @@ const ( // Replaces whitelabeling with the new custom branding feature FlagCustomBranding = "customBranding" + // FlagGrpcServer + // Run GRPC server + FlagGrpcServer = "grpcServer" + // FlagTraceqlEditor // Show the TraceQL editor in the explore page FlagTraceqlEditor = "traceqlEditor" diff --git a/pkg/services/grpcserver/auth.go b/pkg/services/grpcserver/auth.go new file mode 100644 index 00000000000..7802c810812 --- /dev/null +++ b/pkg/services/grpcserver/auth.go @@ -0,0 +1,129 @@ +package grpcserver + +import ( + "context" + "fmt" + "strings" + + "github.com/grafana/grafana/pkg/cmd/grafana-cli/logger" + apikeygenprefix "github.com/grafana/grafana/pkg/components/apikeygenprefixed" + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/services/apikey" + "github.com/grafana/grafana/pkg/services/org" + "github.com/grafana/grafana/pkg/services/user" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +// authenticator can authenticate GRPC requests. +type authenticator struct { + logger log.Logger + APIKey apikey.Service + UserService user.Service +} + +func newAuthenticator(apiKey apikey.Service, userService user.Service) *authenticator { + return &authenticator{ + logger: log.New("grpc-server-authenticator"), + APIKey: apiKey, + UserService: userService, + } +} + +// Authenticate checks that a token exists and is valid, and then removes the token from the +// authorization header in the context. +func (a *authenticator) authenticate(ctx context.Context) (context.Context, error) { + return a.tokenAuth(ctx) +} + +const tokenPrefix = "Bearer " + +func (a *authenticator) tokenAuth(ctx context.Context) (context.Context, error) { + auth, err := extractAuthorization(ctx) + if err != nil { + return ctx, err + } + + if !strings.HasPrefix(auth, tokenPrefix) { + return ctx, status.Error(codes.Unauthenticated, `missing "Bearer " prefix in "authorization" value`) + } + + token := strings.TrimPrefix(auth, tokenPrefix) + if token == "" { + return ctx, status.Error(codes.Unauthenticated, "token required") + } + + newCtx := purgeHeader(ctx, "authorization") + + err = a.validateToken(ctx, token) + if err != nil { + logger.Warn("request with invalid token", "error", err, "token", token) + return ctx, status.Error(codes.Unauthenticated, "invalid token") + } + return newCtx, nil +} + +func (a *authenticator) validateToken(ctx context.Context, keyString string) error { + decoded, err := apikeygenprefix.Decode(keyString) + if err != nil { + return err + } + + hash, err := decoded.Hash() + if err != nil { + return err + } + + apikey, err := a.APIKey.GetAPIKeyByHash(ctx, hash) + if err != nil { + return err + } + + if apikey == nil || apikey.ServiceAccountId == nil { + return status.Error(codes.Unauthenticated, "api key does not have a service account") + } + + querySignedInUser := user.GetSignedInUserQuery{UserID: *apikey.ServiceAccountId, OrgID: apikey.OrgId} + signedInUser, err := a.UserService.GetSignedInUserWithCacheCtx(ctx, &querySignedInUser) + if err != nil { + return err + } + + if !signedInUser.HasRole(org.RoleAdmin) { + return fmt.Errorf("api key does not have admin role") + } + + // disabled service accounts are not allowed to access the API + if signedInUser.IsDisabled { + return fmt.Errorf("service account is disabled") + } + + return nil +} + +func extractAuthorization(ctx context.Context) (string, error) { + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + return "", status.Error(codes.Unauthenticated, "no headers in request") + } + + authHeaders, ok := md["authorization"] + if !ok { + return "", status.Error(codes.Unauthenticated, `no "authorization" header in request`) + } + + if len(authHeaders) != 1 { + return "", status.Error(codes.Unauthenticated, `malformed "authorization" header: one value required`) + } + + return authHeaders[0], nil +} + +func purgeHeader(ctx context.Context, header string) context.Context { + md, _ := metadata.FromIncomingContext(ctx) + mdCopy := md.Copy() + mdCopy[header] = nil + return metadata.NewIncomingContext(ctx, mdCopy) +} diff --git a/pkg/services/grpcserver/auth_test.go b/pkg/services/grpcserver/auth_test.go new file mode 100644 index 00000000000..67de47f56ea --- /dev/null +++ b/pkg/services/grpcserver/auth_test.go @@ -0,0 +1,108 @@ +package grpcserver + +import ( + "context" + "testing" + + apikeygenprefix "github.com/grafana/grafana/pkg/components/apikeygenprefixed" + "github.com/grafana/grafana/pkg/services/apikey" + "github.com/grafana/grafana/pkg/services/org" + "github.com/grafana/grafana/pkg/services/user" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/metadata" +) + +func TestAuthenticator_Authenticate(t *testing.T) { + serviceAccountId := int64(1) + t.Run("accepts service api key with admin role", func(t *testing.T) { + s := newFakeAPIKey(&apikey.APIKey{ + Id: 1, + OrgId: 1, + Key: "admin-api-key", + Name: "Admin API Key", + ServiceAccountId: &serviceAccountId, + }, nil) + a := newAuthenticator(s, &fakeUserService{OrgRole: org.RoleAdmin}) + ctx, err := setupContext() + require.NoError(t, err) + _, err = a.authenticate(ctx) + require.NoError(t, err) + }) + + t.Run("rejects non-admin role", func(t *testing.T) { + s := newFakeAPIKey(&apikey.APIKey{ + Id: 1, + OrgId: 1, + Key: "admin-api-key", + Name: "Admin API Key", + ServiceAccountId: &serviceAccountId, + }, nil) + a := newAuthenticator(s, &fakeUserService{OrgRole: org.RoleEditor}) + ctx, err := setupContext() + require.NoError(t, err) + _, err = a.authenticate(ctx) + require.NotNil(t, err) + }) + + t.Run("removes auth header from context", func(t *testing.T) { + s := newFakeAPIKey(&apikey.APIKey{ + Id: 1, + OrgId: 1, + Key: "admin-api-key", + Name: "Admin API Key", + ServiceAccountId: &serviceAccountId, + }, nil) + a := newAuthenticator(s, &fakeUserService{OrgRole: org.RoleAdmin}) + ctx, err := setupContext() + require.NoError(t, err) + md, ok := metadata.FromIncomingContext(ctx) + require.True(t, ok) + require.NotEmpty(t, md["authorization"]) + ctx, err = a.authenticate(ctx) + require.NoError(t, err) + md, ok = metadata.FromIncomingContext(ctx) + require.True(t, ok) + require.Empty(t, md["authorization"]) + }) +} + +type fakeAPIKey struct { + apikey.Service + key *apikey.APIKey + err error +} + +func newFakeAPIKey(key *apikey.APIKey, err error) *fakeAPIKey { + return &fakeAPIKey{ + key: key, + err: err, + } +} + +func (f *fakeAPIKey) GetAPIKeyByHash(ctx context.Context, hash string) (*apikey.APIKey, error) { + return f.key, f.err +} + +type fakeUserService struct { + user.Service + OrgRole org.RoleType +} + +func (f *fakeUserService) GetSignedInUserWithCacheCtx(ctx context.Context, query *user.GetSignedInUserQuery) (*user.SignedInUser, error) { + return &user.SignedInUser{ + UserID: 1, + OrgID: 1, + OrgRole: f.OrgRole, + }, nil +} + +func setupContext() (context.Context, error) { + ctx := context.Background() + key, err := apikeygenprefix.New("sa") + if err != nil { + return ctx, err + } + md := metadata.New(map[string]string{}) + md["authorization"] = []string{"Bearer " + key.ClientSecret} + return metadata.NewIncomingContext(ctx, md), nil +} diff --git a/pkg/services/grpcserver/health.go b/pkg/services/grpcserver/health.go new file mode 100644 index 00000000000..c00b3e3cd48 --- /dev/null +++ b/pkg/services/grpcserver/health.go @@ -0,0 +1,37 @@ +package grpcserver + +import ( + "context" + + "github.com/grafana/grafana/pkg/setting" + + "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" +) + +// HealthService implements GRPC Health Checking Protocol: +// https://github.com/grpc/grpc/blob/master/doc/health-checking.md +// It also demonstrates how to override authentication for a service – in this +// case we are disabling any auth in AuthFuncOverride. +type HealthService struct { + cfg *setting.Cfg + healthServer *healthServer +} + +type healthServer struct { + *health.Server +} + +// AuthFuncOverride for no auth for health service. +func (s *healthServer) AuthFuncOverride(ctx context.Context, _ string) (context.Context, error) { + return ctx, nil +} + +func ProvideHealthService(cfg *setting.Cfg, grpcServerProvider Provider) (*HealthService, error) { + hs := &healthServer{health.NewServer()} + grpc_health_v1.RegisterHealthServer(grpcServerProvider.GetServer(), hs) + return &HealthService{ + cfg: cfg, + healthServer: hs, + }, nil +} diff --git a/pkg/services/grpcserver/service.go b/pkg/services/grpcserver/service.go new file mode 100644 index 00000000000..e207bd3fa4f --- /dev/null +++ b/pkg/services/grpcserver/service.go @@ -0,0 +1,98 @@ +package grpcserver + +import ( + "context" + "fmt" + "net" + + "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/registry" + "github.com/grafana/grafana/pkg/services/apikey" + "github.com/grafana/grafana/pkg/services/featuremgmt" + "github.com/grafana/grafana/pkg/services/user" + "github.com/grafana/grafana/pkg/setting" + grpcAuth "github.com/grpc-ecosystem/go-grpc-middleware/auth" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/reflection" +) + +type Provider interface { + registry.BackgroundService + GetServer() *grpc.Server +} + +type GPRCServerService struct { + cfg *setting.Cfg + logger log.Logger + server *grpc.Server +} + +func ProvideService(cfg *setting.Cfg, apiKey apikey.Service, userService user.Service) (Provider, error) { + s := &GPRCServerService{ + cfg: cfg, + logger: log.New("grpc-server"), + } + + var opts []grpc.ServerOption + + // Default auth is admin token check, but this can be overridden by + // services which implement ServiceAuthFuncOverride interface. + // See https://github.com/grpc-ecosystem/go-grpc-middleware/blob/master/auth/auth.go#L30. + authenticator := newAuthenticator(apiKey, userService) + opts = append(opts, []grpc.ServerOption{ + grpc.StreamInterceptor(grpcAuth.StreamServerInterceptor(authenticator.authenticate)), + grpc.UnaryInterceptor(grpcAuth.UnaryServerInterceptor(authenticator.authenticate)), + }...) + + if s.cfg.GRPCServerTLSConfig != nil { + opts = append(opts, grpc.Creds(credentials.NewTLS(cfg.GRPCServerTLSConfig))) + } + + grpcServer := grpc.NewServer(opts...) + reflection.Register(grpcServer) + s.server = grpcServer + return s, nil +} + +func (s *GPRCServerService) Run(ctx context.Context) error { + s.logger.Info("Running GRPC server", "address", s.cfg.GRPCServerAddress, "network", s.cfg.GRPCServerNetwork, "tls", s.cfg.GRPCServerTLSConfig != nil) + + listener, err := net.Listen(s.cfg.GRPCServerNetwork, s.cfg.GRPCServerAddress) + if err != nil { + return fmt.Errorf("GRPC server: failed to listen: %w", err) + } + + serveErr := make(chan error, 1) + go func() { + s.logger.Info("GRPC server: starting") + err := s.server.Serve(listener) + if err != nil { + backend.Logger.Error("GRPC server: failed to serve", "err", err) + serveErr <- err + } + }() + + select { + case err := <-serveErr: + backend.Logger.Error("GRPC server: failed to serve", "err", err) + return err + case <-ctx.Done(): + } + s.logger.Warn("GRPC server: shutting down") + s.server.Stop() + return ctx.Err() +} + +func (s *GPRCServerService) IsDisabled() bool { + if s.cfg == nil { + return true + } + return !s.cfg.IsFeatureToggleEnabled(featuremgmt.FlagGrpcServer) +} + +func (s *GPRCServerService) GetServer() *grpc.Server { + return s.server +} diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index d447cd98024..d8b1d36216d 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -5,9 +5,11 @@ package setting import ( "bytes" + "crypto/tls" "encoding/json" "errors" "fmt" + "io/fs" "net/http" "net/url" "os" @@ -463,6 +465,10 @@ type Cfg struct { RBACPermissionCache bool // Enable Permission validation during role creation and provisioning RBACPermissionValidationEnabled bool + // GRPC Server. + GRPCServerNetwork string + GRPCServerAddress string + GRPCServerTLSConfig *tls.Config } type CommandLineArgs struct { @@ -946,6 +952,10 @@ func (cfg *Cfg) Load(args CommandLineArgs) error { return err } + if err := readGRPCServerSettings(cfg, iniFile); err != nil { + return err + } + // read dashboard settings dashboards := iniFile.Section("dashboards") DashboardVersionsToKeep = dashboards.Key("versions_to_keep").MustInt(20) @@ -1492,6 +1502,68 @@ func readAlertingSettings(iniFile *ini.File) error { return nil } +func readGRPCServerSettings(cfg *Cfg, iniFile *ini.File) error { + server := iniFile.Section("grpc_server") + errPrefix := "grpc_server:" + useTLS := server.Key("use_tls").MustBool(false) + certFile := server.Key("cert_file").String() + keyFile := server.Key("cert_key").String() + if useTLS { + serverCert, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return fmt.Errorf("%s error loading X509 key pair: %w", errPrefix, err) + } + cfg.GRPCServerTLSConfig = &tls.Config{ + Certificates: []tls.Certificate{serverCert}, + ClientAuth: tls.NoClientCert, + } + } + + cfg.GRPCServerNetwork = valueAsString(server, "network", "tcp") + cfg.GRPCServerAddress = valueAsString(server, "address", "") + switch cfg.GRPCServerNetwork { + case "unix": + if cfg.GRPCServerAddress != "" { + // Explicitly provided path for unix domain socket. + if stat, err := os.Stat(cfg.GRPCServerAddress); os.IsNotExist(err) { + // File does not exist - nice, nothing to do. + } else if err != nil { + return fmt.Errorf("%s error getting stat for a file: %s", errPrefix, cfg.GRPCServerAddress) + } else { + if stat.Mode()&fs.ModeSocket == 0 { + return fmt.Errorf("%s file %s already exists and is not a unix domain socket", errPrefix, cfg.GRPCServerAddress) + } + // Unix domain socket file, should be safe to remove. + err := os.Remove(cfg.GRPCServerAddress) + if err != nil { + return fmt.Errorf("%s can't remove unix socket file: %s", errPrefix, cfg.GRPCServerAddress) + } + } + } else { + // Use temporary file path for a unix domain socket. + tf, err := os.CreateTemp("", "gf_grpc_server_api") + if err != nil { + return fmt.Errorf("%s error creating tmp file: %v", errPrefix, err) + } + unixPath := tf.Name() + if err := tf.Close(); err != nil { + return fmt.Errorf("%s error closing tmp file: %v", errPrefix, err) + } + if err := os.Remove(unixPath); err != nil { + return fmt.Errorf("%s error removing tmp file: %v", errPrefix, err) + } + cfg.GRPCServerAddress = unixPath + } + case "tcp": + if cfg.GRPCServerAddress == "" { + cfg.GRPCServerAddress = "127.0.0.1:10000" + } + default: + return fmt.Errorf("%s unsupported network %s", errPrefix, cfg.GRPCServerNetwork) + } + return nil +} + // IsLegacyAlertingEnabled returns whether the legacy alerting is enabled or not. // It's safe to be used only after readAlertingSettings() and ReadUnifiedAlertingSettings() are executed. func IsLegacyAlertingEnabled() bool {