LogDog: Use server/auth for authentication.

logdog/server/service/service.go

 // Copyright 2016 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package service
 
 import (
         "flag"
-        "fmt"
         "net/http"
         "net/url"
         "os"
         "os/signal"
         "path/filepath"
         "runtime/pprof"
+        "sort"
+        "strings"
+        "sync"
         "sync/atomic"
         "time"
 
         "github.com/luci/luci-go/appengine/gaesettings"
         "github.com/luci/luci-go/client/authcli"
-        "github.com/luci/luci-go/common/auth"
+        commonAuth "github.com/luci/luci-go/common/auth"
         "github.com/luci/luci-go/common/clock/clockflag"
         "github.com/luci/luci-go/common/config/impl/filesystem"
         "github.com/luci/luci-go/common/data/caching/proccache"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/gcloud/gs"
+        gcps "github.com/luci/luci-go/common/gcloud/pubsub"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/logging/gologger"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/tsmon"
         "github.com/luci/luci-go/common/tsmon/target"
         "github.com/luci/luci-go/grpc/prpc"
         "github.com/luci/luci-go/logdog/api/config/svcconfig"
         "github.com/luci/luci-go/logdog/api/endpoints/coordinator/services/v1"
         "github.com/luci/luci-go/logdog/common/storage"
         "github.com/luci/luci-go/logdog/common/storage/bigtable"
         "github.com/luci/luci-go/logdog/server/retryServicesClient"
         "github.com/luci/luci-go/logdog/server/service/config"
         "github.com/luci/luci-go/luci_config/common/cfgtypes"
         "github.com/luci/luci-go/luci_config/server/cfgclient"
         "github.com/luci/luci-go/luci_config/server/cfgclient/backend/client"
         "github.com/luci/luci-go/luci_config/server/cfgclient/backend/testconfig"
         "github.com/luci/luci-go/luci_config/server/cfgclient/textproto"
+        serverAuth "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/settings"
 
         "github.com/luci/gae/impl/cloud"
 
         "cloud.google.com/go/compute/metadata"
         "cloud.google.com/go/datastore"
+        "cloud.google.com/go/pubsub"
         "golang.org/x/net/context"
-        "golang.org/x/oauth2"
         "google.golang.org/api/option"
 )
 
 var (
         // ErrInvalidConfig is an error that is returned when the supplied
         // configuration is invalid.
         ErrInvalidConfig = errors.New("invalid configuration")
 
         // CoordinatorScopes is the set of OAuth2 scopes to use for the Coordinator
         // client.
         CoordinatorScopes = []string{
-                auth.OAuthScopeEmail,
+                commonAuth.OAuthScopeEmail,
         }
 )
 
 // projectConfigCacheDuration is the amount of time to cache a project's
 // configuration before reloading.
 const (
         projectConfigCacheDuration = 30 * time.Minute
 
         // minAuthTokenLifetime is the amount of time that an access token has before
         // expiring.
         minAuthTokenLifetime = 2 * time.Minute
+
+        // authGlobalCacheSize is the maximum number of elements to store in the auth
+        // global cache LRU.
+        //
+        // We don't expect to load too many different authentication tokens, so a
+        // relatively low number should be fine here.
+        authGlobalCacheSize = 128
 )
 
 // Service is a base class full of common LogDog service application parameters.
 type Service struct {
         // Name is the name of this service. It is used for logging, metrics, and
         // user agent string generation.
         //
         // If empty, a service name will be inferred from the command-line arguments.
         Name string
         // Flags is the set of flags that will be used by the Service.
         Flags flag.FlagSet
 
         shutdownFunc atomic.Value
 
         loggingFlags log.Config
         authFlags    authcli.Flags
         tsMonFlags   tsmon.Flags
 
-        coordinatorHost           string
-        coordinatorInsecure       bool
-        storageCredentialJSONPath string
-        cpuProfilePath            string
-        heapProfilePath           string
+        coordinatorHost     string
+        coordinatorInsecure bool
+        cpuProfilePath      string
+        heapProfilePath     string
 
         // onGCE is true if we're on GCE. We probe this exactly once.
         onGCE        bool
         hasDatastore bool
 
         killCheckInterval clockflag.Duration
         configFilePath    string
         serviceConfig     svcconfig.Config
         configCache       config.ProcCache
 
         // serviceID is the cloud project ID, which is also this service's unique
         // ID. This can be specified by flag or, if on GCE, will automatically be
         // probed from metadata.
         serviceID string
 
         coord logdog.ServicesClient
+
+        // authCache is a cache of instantiated Authenticator instances, keyed on
+        // sorted NULL-delimited scope strings (see authenticatorForScopes).
+        authCacheLock sync.RWMutex
+        authCache     map[string]*commonAuth.Authenticator
 }
 
 // Run performs service-wide initialization and invokes the specified run
 // function.
 func (s *Service) Run(c context.Context, f func(context.Context) error) {
         c = gologger.StdConfig.Use(c)
 
         // If a service name isn't specified, default to the base of the current
         // executable.
         if s.Name == "" {
@@ skipping 60 matching lines @@
 
         // Cancel our Context after we're done our run loop.
         c, cancelFunc := context.WithCancel(c)
         defer cancelFunc()
 
         // Validate the runtime environment.
         if s.serviceID == "" {
                 return errors.New("no service ID was configured (-service-id)")
         }
 
+        // Install our authentication service.
+        c = s.withAuthService(c)
+
         // Install a cloud datastore client. This is non-fatal if it fails.
         dsClient, err := s.initDatastoreClient(c)
         if err == nil {
                 defer dsClient.Close()
 
                 ccfg := cloud.Config{
                         DS: dsClient,
                 }
                 c = ccfg.Use(c)
                 c = settings.Use(c, settings.New(gaesettings.Storage{}))
@@ skipping 65 matching lines @@
         s.loggingFlags.Level = log.Warning
         s.loggingFlags.AddFlags(fs)
 
         // Initialize tsmon flags.
         s.tsMonFlags = tsmon.NewFlags()
         s.tsMonFlags.Flush = tsmon.FlushAuto
         s.tsMonFlags.Target.TargetType = target.TaskType
         s.tsMonFlags.Target.TaskJobName = s.Name
         s.tsMonFlags.Register(fs)
 
-        s.authFlags.Register(fs, auth.Options{})
+        s.authFlags.Register(fs, commonAuth.Options{})
 
         fs.StringVar(&s.serviceID, "service-id", "",
                 "Specify the service ID that this instance is supporting. If empty, the service ID "+
                         "will attempt to be resolved by probing the local environment. This probably will match the "+
                         "App ID of the Coordinator.")
         fs.StringVar(&s.coordinatorHost, "coordinator", "",
                 "The Coordinator service's [host][:port].")
         fs.BoolVar(&s.coordinatorInsecure, "coordinator-insecure", false,
                 "Connect to Coordinator over HTTP (instead of HTTPS).")
-        fs.StringVar(&s.storageCredentialJSONPath, "storage-credential-json-path", "",
-                "If supplied, the path of a JSON credential file to load and use for storage operations.")
         fs.StringVar(&s.cpuProfilePath, "cpu-profile-path", "",
                 "If supplied, enable CPU profiling and write the profile here.")
         fs.StringVar(&s.heapProfilePath, "heap-profile-path", "",
                 "If supplied, enable CPU profiling and write the profile here.")
         fs.Var(&s.killCheckInterval, "config-kill-interval",
                 "If non-zero, poll for configuration changes and kill the application if one is detected.")
         fs.StringVar(&s.configFilePath, "config-file-path", "",
                 "If set, load configuration from a local filesystem rooted here.")
 }
 
@@ skipping 12 matching lines @@
         // project ID.
         if s.serviceID == "" {
                 var err error
                 if s.serviceID, err = metadata.ProjectID(); err != nil {
                         log.WithError(err).Warningf(c, "Failed to probe GCE project ID.")
                 }
         }
 }
 
 func (s *Service) initDatastoreClient(c context.Context) (*datastore.Client, error) {
-        // Initialize Storage authentication.
-        tokenSource, err := s.TokenSource(c, func(o *auth.Options) {
-                o.Scopes = []string{datastore.ScopeDatastore}
-        })
-        if err != nil {
-                log.WithError(err).Errorf(c, "Failed to create datastore TokenSource.")
-                return nil, err
-        }
-
         return datastore.NewClient(c, s.serviceID,
-                option.WithTokenSource(tokenSource))
+                option.WithUserAgent(s.getUserAgent()),
+                option.WithTokenSource(serverAuth.GetTokenSourceAsSelf(c, datastore.ScopeDatastore)))
 }
 
 func (s *Service) initCoordinatorClient(c context.Context) (logdog.ServicesClient, error) {
         if s.coordinatorHost == "" {
                 log.Errorf(c, "Missing Coordinator URL (-coordinator).")
                 return nil, ErrInvalidConfig
         }
 
-        httpClient, err := s.AuthenticatedClient(c, func(o *auth.Options) {
-                o.Scopes = CoordinatorScopes
-        })
+        transport, err := serverAuth.GetRPCTransport(c, serverAuth.AsSelf, serverAuth.WithScopes(CoordinatorScopes...))
         if err != nil {
-                log.WithError(err).Errorf(c, "Failed to create authenticated client.")
+                log.Errorf(c, "Failed to create authenticated transport for Coordinator client.")
                 return nil, err
         }
 
         prpcClient := prpc.Client{
-                C:       httpClient,
+                C: &http.Client{
+                        Transport: transport,
+                },
                 Host:    s.coordinatorHost,
                 Options: prpc.DefaultOptions(),
         }
-        prpcClient.Options.UserAgent = fmt.Sprintf("%s/%s", s.Name, prpc.DefaultUserAgent)
         if s.coordinatorInsecure {
                 prpcClient.Options.Insecure = true
         }
         sc := logdog.NewServicesPRPCClient(&prpcClient)
 
         // Wrap the resulting client in a retry harness.
         return retryServicesClient.New(sc, nil), nil
 }
 
 func (s *Service) initConfig(c *context.Context) error {
@@ skipping 140 matching lines @@
                 log.Errorf(c, "Missing storage configuration.")
                 return nil, ErrInvalidConfig
         }
 
         btcfg := cfg.GetStorage().GetBigtable()
         if btcfg == nil {
                 log.Errorf(c, "Missing BigTable storage configuration")
                 return nil, ErrInvalidConfig
         }
 
