Vault Agent Cache (#6220)

* vault-agent-cache: squashed 250+ commits

* Add proper token revocation validations to the tests

* Add more test cases

* Avoid leaking by not closing request/response bodies; add comments

* Fix revoke orphan use case; update tests

* Add CLI test for making request over unix socket

* agent/cache: remove namespace-related tests

* Strip-off the auto-auth token from the lookup response

* Output listener details along with configuration

* Add scheme to API address output

* leasecache: use IndexNameLease for prefix lease revocations

* Make CLI accept the fully qualified unix address

* export VAULT_AGENT_ADDR=unix://path/to/socket

* unix:/ to unix://

command/agent.go

@@ -4,6 +4,10 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
+	"net/http"
+	"time"
+
 	"os"
 	"sort"
 	"strings"
@@ -23,6 +27,7 @@ import (
 	"github.com/hashicorp/vault/command/agent/auth/gcp"
 	"github.com/hashicorp/vault/command/agent/auth/jwt"
 	"github.com/hashicorp/vault/command/agent/auth/kubernetes"
+	"github.com/hashicorp/vault/command/agent/cache"
 	"github.com/hashicorp/vault/command/agent/config"
 	"github.com/hashicorp/vault/command/agent/sink"
 	"github.com/hashicorp/vault/command/agent/sink/file"
@@ -218,19 +223,6 @@ func (c *AgentCommand) Run(args []string) int {
 		info["cgo"] = "enabled"
 	}
 
-	// Server configuration output
-	padding := 24
-	sort.Strings(infoKeys)
-	c.UI.Output("==> Vault agent configuration:\n")
-	for _, k := range infoKeys {
-		c.UI.Output(fmt.Sprintf(
-			"%s%s: %s",
-			strings.Repeat(" ", padding-len(k)),
-			strings.Title(k),
-			info[k]))
-	}
-	c.UI.Output("")
-
 	// Tests might not want to start a vault server and just want to verify
 	// the configuration.
 	if c.flagTestVerifyOnly {
@@ -332,10 +324,92 @@ func (c *AgentCommand) Run(args []string) int {
 		EnableReauthOnNewCredentials: config.AutoAuth.EnableReauthOnNewCredentials,
 	})
 
-	// Start things running
+	// Start auto-auth and sink servers
 	go ah.Run(ctx, method)
 	go ss.Run(ctx, ah.OutputCh, sinks)
 
+	// Parse agent listener configurations
+	if config.Cache != nil && len(config.Cache.Listeners) != 0 {
+		cacheLogger := c.logger.Named("cache")
+
+		// Create the API proxier
+		apiProxy := cache.NewAPIProxy(&cache.APIProxyConfig{
+			Logger: cacheLogger.Named("apiproxy"),
+		})
+
+		// Create the lease cache proxier and set its underlying proxier to
+		// the API proxier.
+		leaseCache, err := cache.NewLeaseCache(&cache.LeaseCacheConfig{
+			BaseContext: ctx,
+			Proxier:     apiProxy,
+			Logger:      cacheLogger.Named("leasecache"),
+		})
+		if err != nil {
+			c.UI.Error(fmt.Sprintf("Error creating lease cache: %v", err))
+			return 1
+		}
+
+		// Create a muxer and add paths relevant for the lease cache layer
+		mux := http.NewServeMux()
+		mux.Handle("/v1/agent/cache-clear", leaseCache.HandleCacheClear(ctx))
+
+		mux.Handle("/", cache.Handler(ctx, cacheLogger, leaseCache, config.Cache.UseAutoAuthToken, c.client))
+
+		var listeners []net.Listener
+		for i, lnConfig := range config.Cache.Listeners {
+			listener, props, _, err := cache.ServerListener(lnConfig, c.logWriter, c.UI)
+			if err != nil {
+				c.UI.Error(fmt.Sprintf("Error parsing listener configuration: %v", err))
+				return 1
+			}
+
+			listeners = append(listeners, listener)
+
+			scheme := "https://"
+			if props["tls"] == "disabled" {
+				scheme = "http://"
+			}
+			if lnConfig.Type == "unix" {
+				scheme = "unix://"
+			}
+
+			infoKey := fmt.Sprintf("api address %d", i+1)
+			info[infoKey] = scheme + listener.Addr().String()
+			infoKeys = append(infoKeys, infoKey)
+
+			cacheLogger.Info("starting listener", "addr", listener.Addr().String())
+			server := &http.Server{
+				Handler:           mux,
+				ReadHeaderTimeout: 10 * time.Second,
+				ReadTimeout:       30 * time.Second,
+				IdleTimeout:       5 * time.Minute,
+				ErrorLog:          cacheLogger.StandardLogger(nil),
+			}
+			go server.Serve(listener)
+		}
+
+		// Ensure that listeners are closed at all the exits
+		listenerCloseFunc := func() {
+			for _, ln := range listeners {
+				ln.Close()
+			}
+		}
+		defer c.cleanupGuard.Do(listenerCloseFunc)
+	}
+
+	// Server configuration output
+	padding := 24
+	sort.Strings(infoKeys)
+	c.UI.Output("==> Vault agent configuration:\n")
+	for _, k := range infoKeys {
+		c.UI.Output(fmt.Sprintf(
+			"%s%s: %s",
+			strings.Repeat(" ", padding-len(k)),
+			strings.Title(k),
+			info[k]))
+	}
+	c.UI.Output("")
+
 	// Release the log gate.
 	c.logGate.Flush()