Add nonce storage and replay prevention test (#293)

globals/globals.go

@@ -2,12 +2,12 @@ package globals
 
 import "time"
 
-// NOTE:
-// Globals are bad. But there are some situations where we want a value that is
-// constant per invocation but invocation may be in various places, such as a
-// test or via CLI, and placing such a value in any given package leads to
-// import issues. These values should only ever be set at startup, but simply
-// available to reference from anywhere.
+// NOTE: Globals, in the traditional sense, are bad. But there are some
+// situations where we want a value that is constant per invocation but
+// invocation may be in various places, such as a test or via CLI, and placing
+// such a value in any given package leads to import issues. So think of this
+// package as "freeing us from circular dependency hell". These values should
+// only ever be set at startup, but simply available to reference from anywhere.
 
 type ContextMaxRequestSizeType int
 type ContextOriginalRequestPathType int
@@ -26,4 +26,8 @@ var (
 	// ContextOriginalRequestPathTypeKey is a value to keep linters from complaining
 	// about clashing string identifiers
 	ContextOriginalRequestPathTypeKey ContextOriginalRequestPathType
+
+	// RecoveryTokenValidityPeriod is exported so we can modify it in tests if
+	// we want
+	RecoveryTokenValidityPeriod = 5 * time.Minute
 )

internal/auth/auth.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/gen/controller/api/resources/scopes"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/perms"
@@ -38,13 +39,7 @@ const (
 
 type key int
 
-var (
-	// RecoveryTokenValidityPeriod is exported so we can modify it in tests if
-	// we want
-	RecoveryTokenValidityPeriod = 5 * time.Minute
-
-	verifierKey key
-)
+var verifierKey key
 
 // RequestInfo contains request parameters necessary for checking authn/authz
 type RequestInfo struct {
@@ -73,6 +68,7 @@ type verifier struct {
 	logger          hclog.Logger
 	iamRepoFn       common.IamRepoFactory
 	authTokenRepoFn common.AuthTokenRepoFactory
+	serversRepoFn   common.ServersRepoFactory
 	kms             *kms.Kms
 	requestInfo     RequestInfo
 	res             *perms.Resource
@@ -88,12 +84,14 @@ func NewVerifierContext(ctx context.Context,
 	logger hclog.Logger,
 	iamRepoFn common.IamRepoFactory,
 	authTokenRepoFn common.AuthTokenRepoFactory,
+	serversRepoFn common.ServersRepoFactory,
 	kms *kms.Kms,
 	requestInfo RequestInfo) context.Context {
 	return context.WithValue(ctx, verifierKey, &verifier{
 		logger:          logger,
 		iamRepoFn:       iamRepoFn,
 		authTokenRepoFn: authTokenRepoFn,
+		serversRepoFn:   serversRepoFn,
 		kms:             kms,
 		requestInfo:     requestInfo,
 	})
@@ -423,16 +421,22 @@ func (v verifier) performAuthCheck() (aclResults *perms.ACLResults, userId strin
 			retErr = fmt.Errorf("perform auth check: error validating recovery token: %w", err)
 			return
 		}
-
 		// If we add the validity period to the creation time (which we've
 		// verified is before the current time, with a minute of fudging), and
 		// it's before now, it's expired and might be a replay.
-		if info.CreationTime.Add(RecoveryTokenValidityPeriod).Before(time.Now()) {
-			retErr = errors.New("perform auth check: recovery token has expired")
+		if info.CreationTime.Add(globals.RecoveryTokenValidityPeriod).Before(time.Now()) {
+			retErr = errors.New("WARNING: perform auth check: recovery token has expired (possible replay attack)")
+			return
+		}
+		repo, err := v.serversRepoFn()
+		if err != nil {
+			retErr = fmt.Errorf("perform auth check: error fetching servers repo: %w", err)
+			return
+		}
+		if err := repo.AddRecoveryNonce(v.ctx, info.Nonce); err != nil {
+			retErr = fmt.Errorf("WARNING: perform auth check: error adding nonce to database (possible replay attack): %w", err)
 			return
 		}
-		// TODO: verify nonce hasn't been used via the DB
-		_ = info
 		v.logger.Warn("NOTE: recovery KMS was used to authorize a call", "url", v.requestInfo.Path, "method", v.requestInfo.Method)
 	}
 

internal/auth/auth_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/iam"
 	"github.com/hashicorp/boundary/internal/kms"
+	"github.com/hashicorp/boundary/internal/servers"
 	"github.com/hashicorp/boundary/internal/types/action"
 	"github.com/hashicorp/boundary/internal/types/resource"
 	"github.com/kr/pretty"
@@ -281,6 +282,9 @@ func TestAuthTokenAuthenticator(t *testing.T) {
 	iamRepoFn := func() (*iam.Repository, error) {
 		return iamRepo, nil
 	}
+	serversRepoFn := func() (*servers.Repository, error) {
+		return servers.NewRepository(rw, rw, kms)
+	}
 
 	o, _ := iam.TestScopes(t, iamRepo)
 	at := authtoken.TestAuthToken(t, conn, kms, o.GetPublicId())
@@ -362,7 +366,7 @@ func TestAuthTokenAuthenticator(t *testing.T) {
 			if tc.userId == "" {
 				return
 			}
-			ctx := NewVerifierContext(context.Background(), nil, iamRepoFn, tokenRepoFn, kms, requestInfo)
+			ctx := NewVerifierContext(context.Background(), nil, iamRepoFn, tokenRepoFn, serversRepoFn, kms, requestInfo)
 
 			v, ok := ctx.Value(verifierKey).(*verifier)
 			require.True(t, ok)

internal/auth/password/authmethod_test.go

@@ -18,8 +18,7 @@ func TestAuthMethod_New(t *testing.T) {
 	w := db.New(conn)
 
 	type args struct {
-		scopeId string
-		opts    []Option
+		opts []Option
 	}
 
 	var tests = []struct {

internal/auth/testing.go

@@ -8,5 +8,5 @@ func DisabledAuthTestContext(opt ...Option) context.Context {
 	opts := getOpts(opt...)
 	reqInfo.scopeIdOverride = opts.withScopeId
 	reqInfo.userIdOverride = opts.withUserId
-	return NewVerifierContext(context.Background(), nil, nil, nil, opts.withKms, reqInfo)
+	return NewVerifierContext(context.Background(), nil, nil, nil, nil, opts.withKms, reqInfo)
 }

internal/cmd/base/base.go

@@ -19,9 +19,6 @@ import (
 	"github.com/hashicorp/boundary/api"
 	"github.com/hashicorp/boundary/api/authtokens"
 	"github.com/hashicorp/boundary/recovery"
-	wrapping "github.com/hashicorp/go-kms-wrapping"
-	"github.com/hashicorp/shared-secure-libs/configutil"
-	"github.com/hashicorp/vault/sdk/helper/strutil"
 	"github.com/mitchellh/cli"
 	"github.com/pkg/errors"
 	"github.com/posener/complete"
@@ -184,7 +181,7 @@ func (c *Command) Client(opt ...Option) (*api.Client, error) {
 	tokenName := "default"
 	switch {
 	case c.FlagRecoveryConfig != "":
-		wrapper, err := getWrapper(c.Context, c.FlagRecoveryConfig)
+		wrapper, err := recovery.GetWrapper(c.Context, c.FlagRecoveryConfig)
 		if err != nil {
 			return nil, err
 		}
@@ -557,30 +554,3 @@ func printFlagDetail(w io.Writer, f *flag.Flag) {
 	indented := WrapAtLengthWithPadding(usage, 6)
 	fmt.Fprintf(w, "%s\n\n", indented)
 }
-
-func getWrapper(ctx context.Context, path string) (wrapping.Wrapper, error) {
-	kmses, err := configutil.LoadConfigKMSes(path)
-	if err != nil {
-		return nil, fmt.Errorf("Error parsing config file: %w", err)
-	}
-
-	var kms *configutil.KMS
-	for _, v := range kmses {
-		if strutil.StrListContains(v.Purpose, "recovery") {
-			if kms != nil {
-				return nil, fmt.Errorf("Only one %q block marked for %q purpose is allowed", "kms", "recovery")
-			}
-			kms = v
-		}
-	}
-	if kms == nil {
-		return nil, fmt.Errorf("No %q block marked for %q purpose is allowed", "kms", "recovery")
-	}
-
-	wrapper, err := configutil.ConfigureWrapper(kms, nil, nil, nil)
-	if err != nil {
-		return nil, fmt.Errorf("Error configuring kms: %w", err)
-	}
-
-	return wrapper, nil
-}

internal/cmd/base/servers.go

@@ -414,8 +414,10 @@ func (b *Server) ConnectToDatabase(dialect string) error {
 	}
 
 	b.Database = dbase
-	gorm.LogFormatter = db.GetGormLogFormatter(b.Logger)
-	b.Database.SetLogger(db.GetGormLogger(b.Logger))
+	if os.Getenv("BOUNDARY_DISABLE_GORM_FORMATTER") == "" {
+		gorm.LogFormatter = db.GetGormLogFormatter(b.Logger)
+		b.Database.SetLogger(db.GetGormLogger(b.Logger))
+	}
 	return nil
 }
 

internal/db/migrations/postgres.gen.go

@@ -1417,6 +1417,24 @@ before
 insert on servers
   for each row execute procedure default_create_time();
 
+create table recovery_nonces (
+    nonce text
+      primary key,
+    create_time wt_timestamp
+  );
+
+create trigger 
+  default_create_time_column
+before
+insert on recovery_nonces
+  for each row execute procedure default_create_time();
+
+create trigger 
+  immutable_columns
+before
+update on recovery_nonces
+  for each row execute procedure immutable_columns('nonce', 'create_time');
+
 commit;
 
 `),

internal/db/migrations/postgres/08_servers.up.sql

@@ -28,4 +28,22 @@ before
 insert on servers
   for each row execute procedure default_create_time();
 
+create table recovery_nonces (
+    nonce text
+      primary key,
+    create_time wt_timestamp
+  );
+
+create trigger 
+  default_create_time_column
+before
+insert on recovery_nonces
+  for each row execute procedure default_create_time();
+
+create trigger 
+  immutable_columns
+before
+update on recovery_nonces
+  for each row execute procedure immutable_columns('nonce', 'create_time');
+
 commit;

internal/db/read_writer.go

@@ -485,10 +485,11 @@ func (rw *Db) Update(ctx context.Context, i interface{}, fieldMaskPaths []string
 	return rowsUpdated, nil
 }
 
-// Delete an object in the db with options: WithOplog and NewOplogMsg. WithOplog
-// will write an oplog entry for the delete. NewOplogMsg will return in-memory
-// oplog message.  WithOplog and NewOplogMsg cannot be used together. Delete
-// returns the number of rows deleted and any errors.
+// Delete an object in the db with options: WithOplog, NewOplogMsg, WithWhere.
+// WithOplog will write an oplog entry for the delete. NewOplogMsg will return
+// in-memory oplog message. WithOplog and NewOplogMsg cannot be used together.
+// WithWhere allows specifying a constraint. Delete returns the number of rows
+// deleted and any errors.
 func (rw *Db) Delete(ctx context.Context, i interface{}, opt ...Option) (int, error) {
 	if rw.underlying == nil {
 		return NoRowsAffected, fmt.Errorf("delete: missing underlying db %w", ErrNilParameter)
@@ -504,8 +505,10 @@ func (rw *Db) Delete(ctx context.Context, i interface{}, opt ...Option) (int, er
 	// This is not a boundary scope, but rather a gorm Scope:
 	// https://godoc.org/github.com/jinzhu/gorm#DB.NewScope
 	scope := rw.underlying.NewScope(i)
-	if scope.PrimaryKeyZero() {
-		return NoRowsAffected, fmt.Errorf("delete: primary key is not set")
+	if opts.withWhereClause == "" {
+		if scope.PrimaryKeyZero() {
+			return NoRowsAffected, fmt.Errorf("delete: primary key is not set")
+		}
 	}
 	if withOplog {
 		_, err := validateOplogArgs(i, opts)
@@ -521,11 +524,15 @@ func (rw *Db) Delete(ctx context.Context, i interface{}, opt ...Option) (int, er
 			return NoRowsAffected, fmt.Errorf("delete: unable to get ticket: %w", err)
 		}
 	}
-	underlying := rw.underlying.Delete(i)
-	if underlying.Error != nil {
-		return NoRowsAffected, fmt.Errorf("delete: failed %w", underlying.Error)
+	db := rw.underlying
+	if opts.withWhereClause != "" {
+		db = db.Where(opts.withWhereClause, opts.withWhereClauseArgs...)
 	}
-	rowsDeleted := int(underlying.RowsAffected)
+	db = db.Delete(i)
+	if db.Error != nil {
+		return NoRowsAffected, fmt.Errorf("delete: failed %w", db.Error)
+	}
+	rowsDeleted := int(db.RowsAffected)
 	if rowsDeleted > 0 && (withOplog || opts.newOplogMsg != nil) {
 		if withOplog {
 			if err := rw.addOplog(ctx, DeleteOp, opts, ticket, i); err != nil {
@@ -983,14 +990,26 @@ func (rw *Db) SearchWhere(ctx context.Context, resources interface{}, where stri
 		return errors.New("error interface parameter must to be a pointer for search by")
 	}
 	var err error
+	db := rw.underlying.Order(opts.withOrder)
+
+	// Perform limiting
 	switch {
 	case opts.WithLimit < 0: // any negative number signals unlimited results
-		err = rw.underlying.Order(opts.withOrder).Where(where, args...).Find(resources).Error
 	case opts.WithLimit == 0: // zero signals the default value and default limits
-		err = rw.underlying.Order(opts.withOrder).Limit(DefaultLimit).Where(where, args...).Find(resources).Error
+		db = db.Limit(DefaultLimit)
 	default:
-		err = rw.underlying.Order(opts.withOrder).Limit(opts.WithLimit).Where(where, args...).Find(resources).Error
+		db = db.Limit(opts.WithLimit)
 	}
+
+	// Perform argument subst
+	switch len(args) {
+	case 0:
+	default:
+		db = db.Where(where, args...)
+	}
+
+	// Perform the query
+	err = db.Find(resources).Error
 	if err != nil {
 		// searching with a slice parameter does not return a gorm.ErrRecordNotFound
 		return err

internal/servers/controller/controller.go

@@ -113,7 +113,7 @@ func New(conf *Config) (*Controller, error) {
 		return authtoken.NewRepository(dbase, dbase, c.kms)
 	}
 	c.ServersRepoFn = func() (*servers.Repository, error) {
-		return servers.NewRepository(c.logger.Named("servers.repository"), dbase, dbase, c.kms)
+		return servers.NewRepository(dbase, dbase, c.kms)
 	}
 	c.PasswordAuthRepoFn = func() (*password.Repository, error) {
 		return password.NewRepository(dbase, dbase, c.kms)
@@ -136,6 +136,7 @@ func (c *Controller) Start() error {
 	}
 
 	c.startStatusTicking(c.baseContext)
+	c.startRecoveryNonceCleanupTicking(c.baseContext)
 	c.started.Store(true)
 
 	return nil

internal/servers/controller/handler.go

@@ -206,7 +206,7 @@ func wrapHandlerWithCommonFuncs(h http.Handler, c *Controller, props HandlerProp
 		}
 
 		requestInfo.PublicId, requestInfo.Token, requestInfo.TokenFormat = auth.GetTokenFromRequest(c.logger, r)
-		ctx = auth.NewVerifierContext(ctx, c.logger, c.IamRepoFn, c.AuthTokenRepoFn, c.kms, requestInfo)
+		ctx = auth.NewVerifierContext(ctx, c.logger, c.IamRepoFn, c.AuthTokenRepoFn, c.ServersRepoFn, c.kms, requestInfo)
 
 		// Set the context back on the request
 		r = r.WithContext(ctx)

internal/servers/controller/handlers/authtokens/authtoken_service.go

@@ -15,8 +15,6 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-const orgIdFieldName = "org_id"
-
 // Service handles request as described by the pbs.AuthTokenServiceServer interface.
 type Service struct {
 	repoFn func() (*authtoken.Repository, error)

internal/servers/controller/handlers/workers/worker_service.go

@@ -36,7 +36,7 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error aqcuiring repo to store worker status: %v", err)
 	}
 	req.Worker.Type = resource.Worker.String()
-	controllers, _, err := repo.Upsert(ctx, req.Worker)
+	controllers, _, err := repo.UpsertServer(ctx, req.Worker)
 	if err != nil {
 		ws.logger.Error("error storing worker status", "error", err)
 		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err)

internal/servers/controller/testing.go

@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	"github.com/hashicorp/boundary/internal/iam"
+	"github.com/hashicorp/boundary/internal/servers"
 	"github.com/hashicorp/go-hclog"
 	wrapping "github.com/hashicorp/go-kms-wrapping"
 	"github.com/hashicorp/vault/sdk/helper/base62"
@@ -56,6 +57,14 @@ func (tc *TestController) IamRepo() *iam.Repository {
 	return repo
 }
 
+func (tc *TestController) ServersRepo() *servers.Repository {
+	repo, err := tc.c.ServersRepoFn()
+	if err != nil {
+		tc.t.Fatal(err)
+	}
+	return repo
+}
+
 func (tc *TestController) Cancel() {
 	tc.cancel()
 }

internal/servers/controller/tickers.go

@@ -13,6 +13,9 @@ const (
 	statusInterval = 10 * time.Second
 )
 
+// This is exported so it can be tweaked in tests
+var RecoveryNonceCleanupInterval = 2 * time.Minute
+
 func (c *Controller) startStatusTicking(cancelCtx context.Context) {
 	go func() {
 		timer := time.NewTimer(0)
@@ -34,7 +37,7 @@ func (c *Controller) startStatusTicking(cancelCtx context.Context) {
 				if err != nil {
 					c.logger.Error("error fetching repository for status update", "error", err)
 				} else {
-					_, _, err = repo.Upsert(cancelCtx, server)
+					_, _, err = repo.UpsertServer(cancelCtx, server)
 					if err != nil {
 						c.logger.Error("error performing status update", "error", err)
 					} else {
@@ -46,3 +49,30 @@ func (c *Controller) startStatusTicking(cancelCtx context.Context) {
 		}
 	}()
 }
+
+func (c *Controller) startRecoveryNonceCleanupTicking(cancelCtx context.Context) {
+	go func() {
+		timer := time.NewTimer(0)
+		for {
+			select {
+			case <-cancelCtx.Done():
+				c.logger.Info("recovery nonce ticking shutting down")
+				return
+
+			case <-timer.C:
+				repo, err := c.ServersRepoFn()
+				if err != nil {
+					c.logger.Error("error fetching repository for recovery nonce cleanup", "error", err)
+				} else {
+					nonceCount, err := repo.CleanupNonces(cancelCtx)
+					if err != nil {
+						c.logger.Error("error performing recovery nonce cleanup", "error", err)
+					} else if nonceCount > 0 {
+						c.logger.Info("recovery nonce cleanup successful", "nonces_cleaned", nonceCount)
+					}
+				}
+				timer.Reset(RecoveryNonceCleanupInterval)
+			}
+		}
+	}()
+}

internal/servers/query.go

@@ -0,0 +1,5 @@
+package servers
+
+const (
+	deleteWhereSql = `create_time < $1`
+)

internal/servers/repository.go

@@ -6,28 +6,27 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/db"
 	timestamp "github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/types/resource"
-	"github.com/hashicorp/go-hclog"
 )
 
 const (
 	defaultLiveness = 15 * time.Second
 )
 
-// Repository is the jobs database repository
+// Repository is the servers database repository
 type Repository struct {
-	logger hclog.Logger
 	reader db.Reader
 	writer db.Writer
 	kms    *kms.Kms
 }
 
-// NewRepository creates a new jobs Repository. Supports the options: WithLimit
+// NewRepository creates a new servers Repository. Supports the options: WithLimit
 // which sets a default limit on results returned by repo operations.
-func NewRepository(logger hclog.Logger, r db.Reader, w db.Writer, kms *kms.Kms) (*Repository, error) {
+func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms) (*Repository, error) {
 	if r == nil {
 		return nil, errors.New("error creating server repository with nil reader")
 	}
@@ -38,7 +37,6 @@ func NewRepository(logger hclog.Logger, r db.Reader, w db.Writer, kms *kms.Kms)
 		return nil, errors.New("error creating server repository with nil kms")
 	}
 	return &Repository{
-		logger: logger,
 		reader: r,
 		writer: w,
 		kms:    kms,
@@ -47,7 +45,7 @@ func NewRepository(logger hclog.Logger, r db.Reader, w db.Writer, kms *kms.Kms)
 
 // list will return a listing of resources and honor the WithLimit option or the
 // repo defaultLimit
-func (r *Repository) List(ctx context.Context, serverType string, opt ...Option) ([]*Server, error) {
+func (r *Repository) ListServers(ctx context.Context, serverType string, opt ...Option) ([]*Server, error) {
 	opts := getOpts(opt...)
 	liveness := opts.withLiveness
 	if liveness == 0 {
@@ -71,6 +69,7 @@ func (r *Repository) List(ctx context.Context, serverType string, opt ...Option)
 		return nil, fmt.Errorf("error performing server list: %w", err)
 	}
 	results := make([]*Server, 0, 3)
+	var scanErr error
 	for rows.Next() {
 		server := &Server{
 			CreateTime: new(timestamp.Timestamp),
@@ -85,19 +84,22 @@ func (r *Repository) List(ctx context.Context, serverType string, opt ...Option)
 			server.CreateTime,
 			server.UpdateTime,
 		); err != nil {
-			r.logger.Error("error scanning server row", "error", err)
+			scanErr = fmt.Errorf("error scanning server row: %w", err)
 			break
 		}
 		results = append(results, server)
 	}
+	if scanErr != nil {
+		return nil, scanErr
+	}
 	if err := rows.Err(); err != nil {
 		return nil, fmt.Errorf("error performing scan over server rows: %w", err)
 	}
 	return results, nil
 }
 
-// upsert will upsert
-func (r *Repository) Upsert(ctx context.Context, server *Server, opt ...Option) ([]*Server, int, error) {
+// UpsertServer adds or updates a server in the DB
+func (r *Repository) UpsertServer(ctx context.Context, server *Server, opt ...Option) ([]*Server, int, error) {
 	if server == nil {
 		return nil, db.NoRowsAffected, errors.New("cannot update server that is nil")
 	}
@@ -139,6 +141,43 @@ func (r *Repository) Upsert(ctx context.Context, server *Server, opt ...Option)
 		return nil, int(rowsAffected), nil
 	}
 	// Fetch current controllers to feed to the workers
-	controllers, err := r.List(ctx, resource.Controller.String())
+	controllers, err := r.ListServers(ctx, resource.Controller.String())
 	return controllers, len(controllers), err
 }
+
+type RecoveryNonce struct {
+	Nonce string
+}
+
+// AddRecoveryNonce adds a nonce
+func (r *Repository) AddRecoveryNonce(ctx context.Context, nonce string, opt ...Option) error {
+	if nonce == "" {
+		return errors.New("empty nonce provided")
+	}
+	rn := &RecoveryNonce{Nonce: nonce}
+	if err := r.writer.Create(ctx, rn); err != nil {
+		return fmt.Errorf("error performing nonce insertion: %w", err)
+	}
+	return nil
+}
+
+// CleanupNonces removes nonces that no longer need to be stored
+func (r *Repository) CleanupNonces(ctx context.Context, opt ...Option) (int, error) {
+	// If something was inserted before 3x the actual validity period, clean it out
+	endTime := time.Now().Add(-3 * globals.RecoveryTokenValidityPeriod)
+
+	rows, err := r.writer.Delete(ctx, &RecoveryNonce{}, db.WithWhere(deleteWhereSql, endTime))
+	if err != nil {
+		return db.NoRowsAffected, fmt.Errorf("error performing nonce cleanup: %w", err)
+	}
+	return rows, nil
+}
+
+// ListNonces lists nonces. Used only for tests at the moment.
+func (r *Repository) ListNonces(ctx context.Context, opt ...Option) ([]*RecoveryNonce, error) {
+	var nonces []*RecoveryNonce
+	if err := r.reader.SearchWhere(ctx, &nonces, "", nil, db.WithLimit(-1)); err != nil {
+		return nil, fmt.Errorf("error listing nonces: %w", err)
+	}
+	return nonces, nil
+}

internal/servers/repository_test.go

@@ -0,0 +1,83 @@
+package servers_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hashicorp/boundary/api/roles"
+	"github.com/hashicorp/boundary/globals"
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/servers/controller"
+	"github.com/hashicorp/boundary/recovery"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRecoveryNonces(t *testing.T) {
+	assert, require := assert.New(t), require.New(t)
+
+	// Set these low so that we can not have the test run forever
+	globals.RecoveryTokenValidityPeriod = 10 * time.Second
+	controller.RecoveryNonceCleanupInterval = 20 * time.Second
+
+	wrapper := db.TestWrapper(t)
+	tc := controller.NewTestController(t, &controller.TestControllerOpts{
+		RecoveryKms: wrapper,
+	})
+	defer tc.Shutdown()
+
+	client := tc.Client()
+	repo := tc.ServersRepo()
+
+	// First, validate that we can't use the same token twice. Get two tokens,
+	// try to use the first twice (should succeed first time, fail second), then
+	// ensure second token works
+	token1, err := recovery.GenerateRecoveryToken(tc.Context(), wrapper)
+	require.NoError(err)
+	token2, err := recovery.GenerateRecoveryToken(tc.Context(), wrapper)
+	require.NoError(err)
+
+	// Token 1, try 1
+	client.SetToken(token1)
+	roleClient := roles.NewRolesClient(client)
+	_, apiErr, err := roleClient.Create(tc.Context())
+	require.NoError(err)
+	assert.Nil(apiErr)
+	nonces, err := repo.ListNonces(tc.Context())
+	require.NoError(err)
+	assert.Len(nonces, 1)
+
+	// Token 1, try 2
+	_, apiErr, err = roleClient.Create(tc.Context())
+	require.NoError(err)
+	assert.NotNil(apiErr)
+	nonces, err = repo.ListNonces(tc.Context())
+	require.NoError(err)
+	assert.Len(nonces, 1)
+
+	// Token 2
+	client.SetToken(token2)
+	_, apiErr, err = roleClient.Create(tc.Context())
+	require.NoError(err)
+	assert.Nil(apiErr)
+	nonces, err = repo.ListNonces(tc.Context())
+	require.NoError(err)
+	assert.Len(nonces, 2)
+
+	// Make sure they get cleaned up
+	time.Sleep(2 * controller.RecoveryNonceCleanupInterval)
+	nonces, err = repo.ListNonces(tc.Context())
+	require.NoError(err)
+	assert.Len(nonces, 0)
+
+	// And finally, make sure they still can't be used
+	for _, token := range []string{token1, token2} {
+		client.SetToken(token)
+		_, apiErr, err = roleClient.Create(tc.Context())
+		require.NoError(err)
+		assert.NotNil(apiErr)
+		nonces, err = repo.ListNonces(tc.Context())
+		require.NoError(err)
+		assert.Len(nonces, 0)
+	}
+}

recovery/recovery.go

@@ -11,6 +11,8 @@ import (
 
 	wrapping "github.com/hashicorp/go-kms-wrapping"
 	"github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/shared-secure-libs/configutil"
+	"github.com/hashicorp/vault/sdk/helper/strutil"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -28,6 +30,33 @@ type Info struct {
 	CreationTime time.Time `json:"creation_time"`
 }
 
+func GetWrapper(ctx context.Context, path string) (wrapping.Wrapper, error) {
+	kmses, err := configutil.LoadConfigKMSes(path)
+	if err != nil {
+		return nil, fmt.Errorf("Error parsing config file: %w", err)
+	}
+
+	var kms *configutil.KMS
+	for _, v := range kmses {
+		if strutil.StrListContains(v.Purpose, "recovery") {
+			if kms != nil {
+				return nil, fmt.Errorf("Only one %q block marked for %q purpose is allowed", "kms", "recovery")
+			}
+			kms = v
+		}
+	}
+	if kms == nil {
+		return nil, fmt.Errorf("No %q block marked for %q purpose is allowed", "kms", "recovery")
+	}
+
+	wrapper, err := configutil.ConfigureWrapper(kms, nil, nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("Error configuring kms: %w", err)
+	}
+
+	return wrapper, nil
+}
+
 func GenerateRecoveryToken(ctx context.Context, wrapper wrapping.Wrapper) (string, error) {
 	b, err := uuid.GenerateRandomBytes(nonceLength)
 	if err != nil {