Merge branch 'release/v0.28.0'

CHANGELOG.md

@@ -5,6 +5,19 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+## [v0.28.0] - 2021-10-18
+### Added
+- mongo: adds support for `mongodb+srv` connection strings.
+- mongo: binds in a default TLS Config if `ssl=true` and a TLS config has not been provided.
+- storage: adds `Expirer` interface to enable stores to add support for configuring record expiration.
+- mongo: implements `storage.Expirer` interface to enable TTL based expiry on tokens.
+
+### Changed
+- mongo: migrated internal use of `isDup(err)` to `mongo.IsDuplicateKeyError(err)`.
+
+### Removed
+- mongo: removed internal `isDup(err)` function.
+
 ## [v0.27.0] - 2021-09-24
 This release will add a new hashed index on `signature` for the `accessTokens`
 collection. This makes the old `accessTokens.idxSignatureId` index redundant and
@@ -641,6 +654,7 @@ clear out the password field before sending the response.
 - General pre-release!
 
 [Unreleased]: https://github.com/matthewhartstonge/storage/tree/master
+[v0.28.0]: https://github.com/matthewhartstonge/storage/tree/v0.28.0
 [v0.27.0]: https://github.com/matthewhartstonge/storage/tree/v0.27.0
 [v0.26.0]: https://github.com/matthewhartstonge/storage/tree/v0.26.0
 [v0.25.1]: https://github.com/matthewhartstonge/storage/tree/v0.25.1

README.md

