Enhance worker connection security (#1641)

This stores worker connection nonces within the database to ensure
uniqueness of connections. Prior the connection required any attacker to
be able to decrypt the incoming encrypted authentication message in
order to replay (and if the attacker has access to the KMS you have
bigger problems), but with this change even if an attacker does have
access to the KMS key, they cannot reuse the same message. (If they're
able to encrypt against the KMS key, they can still act as a worker with
a new message, so this only really helps if they only have decryption
capabilities.)

internal/servers/controller/controller.go

@@ -26,7 +26,6 @@ import (
 	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
-	"github.com/patrickmn/go-cache"
 	ua "go.uber.org/atomic"
 	"google.golang.org/grpc"
 )
@@ -39,10 +38,10 @@ type Controller struct {
 	baseCancel  context.CancelFunc
 	started     *ua.Bool
 
-	tickerWg    sync.WaitGroup
-	schedulerWg sync.WaitGroup
+	tickerWg    *sync.WaitGroup
+	schedulerWg *sync.WaitGroup
 
-	workerAuthCache *cache.Cache
+	workerAuthCache *sync.Map
 
 	// Used for testing and tracking worker health
 	workerStatusUpdateTimes *sync.Map
@@ -74,6 +73,9 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 		conf:                    conf,
 		logger:                  conf.Logger.Named("controller"),
 		started:                 ua.NewBool(false),
+		tickerWg:                new(sync.WaitGroup),
+		schedulerWg:             new(sync.WaitGroup),
+		workerAuthCache:         new(sync.Map),
 		workerStatusUpdateTimes: new(sync.Map),
 	}
 
@@ -163,7 +165,6 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 	c.SessionRepoFn = func() (*session.Repository, error) {
 		return session.NewRepository(dbase, dbase, c.kms)
 	}
-	c.workerAuthCache = cache.New(0, 0)
 
 	return c, nil
 }
@@ -178,7 +179,7 @@ func (c *Controller) Start() error {
 	if err := c.registerJobs(); err != nil {
 		return fmt.Errorf("error registering jobs: %w", err)
 	}
-	if err := c.scheduler.Start(c.baseContext, &c.schedulerWg); err != nil {
+	if err := c.scheduler.Start(c.baseContext, c.schedulerWg); err != nil {
 		return fmt.Errorf("error starting scheduler: %w", err)
 	}
 	if err := c.startListeners(c.baseContext); err != nil {
@@ -192,7 +193,7 @@ func (c *Controller) Start() error {
 	}()
 	go func() {
 		defer c.tickerWg.Done()
-		c.startRecoveryNonceCleanupTicking(c.baseContext)
+		c.startNonceCleanupTicking(c.baseContext)
 	}()
 	go func() {
 		defer c.tickerWg.Done()