Add Kubernetes service registration

command/commands.go

@@ -66,6 +66,7 @@ import (
 
 	sr "github.com/hashicorp/vault/serviceregistration"
 	csr "github.com/hashicorp/vault/serviceregistration/consul"
+	ksr "github.com/hashicorp/vault/serviceregistration/kubernetes"
 )
 
 const (
@@ -161,7 +162,8 @@ var (
 	}
 
 	serviceRegistrations = map[string]sr.Factory{
-		"consul": csr.NewServiceRegistration,
+		"consul":     csr.NewServiceRegistration,
+		"kubernetes": ksr.NewServiceRegistration,
 	}
 )
 

command/server.go

@@ -1279,7 +1279,7 @@ CLUSTER_SYNTHESIS_COMPLETE:
 
 	// If ServiceRegistration is configured, then the backend must support HA
 	isBackendHA := coreConfig.HAPhysical != nil && coreConfig.HAPhysical.HAEnabled()
-	if (coreConfig.ServiceRegistration != nil) && !isBackendHA {
+	if !c.flagDev && (coreConfig.ServiceRegistration != nil) && !isBackendHA {
 		c.UI.Output("service_registration is configured, but storage does not support HA")
 		return 1
 	}

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/aws/aws-sdk-go v1.25.41
 	github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932 // indirect
 	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
+	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/chrismalek/oktasdk-go v0.0.0-20181212195951-3430665dfaa0
 	github.com/cockroachdb/apd v1.1.0 // indirect
 	github.com/cockroachdb/cockroach-go v0.0.0-20181001143604-e0a95dfd547c

sdk/helper/certutil/certutil_test.go

@@ -403,6 +403,31 @@ func TestTLSConfig(t *testing.T) {
 	}
 }
 