-        // Initialize Storage authentication.
-        tokenSource, err := s.TokenSource(c, func(o *auth.Options) {
-                o.Scopes = bigtable.StorageScopes
-                if s.storageCredentialJSONPath != "" {
-                        o.ServiceAccountJSONPath = s.storageCredentialJSONPath
-                }
-        })
-        if err != nil {
-                log.WithError(err).Errorf(c, "Failed to create BigTable TokenSource.")
-                return nil, err
-        }
-
+        // Initialize RPC credentials.
         bt, err := bigtable.New(c, bigtable.Options{
                 Project:  btcfg.Project,
                 Instance: btcfg.Instance,
                 LogTable: btcfg.LogTableName,
                 ClientOptions: []option.ClientOption{
-                        option.WithTokenSource(tokenSource),
+                        option.WithUserAgent(s.getUserAgent()),
+                        option.WithTokenSource(serverAuth.GetTokenSourceAsSelf(c, bigtable.StorageScopes...)),
                 },
         })
         if err != nil {
                 return nil, err
         }
         return bt, nil
 }
 
 // GSClient returns an authenticated Google Storage client instance.
 func (s *Service) GSClient(c context.Context) (gs.Client, error) {
-        rt, err := s.AuthenticatedTransport(c, func(o *auth.Options) {
-                o.Scopes = gs.ReadWriteScopes
-        })
+        // Get an Authenticator bound to the token scopes that we need for
+        // authenticated Cloud Storage access.
+        transport, err := serverAuth.GetRPCTransport(c, serverAuth.AsSelf, serverAuth.WithScopes(gs.ReadWriteScopes...))
         if err != nil {
-                log.WithError(err).Errorf(c, "Failed to create authenticated GS transport.")
+                log.WithError(err).Errorf(c, "Failed to create authenticated transport for Google Storage client.")
                 return nil, err
         }
 
-        client, err := gs.NewProdClient(c, rt)
+        client, err := gs.NewProdClient(c, transport)
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create Google Storage client.")
                 return nil, err
         }
         return client, nil
 }
 
