Agent auto auth wrapping new config checks (#6479)

* Simplify Run(): the function that was being sent over a channel doesn't
need to close over anything except latestToken, and we don't need to
create a new one each iteration.  Instead just pass the relevant items,
namely the token and sink to work on.

* Disallow the following config combinations:
1. auto_auth.method.wrap_ttl > 0 and multiple file sinks
2. auto_auth.method.wrap_ttl > 0 and single file sink with wrap_ttl > 0
3. auto_auth.method.wrap_ttl > 0 and cache.use_auto_auth_token = true

* Expose errors that occur when APIProxy is forwarding request to Vault.

* Fix merge issues.

.gitignore

@@ -47,14 +47,8 @@ Vagrantfile
 
 # Configs
 *.hcl
-!command/agent/config/test-fixtures/config.hcl
-!command/agent/config/test-fixtures/config-cache.hcl
-!command/agent/config/test-fixtures/config-embedded-type.hcl
-!command/agent/config/test-fixtures/config-cache-embedded-type.hcl
-!command/agent/config/test-fixtures/bad-config-cache-inconsistent-auto_auth.hcl
-!command/agent/config/test-fixtures/bad-config-cache-no-listeners.hcl
-!command/agent/config/test-fixtures/config-cache-no-auto_auth.hcl
-!command/agent/config/test-fixtures/config-cache-auto_auth-no-sink.hcl
+!command/agent/config/test-fixtures/*.hcl
+
 
 .DS_Store
 .idea

command/agent/cache/api_proxy.go

@@ -45,6 +45,9 @@ func (ap *APIProxy) Send(ctx context.Context, req *SendRequest) (*SendResponse,
 	ap.logger.Info("forwarding request", "path", req.Request.URL.Path, "method", req.Request.Method)
 
 	resp, err := client.RawRequestWithContext(ctx, fwReq)
+	if resp == nil && err != nil {
+		return nil, err
+	}
 
 	// Before error checking from the request call, we'd want to initialize a SendResponse to
 	// potentially return

command/agent/cache/proxy.go

@@ -3,7 +3,6 @@ package cache
 import (
 	"bytes"
 	"context"
-	"fmt"
 	"io/ioutil"
 	"net/http"
 	"time"
@@ -49,10 +48,6 @@ type Proxier interface {
 // NewSendResponse creates a new SendResponse and takes care of initializing its
 // fields properly.
 func NewSendResponse(apiResponse *api.Response, responseBody []byte) (*SendResponse, error) {
-	if apiResponse == nil {
-		return nil, fmt.Errorf("nil api response provided")
-	}
-
 	resp := &SendResponse{
 		Response:  apiResponse,
 		CacheMeta: &CacheMeta{},

command/agent/config/config.go

@@ -127,9 +127,20 @@ func LoadConfig(path string, logger log.Logger) (*Config, error) {
 			return nil, fmt.Errorf("at least one listener required when cache enabled")
 		}
 
-		if result.Cache.UseAutoAuthToken && result.AutoAuth == nil {
+		if result.Cache.UseAutoAuthToken {
+			if result.AutoAuth == nil {
 				return nil, fmt.Errorf("cache.use_auto_auth_token is true but auto_auth not configured")
 			}
+			if result.AutoAuth.Method.WrapTTL > 0 {
+				return nil, fmt.Errorf("cache.use_auto_auth_token is true and auto_auth uses wrapping")
+			}
+		}
+	}
+
+	if result.AutoAuth != nil {
+		if len(result.AutoAuth.Sinks) == 0 && (result.Cache == nil || !result.Cache.UseAutoAuthToken) {
+			return nil, fmt.Errorf("auto_auth requires at least one sink or cache.use_auto_auth_token=true ")
+		}
 	}
 
 	err = parseVault(&result, list)
@@ -268,6 +279,16 @@ func parseAutoAuth(result *Config, list *ast.ObjectList) error {
 		return errwrap.Wrapf("error parsing 'sink' stanzas: {{err}}", err)
 	}
 
+	if result.AutoAuth.Method.WrapTTL > 0 {
+		if len(result.AutoAuth.Sinks) != 1 {
+			return fmt.Errorf("error parsing auto_auth: wrapping enabled on auth method and 0 or many sinks defined")
+		}
+
+		if result.AutoAuth.Sinks[0].WrapTTL > 0 {
+			return fmt.Errorf("error parsing auto_auth: wrapping enabled both on auth method and sink")
+		}
+	}
+
 	return nil
 }
 

command/agent/config/config_test.go

@@ -22,7 +22,6 @@ func TestLoadConfigFile_AgentCache(t *testing.T) {
 		AutoAuth: &AutoAuth{
 			Method: &Method{
 				Type:      "aws",
-				WrapTTL:   300 * time.Second,
 				MountPath: "auth/aws",
 				Config: map[string]interface{}{
 					"role": "foobar",
@@ -110,7 +109,6 @@ func TestLoadConfigFile(t *testing.T) {
 		AutoAuth: &AutoAuth{
 			Method: &Method{
 				Type:      "aws",
-				WrapTTL:   300 * time.Second,
 				MountPath: "auth/aws",
 				Config: map[string]interface{}{
 					"role": "foobar",
@@ -155,6 +153,41 @@ func TestLoadConfigFile(t *testing.T) {
 	}
 }
 
+func TestLoadConfigFile_Method_Wrapping(t *testing.T) {
+	logger := logging.NewVaultLogger(log.Debug)
+
+	config, err := LoadConfig("./test-fixtures/config-method-wrapping.hcl", logger)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	expected := &Config{
+		AutoAuth: &AutoAuth{
+			Method: &Method{
+				Type:      "aws",
+				MountPath: "auth/aws",
+				WrapTTL:   5 * time.Minute,
+				Config: map[string]interface{}{
+					"role": "foobar",
+				},
+			},
+			Sinks: []*Sink{
+				&Sink{
+					Type: "file",
+					Config: map[string]interface{}{
+						"path": "/tmp/file-foo",
+					},
+				},
+			},
+		},
+		PidFile: "./pidfile",
+	}
+
+	if diff := deep.Equal(config, expected); diff != nil {
+		t.Fatal(diff)
+	}
+}
+
 func TestLoadConfigFile_AgentCache_NoAutoAuth(t *testing.T) {
 	logger := logging.NewVaultLogger(log.Debug)
 
@@ -200,6 +233,33 @@ func TestLoadConfigFile_Bad_AgentCache_NoListeners(t *testing.T) {
 	}
 }
 
+func TestLoadConfigFile_Bad_AutoAuth_Wrapped_Multiple_Sinks(t *testing.T) {
+	logger := logging.NewVaultLogger(log.Debug)
+
+	_, err := LoadConfig("./test-fixtures/bad-config-auto_auth-wrapped-multiple-sinks", logger)
+	if err == nil {
+		t.Fatal("LoadConfig should return an error when auth_auth.method.wrap_ttl nonzero and multiple sinks defined")
+	}
+}
+
+func TestLoadConfigFile_Bad_AutoAuth_Both_Wrapping_Types(t *testing.T) {
+	logger := logging.NewVaultLogger(log.Debug)
+
+	_, err := LoadConfig("./test-fixtures/bad-config-method-wrapping-and-sink-wrapping.hcl", logger)
+	if err == nil {
+		t.Fatal("LoadConfig should return an error when auth_auth.method.wrap_ttl nonzero and sinks.wrap_ttl nonzero")
+	}
+}
+
+func TestLoadConfigFile_Bad_AgentCache_AutoAuth_Method_wrapping(t *testing.T) {
+	logger := logging.NewVaultLogger(log.Debug)
+
+	_, err := LoadConfig("./test-fixtures/bad-config-cache-auto_auth-method-wrapping.hcl", logger)
+	if err == nil {
+		t.Fatal("LoadConfig should return an error when auth_auth.method.wrap_ttl nonzero and cache.use_auto_auth_token=true")
+	}
+}
+
 func TestLoadConfigFile_AgentCache_AutoAuth_NoSink(t *testing.T) {
 	logger := logging.NewVaultLogger(log.Debug)
 
@@ -212,7 +272,6 @@ func TestLoadConfigFile_AgentCache_AutoAuth_NoSink(t *testing.T) {
 		AutoAuth: &AutoAuth{
 			Method: &Method{
 				Type:      "aws",
-				WrapTTL:   300 * time.Second,
 				MountPath: "auth/aws",
 				Config: map[string]interface{}{
 					"role": "foobar",

command/agent/config/test-fixtures/bad-config-auto_auth-wrapped-multiple-sinks.hcl

@@ -0,0 +1,23 @@
+pid_file = "./pidfile"
+
+auto_auth {
+  method "aws" {
+    mount_path = "auth/aws"
+    wrap_ttl = 300
+    config = {
+      role = "foobar"
+    }
+  }
+
+  sink "file" {
+    config = {
+      path = "/tmp/file-foo"
+    }
+  }
+
+  sink "file" {
+    config = {
+      path = "/tmp/file-bar"
+    }
+  }
+}

command/agent/config/test-fixtures/bad-config-cache-auto_auth-method-wrapping.hcl

@@ -0,0 +1,29 @@
+pid_file = "./pidfile"
+
+auto_auth {
+	method {
+		type = "aws"
+		wrap_ttl = 300
+		config = {
+			role = "foobar"
+		}
+	}
+
+	sink {
+		type = "file"
+		config = {
+			path = "/tmp/file-foo"
+		}
+	}
+}
+
+cache {
+	use_auto_auth_token = true
+}
+
+listener "tcp" {
+    address = "127.0.0.1:8300"
+    tls_disable = true
+}
+
+

command/agent/config/test-fixtures/bad-config-method-wrapping-and-sink-wrapping.hcl

@@ -0,0 +1,19 @@
+pid_file = "./pidfile"
+
+auto_auth {
+	method {
+		type = "aws"
+		wrap_ttl = 300
+		config = {
+			role = "foobar"
+		}
+	}
+
+	sink {
+		type = "file"
+		wrap_ttl = 300
+		config = {
+			path = "/tmp/file-foo"
+		}
+	}
+}

command/agent/config/test-fixtures/config-cache-auto_auth-no-sink.hcl

@@ -3,7 +3,6 @@ pid_file = "./pidfile"
 auto_auth {
 	method {
 		type = "aws"
-		wrap_ttl = 300
 		config = {
 			role = "foobar"
 		}

command/agent/config/test-fixtures/config-cache-embedded-type.hcl

@@ -3,7 +3,6 @@ pid_file = "./pidfile"
 auto_auth {
 	method {
 		type = "aws"
-		wrap_ttl = 300
 		config = {
 			role = "foobar"
 		}

command/agent/config/test-fixtures/config-cache.hcl

@@ -3,7 +3,6 @@ pid_file = "./pidfile"
 auto_auth {
 	method {
 		type = "aws"
-		wrap_ttl = 300
 		config = {
 			role = "foobar"
 		}

command/agent/config/test-fixtures/config-embedded-type.hcl

@@ -3,7 +3,6 @@ pid_file = "./pidfile"
 auto_auth {
 	method "aws" {
 		mount_path = "auth/aws"
-		wrap_ttl = 300
 		config = {
 			role = "foobar"
 		}

command/agent/config/test-fixtures/config-method-wrapping.hcl

@@ -0,0 +1,18 @@
+pid_file = "./pidfile"
+
+auto_auth {
+	method {
+		type = "aws"
+		wrap_ttl = 300
+		config = {
+			role = "foobar"
+		}
+	}
+
+	sink {
+		type = "file"
+		config = {
+			path = "/tmp/file-foo"
+		}
+	}
+}

command/agent/config/test-fixtures/config.hcl

@@ -3,7 +3,6 @@ pid_file = "./pidfile"
 auto_auth {
 	method {
 		type = "aws"
-		wrap_ttl = 300
 		config = {
 			role = "foobar"
 		}

command/agent/sink/sink.go

@@ -71,8 +71,30 @@ func NewSinkServer(conf *SinkServerConfig) *SinkServer {
 // Run executes the server's run loop, which is responsible for reading
 // in new tokens and pushing them out to the various sinks.
 func (ss *SinkServer) Run(ctx context.Context, incoming chan string, sinks []*SinkConfig) {
+	latestToken := new(string)
+	writeSink := func(currSink *SinkConfig, currToken string) error {
+		if currToken != *latestToken {
+			return nil
+		}
+		var err error
+
+		if currSink.WrapTTL != 0 {
+			if currToken, err = currSink.wrapToken(ss.client, currSink.WrapTTL, currToken); err != nil {
+				return err
+			}
+		}
+
+		if currSink.DHType != "" {
+			if currToken, err = currSink.encryptToken(currToken); err != nil {
+				return err
+			}
+		}
+
+		return currSink.WriteToken(currToken)
+	}
+
 	if incoming == nil {
-		panic("incoming or shutdown channel are nil")
+		panic("incoming channel is nil")
 	}
 
 	ss.logger.Info("starting sink server")
@@ -81,8 +103,11 @@ func (ss *SinkServer) Run(ctx context.Context, incoming chan string, sinks []*Si
 		close(ss.DoneCh)
 	}()
 
-	latestToken := new(string)
-	sinkCh := make(chan func() error, len(sinks))
+	type sinkToken struct {
+		sink  *SinkConfig
+		token string
+	}
+	sinkCh := make(chan sinkToken, len(sinks))
 	for {
 		select {
 		case <-ctx.Done():
@@ -104,36 +129,13 @@ func (ss *SinkServer) Run(ctx context.Context, incoming chan string, sinks []*Si
 
 				*latestToken = token
 
-				sinkFunc := func(currSink *SinkConfig, currToken string) func() error {
-					return func() error {
-						if currToken != *latestToken {
-							return nil
-						}
-						var err error
-
-						if currSink.WrapTTL != 0 {
-							if currToken, err = currSink.wrapToken(ss.client, currSink.WrapTTL, currToken); err != nil {
-								return err
-							}
-						}
-
-						if currSink.DHType != "" {
-							if currToken, err = currSink.encryptToken(currToken); err != nil {
-								return err
-							}
-						}
-
-						return currSink.WriteToken(currToken)
-					}
-				}
-
 				for _, s := range sinks {
 					atomic.AddInt32(ss.remaining, 1)
-					sinkCh <- sinkFunc(s, token)
+					sinkCh <- sinkToken{s, token}
 				}
 			}
 
-		case sinkFunc := <-sinkCh:
+		case st := <-sinkCh:
 			atomic.AddInt32(ss.remaining, -1)
 			select {
 			case <-ctx.Done():
@@ -141,7 +143,7 @@ func (ss *SinkServer) Run(ctx context.Context, incoming chan string, sinks []*Si
 			default:
 			}
 
-			if err := sinkFunc(); err != nil {
+			if err := writeSink(st.sink, st.token); err != nil {
 				backoff := 2*time.Second + time.Duration(ss.random.Int63()%int64(time.Second*2)-int64(time.Second))
 				ss.logger.Error("error returned by sink function, retrying", "error", err, "backoff", backoff.String())
 				select {
@@ -149,7 +151,7 @@ func (ss *SinkServer) Run(ctx context.Context, incoming chan string, sinks []*Si
 					return
 				case <-time.After(backoff):
 					atomic.AddInt32(ss.remaining, 1)
-					sinkCh <- sinkFunc
+					sinkCh <- st
 				}
 			} else {
 				if atomic.LoadInt32(ss.remaining) == 0 && ss.exitAfterAuth {