+func TestNewCertPool(t *testing.T) {
+	caExample := `-----BEGIN CERTIFICATE-----
+MIIC5zCCAc+gAwIBAgIBATANBgkqhkiG9w0BAQsFADAVMRMwEQYDVQQDEwptaW5p
+a3ViZUNBMB4XDTE5MTIxMDIzMDUxOVoXDTI5MTIwODIzMDUxOVowFTETMBEGA1UE
+AxMKbWluaWt1YmVDQTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBANFi
+/RIdMHd865X6JygTb9riX01DA3QnR+RoXDXNnj8D3LziLG2n8ItXMJvWbU3sxxyy
+nX9HxJ0SIeexj1cYzdQBtJDjO1/PeuKc4CZ7zCukCAtHz8mC7BDPOU7F7pggpcQ0
+/t/pa2m22hmCu8aDF9WlUYHtJpYATnI/A5vz/VFLR9daxmkl59Qo3oHITj7vAzSx
+/75r9cibpQyJ+FhiHOZHQWYY2JYw2g4v5hm5hg5SFM9yFcZ75ISI9ebyFFIl9iBY
+zAk9jqv1mXvLr0Q39AVwMTamvGuap1oocjM9NIhQvaFL/DNqF1ouDQjCf5u2imLc
+TraO1/2KO8fqwOZCOrMCAwEAAaNCMEAwDgYDVR0PAQH/BAQDAgKkMB0GA1UdJQQW
+MBQGCCsGAQUFBwMCBggrBgEFBQcDATAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3
+DQEBCwUAA4IBAQBtVZCwCPqUUUpIClAlE9nc2fo2bTs9gsjXRmqdQ5oaSomSLE93
+aJWYFuAhxPXtlApbLYZfW2m1sM3mTVQN60y0uE4e1jdSN1ErYQ9slJdYDAMaEmOh
+iSexj+Nd1scUiMHV9lf3ps5J8sYeCpwZX3sPmw7lqZojTS12pANBDcigsaj5RRyN
+9GyP3WkSQUsTpWlDb9Fd+KNdkCVw7nClIpBPA2KW4BQKw/rNSvOFD61mbzc89lo0
+Q9IFGQFFF8jO18lbyWqnRBGXcS4/G7jQ3S7C121d14YLUeAYOM7pJykI1g4CLx9y
+vitin0L6nprauWkKO38XgM4T75qKZpqtiOcT
+-----END CERTIFICATE-----
+`
+	if _, err := NewCertPool(bytes.NewReader([]byte(caExample))); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func refreshRSA8CertBundle() *CertBundle {
 	initTest.Do(setCerts)
 	return &CertBundle{

sdk/helper/certutil/helpers.go

@@ -14,6 +14,8 @@ import (
 	"encoding/pem"
 	"errors"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"math/big"
 	"net"
 	"net/url"
@@ -804,3 +806,50 @@ func SignCertificate(data *CreationBundle) (*ParsedCertBundle, error) {
 
 	return result, nil
 }
+
+func NewCertPool(reader io.Reader) (*x509.CertPool, error) {
+	pemBlock, err := ioutil.ReadAll(reader)
+	if err != nil {
+		return nil, err
+	}
+	certs, err := parseCertsPEM(pemBlock)
+	if err != nil {
+		return nil, fmt.Errorf("error reading certs: %s", err)
+	}
+	pool := x509.NewCertPool()
+	for _, cert := range certs {
+		pool.AddCert(cert)
+	}
+	return pool, nil
+}
+
+// parseCertsPEM returns the x509.Certificates contained in the given PEM-encoded byte array
+// Returns an error if a certificate could not be parsed, or if the data does not contain any certificates
+func parseCertsPEM(pemCerts []byte) ([]*x509.Certificate, error) {
+	ok := false
+	certs := []*x509.Certificate{}
+	for len(pemCerts) > 0 {
+		var block *pem.Block
+		block, pemCerts = pem.Decode(pemCerts)
+		if block == nil {
+			break
+		}
+		// Only use PEM "CERTIFICATE" blocks without extra headers
+		if block.Type != "CERTIFICATE" || len(block.Headers) != 0 {
+			continue
+		}
+
+		cert, err := x509.ParseCertificate(block.Bytes)
+		if err != nil {
+			return certs, err
+		}
+
+		certs = append(certs, cert)
+		ok = true
+	}
+
+	if !ok {
+		return certs, errors.New("data does not contain any valid RSA or ECDSA certificates")
+	}
+	return certs, nil
+}

serviceregistration/kubernetes/client/client.go

@@ -0,0 +1,254 @@
+package client
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/hashicorp/go-cleanhttp"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-retryablehttp"
+)
+
+var (
+	// Retry configuration
+	RetryWaitMin = 500 * time.Millisecond
+	RetryWaitMax = 30 * time.Second
+	RetryMax     = 10
+
+	// Standard errs
+	ErrNamespaceUnset = errors.New(`"namespace" is unset`)
+	ErrPodNameUnset   = errors.New(`"podName" is unset`)
+	ErrNotInCluster   = errors.New("unable to load in-cluster configuration, KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT must be defined")
+)
+
+// New instantiates a Client. The stopCh is used for exiting retry loops
+// when closed.
+func New(logger hclog.Logger, stopCh <-chan struct{}) (*Client, error) {
+	config, err := inClusterConfig()
+	if err != nil {
+		return nil, err
+	}
+	return &Client{
+		logger: logger,
+		config: config,
+		stopCh: stopCh,
+	}, nil
+}
+
+// Client is a minimal Kubernetes client. We rolled our own because the existing
+// Kubernetes client-go library available externally has a high number of dependencies
+// and we thought it wasn't worth it for only two API calls. If at some point they break
+// the client into smaller modules, or if we add quite a few methods to this client, it may
+// be worthwhile to revisit that decision.
+type Client struct {
+	logger hclog.Logger
+	config *Config
+	stopCh <-chan struct{}
+}
+
+// GetPod gets a pod from the Kubernetes API.
+func (c *Client) GetPod(namespace, podName string) (*Pod, error) {
+	endpoint := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s", namespace, podName)
+	method := http.MethodGet
+
+	// Validate that we received required parameters.
+	if namespace == "" {
+		return nil, ErrNamespaceUnset
+	}
+	if podName == "" {
+		return nil, ErrPodNameUnset
+	}
+
+	req, err := http.NewRequest(method, c.config.Host+endpoint, nil)
+	if err != nil {
+		return nil, err
+	}
+	pod := &Pod{}
+	if err := c.do(req, pod); err != nil {
+		return nil, err
+	}
+	return pod, nil
+}
+
+// PatchPod updates the pod's tags to the given ones.
+// It does so non-destructively, or in other words, without tearing down
+// the pod.
+func (c *Client) PatchPod(namespace, podName string, patches ...*Patch) error {
+	endpoint := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s", namespace, podName)
+	method := http.MethodPatch
+
+	// Validate that we received required parameters.
+	if namespace == "" {
+		return ErrNamespaceUnset
+	}
+	if podName == "" {
+		return ErrPodNameUnset
+	}
+	if len(patches) == 0 {
+		// No work to perform.
+		return nil
+	}
+
+	var jsonPatches []map[string]interface{}
+	for _, patch := range patches {
+		if patch.Operation == Unset {
+			return errors.New("patch operation must be set")
+		}
+		jsonPatches = append(jsonPatches, map[string]interface{}{
+			"op":    patch.Operation,
+			"path":  patch.Path,
+			"value": patch.Value,
+		})
+	}
+	body, err := json.Marshal(jsonPatches)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequest(method, c.config.Host+endpoint, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json-patch+json")
+	return c.do(req, nil)
+}
+
+// do executes the given request, retrying if necessary.
+func (c *Client) do(req *http.Request, ptrToReturnObj interface{}) error {
+	// Finish setting up a valid request.
+	retryableReq, err := retryablehttp.FromRequest(req)
+	if err != nil {
+		return err
+	}
+
+	// Build a context that will call the cancelFunc when we receive
+	// a stop from our stopChan. This allows us to exit from our retry
+	// loop during a shutdown, rather than hanging.
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	go func(stopCh <-chan struct{}) {
+		<-stopCh
+		cancelFunc()
+	}(c.stopCh)
+	retryableReq.WithContext(ctx)
+
+	retryableReq.Header.Set("Authorization", "Bearer "+c.config.BearerToken)
+	retryableReq.Header.Set("Accept", "application/json")
+
+	client := &retryablehttp.Client{
+		HTTPClient:   cleanhttp.DefaultClient(),
+		RetryWaitMin: RetryWaitMin,
+		RetryWaitMax: RetryWaitMax,
+		RetryMax:     RetryMax,
+		CheckRetry:   c.getCheckRetry(req),
+		Backoff:      retryablehttp.DefaultBackoff,
+	}
+	client.HTTPClient.Transport = &http.Transport{
+		TLSClientConfig: &tls.Config{
+			RootCAs: c.config.CACertPool,
+		},
+	}
+
+	// Execute and retry the request. This client comes with exponential backoff and
+	// jitter already rolled in.
+	resp, err := client.Do(retryableReq)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err := resp.Body.Close(); err != nil {
+			if c.logger.IsWarn() {
+				// Failing to close response bodies can present as a memory leak so it's
+				// important to surface it.
+				c.logger.Warn(fmt.Sprintf("unable to close response body: %s", err))
+			}
+		}
+	}()
+
+	// If we're not supposed to read out the body, we have nothing further
+	// to do here.
+	if ptrToReturnObj == nil {
+		return nil
+	}
+
+	// Attempt to read out the body into the given return object.
+	return json.NewDecoder(resp.Body).Decode(ptrToReturnObj)
+}
+
+func (c *Client) getCheckRetry(req *http.Request) retryablehttp.CheckRetry {
+	return func(ctx context.Context, resp *http.Response, err error) (bool, error) {
+		if resp == nil {
+			return true, fmt.Errorf("nil response: %s", req.URL.RequestURI())
+		}
+		switch resp.StatusCode {
+		case 200, 201, 202, 204:
+			// Success.
+			return false, nil
+		case 401, 403:
+			// Perhaps the token from our bearer token file has been refreshed.
+			config, err := inClusterConfig()
+			if err != nil {
+				return false, err
+			}
+			if config.BearerToken == c.config.BearerToken {
+				// It's the same token.
+				return false, fmt.Errorf("bad status code: %s", sanitizedDebuggingInfo(req, resp.StatusCode))
+			}
+			c.config = config
+			// Continue to try again, but return the error too in case the caller would rather read it out.
+			return true, fmt.Errorf("bad status code: %s", sanitizedDebuggingInfo(req, resp.StatusCode))
+		case 404:
+			return false, &ErrNotFound{debuggingInfo: sanitizedDebuggingInfo(req, resp.StatusCode)}
+		case 500, 502, 503, 504:
+			// Could be transient.
+			return true, fmt.Errorf("unexpected status code: %s", sanitizedDebuggingInfo(req, resp.StatusCode))
+		}
+		// Unexpected.
+		return false, fmt.Errorf("unexpected status code: %s", sanitizedDebuggingInfo(req, resp.StatusCode))
+	}
+}
+
+type Pod struct {
+	Metadata *Metadata `json:"metadata,omitempty"`
+}
+
+type Metadata struct {
+	Name string `json:"name,omitempty"`
+
+	// This map will be nil if no "labels" key was provided.
+	// It will be populated but have a length of zero if the
+	// key was provided, but no values.
+	Labels map[string]string `json:"labels,omitempty"`
+}
+
+type PatchOperation string
+
+const (
+	Unset   PatchOperation = "unset"
+	Add                    = "add"
+	Replace                = "replace"
+)
+
+type Patch struct {
+	Operation PatchOperation
+	Path      string
+	Value     interface{}
+}
+
+type ErrNotFound struct {
+	debuggingInfo string
+}
+
+func (e *ErrNotFound) Error() string {
+	return e.debuggingInfo
+}
+
+// sanitizedDebuggingInfo provides a returnable string that can be used for debugging. This is intentionally somewhat vague
+// because we don't want to leak secrets that may be in a request or response body.
+func sanitizedDebuggingInfo(req *http.Request, respStatus int) string {
+	return fmt.Sprintf("req method: %s, req url: %s, resp statuscode: %d", req.Method, req.URL, respStatus)
+}