@@ -9,7 +9,6 @@ that conforms to *all the interfaces!* required by [fosite][fosite].
 - [Development](#development)
     - [Testing](#testing)
 - [Examples](#examples)
-- [Disclaimer](#disclaimer)
 
 ## Compatibility
 The following table lists the compatible versions of fosite-storage-mongo with
@@ -18,11 +17,11 @@ know what versions you are successfully paired with.
 
 | storage version | minimum fosite version | maximum fosite version | 
 |----------------:|-----------------------:|-----------------------:|
+|       `v0.28.X` |              `v0.32.X` |              `v0.34.X` |
 |       `v0.27.X` |              `v0.32.X` |              `v0.34.X` |
 |       `v0.26.X` |              `v0.32.X` |              `v0.34.X` |
 |       `v0.25.X` |              `v0.32.X` |              `v0.34.X` |
 |       `v0.24.X` |              `v0.32.X` |              `v0.32.X` |
-|       `v0.22.X` |              `v0.32.X` |              `v0.32.X` |
 
 ## Development
 To start hacking:
@@ -39,13 +38,6 @@ repo for reference:
 
 - [MongoDB Example](./examples/mongo)
 
-## Disclaimer
-* We are currently using this project in house with Storage `v0.26.x` and Fosite
-  `v0.32.x` with good success.
-* If you are able to provide help in keeping storage up to date, feel free to 
-    raise a github issue and discuss where you are able/willing to help. I'm 
-    always happy to review PRs and merge code in :ok_hand:
-
 ## Licensing
 storage is under the Apache 2.0 License.
 

mongo/client_manager.go

@@ -208,7 +208,7 @@ func (c *ClientManager) Create(ctx context.Context, client storage.Client) (resu
 	collection := c.DB.Collection(storage.EntityClients)
 	_, err = collection.InsertOne(ctx, client)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing
@@ -398,7 +398,7 @@ func (c *ClientManager) Update(ctx context.Context, clientID string, updatedClie
 	collection := c.DB.Collection(storage.EntityClients)
 	res, err := collection.ReplaceOne(ctx, selector, updatedClient)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing
@@ -466,7 +466,7 @@ func (c *ClientManager) Migrate(ctx context.Context, migratedClient storage.Clie
 	opts := options.Replace().SetUpsert(true)
 	res, err := collection.ReplaceOne(ctx, selector, migratedClient, opts)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing

mongo/jti_manager.go

@@ -106,7 +106,7 @@ func (d *DeniedJtiManager) Create(ctx context.Context, deniedJTI storage.DeniedJ
 	collection := d.DB.Collection(storage.EntityJtiDenylist)
 	_, err = collection.InsertOne(ctx, deniedJTI)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing

mongo/mongo.go

@@ -4,7 +4,6 @@ import (
 	// Standard Library Imports
 	"context"
 	"crypto/tls"
-	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -126,6 +125,8 @@ type Config struct {
 	Timeout      uint        `default:"10"        envconfig:"CONNECTIONS_MONGO_TIMEOUT"`
 	PoolMinSize  uint64      `default:"0"         envconfig:"CONNECTIONS_MONGO_POOL_MIN_SIZE"`
 	PoolMaxSize  uint64      `default:"100"       envconfig:"CONNECTIONS_MONGO_POOL_MAX_SIZE"`
+	Compressors  []string    `default:""          envconfig:"CONNECTIONS_MONGO_COMPRESSORS"`
+	TokenTTL     uint32      `default:"0"         envconfig:"CONNECTIONS_MONGO_TOKEN_TTL"`
 	TLSConfig    *tls.Config `ignored:"true"`
 }
 
@@ -148,23 +149,29 @@ func ConnectionInfo(cfg *Config) *options.ClientOptions {
 		cfg.DatabaseName = defaultDatabaseName
 	}
 
-	if cfg.Port > 0 {
+	clientOpts := options.Client()
+	if len(cfg.Hostnames) == 1 && strings.HasPrefix(cfg.Hostnames[0], "mongodb+srv://") {
+		// MongoDB SRV records can only be configured with ApplyURI,
+		// but we can continue to mung with client options after it's set.
+		clientOpts.ApplyURI(cfg.Hostnames[0])
+	} else {
 		for i := range cfg.Hostnames {
 			cfg.Hostnames[i] = fmt.Sprintf("%s:%d", cfg.Hostnames[i], cfg.Port)
 		}
+		clientOpts.SetHosts(cfg.Hostnames)
 	}
 
 	if cfg.Timeout == 0 {
 		cfg.Timeout = 10
 	}
 
-	dialInfo := options.Client().
-		SetHosts(cfg.Hostnames).
-		SetReplicaSet(cfg.Replset).
+	clientOpts.SetReplicaSet(cfg.Replset).
 		SetConnectTimeout(time.Second * time.Duration(cfg.Timeout)).
 		SetReadPreference(readpref.SecondaryPreferred()).
 		SetMinPoolSize(cfg.PoolMinSize).
-		SetMaxPoolSize(cfg.PoolMaxSize)
+		SetMaxPoolSize(cfg.PoolMaxSize).
+		SetCompressors(cfg.Compressors).
+		SetAppName(cfg.DatabaseName)
 
 	if cfg.Username != "" || cfg.Password != "" {
 		auth := options.Credential{
@@ -173,14 +180,24 @@ func ConnectionInfo(cfg *Config) *options.ClientOptions {
 			Username:      cfg.Username,
 			Password:      cfg.Password,
 		}
-		dialInfo.SetAuth(auth)
+		clientOpts.SetAuth(auth)
 	}
 
 	if cfg.SSL {
-		dialInfo = dialInfo.SetTLSConfig(cfg.TLSConfig)
+		tlsConfig := cfg.TLSConfig
+		if tlsConfig == nil {
+			// Inject a default TLS config if the SSL switch is toggled, but a
+			// TLS config has not been provided programmatically.
+			tlsConfig = &tls.Config{
+				InsecureSkipVerify: false,
+				MinVersion:         tls.VersionTLS12,
+			}
+		}
+
+		clientOpts.SetTLSConfig(tlsConfig)
 	}
 
-	return dialInfo
+	return clientOpts
 }
 
 // Connect returns a connection to a mongo database.
@@ -254,14 +271,6 @@ func New(cfg *Config, hashee fosite.Hasher) (*Store, error) {
 		Users:   mongoUsers,
 	}
 
-	// Init DB collections, indices e.t.c.
-	managers := []storage.Configurer{
-		mongoClients,
-		mongoDeniedJtis,
-		mongoUsers,
-		mongoRequests,
-	}
-
 	// attempt to perform index updates in a session.
 	var closeSession func()
 	ctx, closeSession, err := newSession(context.Background(), mongoDB)
@@ -271,11 +280,14 @@ func New(cfg *Config, hashee fosite.Hasher) (*Store, error) {
 	}
 	defer closeSession()
 
-	// Configure the mongo collections on first up.
-	for _, manager := range managers {
-		err := manager.Configure(ctx)
-		if err != nil {
-			log.WithError(err).Error("Unable to configure mongo collections!")
+	// Configure DB collections, indices, TTLs e.t.c.
+	if err = configureDatabases(ctx, mongoClients, mongoDeniedJtis, mongoUsers, mongoRequests); err != nil {
+		log.WithError(err).Error("Unable to configure mongo collections!")
+		return nil, err
+	}
+	if cfg.TokenTTL > 0 {
+		if err = configureExpiry(ctx, int(cfg.TokenTTL), mongoRequests); err != nil {
+			log.WithError(err).Error("Unable to configure mongo expiry!")
 			return nil, err
 		}
 	}
@@ -294,31 +306,35 @@ func New(cfg *Config, hashee fosite.Hasher) (*Store, error) {
 	return store, nil
 }
 
-// NewDefaultStore returns a Store configured with the default mongo
-// configuration and default Hasher.
-func NewDefaultStore() (*Store, error) {
-	cfg := DefaultConfig()
-	return New(cfg, nil)
-}
+// configureDatabases calls the configuration handler for the provided
+// configurers.
+func configureDatabases(ctx context.Context, configurers ...storage.Configurer) error {
+	for _, configurer := range configurers {
+		if err := configurer.Configure(ctx); err != nil {
+			return err
+		}
+	}
 
-const (
-	// errCodeDuplicate provides the mongo error code for duplicate key error.
-	errCodeDuplicate = 11000
-)
+	return nil
+}
 
-// isDup replicates mgo.IsDup functionality for the official driver in order
-// to know when a conflict has occurred.
-func isDup(err error) (isDup bool) {
-	var e mongo.WriteException
-	if errors.As(err, &e) {
-		for _, we := range e.WriteErrors {
-			if we.Code == errCodeDuplicate {
-				return true
-			}
+// configureExpiry calls the configuration handler for the provided expirers.
+// ttl should be a positive integer.
+func configureExpiry(ctx context.Context, ttl int, expirers ...storage.Expirer) error {
+	for _, expirer := range expirers {
+		if err := expirer.ConfigureExpiryWithTTL(ctx, ttl); err != nil {
+			return err
 		}
 	}
 
-	return
+	return nil
+}
+
+// NewDefaultStore returns a Store configured with the default mongo
+// configuration and default Hasher.
+func NewDefaultStore() (*Store, error) {
+	cfg := DefaultConfig()
+	return New(cfg, nil)
 }
 
 // NewIndex generates a new index model, ready to be saved in mongo.
@@ -342,6 +358,16 @@ func NewUniqueIndex(name string, keys ...string) mongo.IndexModel {
 	}
 }
 
+// NewExpiryIndex generates a new index with a time to live value before the
+// record expires in mongodb.
+func NewExpiryIndex(name string, key string, expireAfter int) (model mongo.IndexModel) {
+	return mongo.IndexModel{
+		Keys: bson.D{{Key: key, Value: int32(1)}},
+		Options: generateIndexOptions(name, false).
+			SetExpireAfterSeconds(int32(expireAfter)),
+	}
+}
+
 // generateIndexKeys given a number of stringy keys will return a bson
 // document containing keys in the structure required by mongo for defining
 // index and sort order.

mongo/mongo_meta.go

@@ -15,6 +15,10 @@ const (
 	// IdxExpires provides a mongo index based on expires
 	IdxExpires = "idxExpires"
 
+	// IdxExpiry provides a mongo index for generating ttl based record
+	// expiration indices.
+	IdxExpiry = "idxExpiry"
+
 	// IdxUserID provides a mongo index based on userId
 	IdxUserID = "idxUserId"
 

mongo/request_manager.go

@@ -86,6 +86,35 @@ func (r *RequestManager) Configure(ctx context.Context) (err error) {
 	return nil
 }
 
+// ConfigureExpiryWithTTL implements storage.Expirer.
+func (r *RequestManager) ConfigureExpiryWithTTL(ctx context.Context, ttl int) error {
+	collections := []string{
+		storage.EntityAccessTokens,
+		storage.EntityAuthorizationCodes,
+		storage.EntityOpenIDSessions,
+		storage.EntityPKCESessions,
+		storage.EntityRefreshTokens,
+	}
+
+	for _, entityName := range collections {
+		log := logger.WithFields(logrus.Fields{
+			"package":    "mongo",
+			"collection": entityName,
+			"method":     "ConfigureExpiryWithTTL",
+		})
+
+		index := NewExpiryIndex(IdxExpiry+"RequestedAt", "requestedAt", ttl)
+		collection := r.DB.Collection(entityName)
+		_, err := collection.Indexes().CreateOne(ctx, index)
+		if err != nil {
+			log.WithError(err).Error(logError)
+			return err
+		}
+	}
+
+	return nil
+}
+
 // getConcrete returns a Request resource.
 func (r *RequestManager) getConcrete(ctx context.Context, entityName string, requestID string) (result storage.Request, err error) {
 	log := logger.WithFields(logrus.Fields{
@@ -220,7 +249,7 @@ func (r *RequestManager) Create(ctx context.Context, entityName string, request
 	collection := r.DB.Collection(entityName)
 	_, err = collection.InsertOne(ctx, request)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing
@@ -317,7 +346,7 @@ func (r *RequestManager) Update(ctx context.Context, entityName string, requestI
 	collection := r.DB.Collection(entityName)
 	res, err := collection.ReplaceOne(ctx, selector, updatedRequest)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing

mongo/user_manager.go

@@ -200,7 +200,7 @@ func (u *UserManager) Create(ctx context.Context, user storage.User) (result sto
 	collection := u.DB.Collection(storage.EntityUsers)
 	_, err = collection.InsertOne(ctx, user)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing
@@ -332,7 +332,7 @@ func (u *UserManager) Update(ctx context.Context, userID string, updatedUser sto
 	collection := u.DB.Collection(storage.EntityUsers)
 	res, err := collection.ReplaceOne(ctx, selector, updatedUser)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing
@@ -399,7 +399,7 @@ func (u *UserManager) Migrate(ctx context.Context, migratedUser storage.User) (r
 	opts := options.Replace().SetUpsert(true)
 	_, err = collection.ReplaceOne(ctx, selector, migratedUser, opts)
 	if err != nil {
-		if isDup(err) {
+		if mongo.IsDuplicateKeyError(err) {
 			// Log to StdOut
 			log.WithError(err).Debug(logConflict)
 			// Log to OpenTracing

storage.go

@@ -88,3 +88,9 @@ type Configurer interface {
 	// any needed migrations and configuration of indexes as required.
 	Configure(ctx context.Context) error
 }
+
+type Expirer interface {
+	// ConfigureExpiryWithTTL enables a datastore provider to purge data
+	// automatically once expired.
+	ConfigureExpiryWithTTL(ctx context.Context, ttl int) error
+}