-// Authenticator returns an Authenticator instance. The Authenticator is
-// configured from a base set of Authenticator Options.
-//
-// An optional permutation function can be provided to modify those Options
-// before the Authenticator is created.
-func (s *Service) Authenticator(c context.Context, f func(o *auth.Options)) (*auth.Authenticator, error) {
+// PubSubSubscriberClient returns a Pub/Sub client instance that is
+// authenticated with Pub/Sub subscriber scopes.
+func (s *Service) PubSubSubscriberClient(c context.Context, projectID string) (*pubsub.Client, error) {
+        return pubsub.NewClient(c, projectID,
+                option.WithUserAgent(s.getUserAgent()),
+                option.WithTokenSource(serverAuth.GetTokenSourceAsSelf(c, gcps.SubscriberScopes...)))
+}
+
+func (s *Service) unauthenticatedTransport() http.RoundTripper {
+        smt := serviceModifyingTransport{
+                userAgent: s.getUserAgent(),
+        }
+        return smt.roundTripper(nil)
+}
+
+func (s *Service) getUserAgent() string { return s.Name + " / " + s.serviceID }
+
+// withAuthService configures service-wide authentication and installs it into
+// the supplied Context.
+func (s *Service) withAuthService(c context.Context) context.Context {
+        return serverAuth.SetConfig(c, serverAuth.Config{
+                DBProvider: nil, // We don't need to store an auth DB.
+                Signer:     nil, // We don't need to sign anything.
+                AccessTokenProvider: func(ic context.Context, scopes []string) (commonAuth.Token, error) {
+                        // Create a new Authenticator for the supplied scopes.
+                        //
+                        // Pass our outer Context, since we don't want the cached Authenticator
+                        // instance to be permanently bound to the inner Context.
+                        a, err := s.authenticatorForScopes(c, scopes)
+                        if err != nil {
+                                return commonAuth.Token{}, err
+                        }
+                        return a.GetAccessToken(minAuthTokenLifetime)
+                },
+                AnonymousTransport: func(ic context.Context) http.RoundTripper {
+                        return s.unauthenticatedTransport()
+                },
+                GlobalCache: serverAuth.ProcGlobalCache(authGlobalCacheSize),
+        })
+}
+
+func (s *Service) authenticatorForScopes(c context.Context, scopes []string) (*commonAuth.Authenticator, error) {
+        sort.Strings(scopes)
+        key := strings.Join(scopes, "\x00")
+
+        // First, check holding read lock.
+        s.authCacheLock.RLock()
+        a := s.authCache[key]
+        s.authCacheLock.RUnlock()
+
+        if a != nil {
+                return a, nil
+        }
+
+        // No authenticator yet, check again with write lock.
+        s.authCacheLock.Lock()
+        defer s.authCacheLock.Unlock()
+
+        if a = s.authCache[key]; a != nil {
+                // One was created in between locking!
+                return a, nil
+        }
+
+        // Create a new Authenticator.
         authOpts, err := s.authFlags.Options()
         if err != nil {
                 return nil, ErrInvalidConfig
         }
-        if f != nil {
-                f(&authOpts)
+        authOpts.Scopes = append([]string(nil), scopes...)
+        authOpts.Transport = s.unauthenticatedTransport()
+
+        a = commonAuth.NewAuthenticator(c, commonAuth.SilentLogin, authOpts)
+        if s.authCache == nil {
+                s.authCache = make(map[string]*commonAuth.Authenticator)
         }
-        return auth.NewAuthenticator(c, auth.SilentLogin, authOpts), nil
+        s.authCache[key] = a
+        return a, nil
 }
-
-// AuthenticatedTransport returns an authenticated http.RoundTripper transport.
-// The transport is configured from a base set of Authenticator Options.
-//
-// An optional permutation function can be provided to modify those Options
-// before the Authenticator is created.
-func (s *Service) AuthenticatedTransport(c context.Context, f func(o *auth.Options)) (http.RoundTripper, error) {
-        a, err := s.Authenticator(c, f)
-        if err != nil {
-                return nil, err
-        }
-        return a.Transport()
-}
-
-// AuthenticatedClient returns an authenticated http.Client. The Client is
-// configured from a base set of Authenticator Options.
-//
-// An optional permutation function can be provided to modify those Options
-// before the Authenticator is created.
-func (s *Service) AuthenticatedClient(c context.Context, f func(o *auth.Options)) (*http.Client, error) {
-        a, err := s.Authenticator(c, f)
-        if err != nil {
-                return nil, err
-        }
-        return a.Client()
-}
-
-// TokenSource returns oauth2.TokenSource configured from a base set of
-// Authenticator Options.
-//
-// An optional permutation function can be provided to modify those Options
-// before the Authenticator is created.
-func (s *Service) TokenSource(c context.Context, f func(o *auth.Options)) (oauth2.TokenSource, error) {
-        a, err := s.Authenticator(c, f)
-        if err != nil {
-                return nil, err
-        }
-        return a.TokenSource()
-}