Merge pull request #544 from nokia/cluster-client-tls

add configurable TLS cert for the clustering API client

Author: karimra

docs/user_guide/HA.md

@@ -123,6 +123,19 @@ clustering:
     renew-period: 5s
     # debug, enable extra logging messages
     debug: false
+  # tls config for the REST API client
+  tls:
+    # string, path to the CA certificate file,
+    # this will be used to verify the certificates of the gNMIc cluster members
+    # when `skip-verify` is false
+    ca-file:
+    # string, client certificate file.
+    cert-file:
+    # string, client key file.
+    key-file:
+    # boolean, if true, the client will not verify the server
+    # certificate against the available certificate chain.
+    skip-verify: false
 ```
 
 A `gnmic` instance creates gNMI subscriptions only towards targets for which it acquired locks. It is also responsible for maintaining that lock for the duration of the subscription.

pkg/app/app.go

@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"net/http"
 	"os"
 	"sort"
 	"strings"
@@ -76,8 +77,9 @@ type App struct {
 	targetsLockFn map[string]context.CancelFunc
 	rootDesc      desc.Descriptor
 	// end collector
-	router *mux.Router
-	locker lockers.Locker
+	router           *mux.Router
+	locker           lockers.Locker
+	clusteringClient *http.Client
 	// api
 	apiServices map[string]*lockers.Service
 	isLeader    bool

pkg/app/clustering.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/openconfig/gnmic/pkg/api/types"
+	"github.com/openconfig/gnmic/pkg/api/utils"
 	"github.com/openconfig/gnmic/pkg/lockers"
 )
 
@@ -31,6 +32,7 @@ const (
 	retryTimer         = 10 * time.Second
 	lockWaitTime       = 100 * time.Millisecond
 	apiServiceName     = "gnmic-api"
+	protocolTagName    = "__protocol"
 )
 
 var (
@@ -81,9 +83,9 @@ func (a *App) apiServiceRegistration() {
 	tags = append(tags, fmt.Sprintf("cluster-name=%s", a.Config.Clustering.ClusterName))
 	tags = append(tags, fmt.Sprintf("instance-name=%s", a.Config.Clustering.InstanceName))
 	if a.Config.APIServer.TLS != nil {
-		tags = append(tags, "protocol=https")
+		tags = append(tags, protocolTagName+"=https")
 	} else {
-		tags = append(tags, "protocol=http")
+		tags = append(tags, protocolTagName+"=http")
 	}
 	tags = append(tags, a.Config.Clustering.Tags...)
 
@@ -538,25 +540,13 @@ func (a *App) getHighestTagsMatches(tagsCount map[string]int) []string {
 }
 
 func (a *App) deleteTarget(ctx context.Context, name string) error {
+	err := a.createAPIClient()
+	if err != nil {
+		return err
+	}
 	errs := make([]error, 0, len(a.apiServices))
 	for _, s := range a.apiServices {
-		scheme := "http"
-		client := &http.Client{
-			Timeout: defaultHTTPClientTimeout,
-		}
-		for _, t := range s.Tags {
-			if strings.HasPrefix(t, "protocol=") {
-				scheme = strings.Split(t, "=")[1]
-				break
-			}
-		}
-		if scheme == "https" {
-			client.Transport = &http.Transport{
-				TLSClientConfig: &tls.Config{
-					InsecureSkipVerify: true,
-				},
-			}
-		}
+		scheme := a.getServiceScheme(s)
 		ctx, cancel := context.WithCancel(ctx)
 		defer cancel()
 		url := fmt.Sprintf("%s://%s/api/v1/config/targets/%s", scheme, s.Address, name)
@@ -567,7 +557,7 @@ func (a *App) deleteTarget(ctx context.Context, name string) error {
 			continue
 		}
 
-		rsp, err := client.Do(req)
+		rsp, err := a.clusteringClient.Do(req)
 		if err != nil {
 			rsp.Body.Close()
 			a.Logger.Printf("failed deleting target %q: %v", name, err)
@@ -590,29 +580,17 @@ func (a *App) assignTarget(ctx context.Context, tc *types.TargetConfig, service
 	if err != nil {
 		return err
 	}
-	scheme := "http"
-	client := &http.Client{
-		Timeout: defaultHTTPClientTimeout,
-	}
-	for _, t := range service.Tags {
-		if strings.HasPrefix(t, "protocol=") {
-			scheme = strings.Split(t, "=")[1]
-			break
-		}
-	}
-	if scheme == "https" {
-		client.Transport = &http.Transport{
-			TLSClientConfig: &tls.Config{
-				InsecureSkipVerify: true,
-			},
-		}
+	err = a.createAPIClient()
+	if err != nil {
+		return err
 	}
+	scheme := a.getServiceScheme(service)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("%s://%s/api/v1/config/targets", scheme, service.Address), buffer)
 	if err != nil {
 		return err
 	}
 	req.Header.Set("Content-Type", "application/json")
-	resp, err := client.Do(req)
+	resp, err := a.clusteringClient.Do(req)
 	if err != nil {
 		return err
 	}
@@ -626,7 +604,7 @@ func (a *App) assignTarget(ctx context.Context, tc *types.TargetConfig, service
 	if err != nil {
 		return err
 	}
-	resp, err = client.Do(req)
+	resp, err = a.clusteringClient.Do(req)
 	if err != nil {
 		return err
 	}
@@ -639,27 +617,15 @@ func (a *App) assignTarget(ctx context.Context, tc *types.TargetConfig, service
 }
 
 func (a *App) unassignTarget(ctx context.Context, name string, serviceID string) error {
+	err := a.createAPIClient()
+	if err != nil {
+		return err
+	}
 	for _, s := range a.apiServices {
 		if s.ID != serviceID {
 			continue
 		}
-		scheme := "http"
-		client := &http.Client{
-			Timeout: defaultHTTPClientTimeout,
-		}
-		for _, t := range s.Tags {
-			if strings.HasPrefix(t, "protocol=") {
-				scheme = strings.Split(t, "=")[1]
-				break
-			}
-		}
-		if scheme == "https" {
-			client.Transport = &http.Transport{
-				TLSClientConfig: &tls.Config{
-					InsecureSkipVerify: true,
-				},
-			}
-		}
+		scheme := a.getServiceScheme(s)
 		url := fmt.Sprintf("%s://%s/api/v1/targets/%s", scheme, s.Address, name)
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 		defer cancel()
@@ -668,7 +634,7 @@ func (a *App) unassignTarget(ctx context.Context, name string, serviceID string)
 			a.Logger.Printf("failed to create HTTP request: %v", err)
 			continue
 		}
-		rsp, err := client.Do(req)
+		rsp, err := a.clusteringClient.Do(req)
 		if err != nil {
 			// don't close the body here since Body will be nil
 			a.Logger.Printf("failed HTTP request: %v", err)
@@ -680,3 +646,49 @@ func (a *App) unassignTarget(ctx context.Context, name string, serviceID string)
 	}
 	return nil
 }
+
+func (a *App) getServiceScheme(service *lockers.Service) string {
+	scheme := "http"
+	for _, t := range service.Tags {
+		if strings.HasPrefix(t, protocolTagName+"=") {
+			scheme = strings.Split(t, "=")[1]
+			break
+		}
+	}
+	return scheme
+}
+
+func (a *App) createAPIClient() error {
+	if a.clusteringClient != nil {
+		return nil
+	}
+	// no certs
+	if a.Config.Clustering.TLS == nil {
+		a.clusteringClient = &http.Client{
+			Timeout: defaultHTTPClientTimeout,
+			Transport: &http.Transport{
+				TLSClientConfig: &tls.Config{
+					InsecureSkipVerify: true,
+				},
+			},
+		}
+		return nil
+	}
+	// with certs
+	tlsConfig, err := utils.NewTLSConfig(
+		a.Config.Clustering.TLS.CaFile,
+		a.Config.Clustering.TLS.CertFile,
+		a.Config.Clustering.TLS.KeyFile, "",
+		a.Config.Clustering.TLS.SkipVerify,
+		false)
+	if err != nil {
+		return err
+	}
+	a.clusteringClient = &http.Client{
+		Timeout: defaultHTTPClientTimeout,
+		Transport: &http.Transport{
+			TLSClientConfig: tlsConfig,
+		},
+	}
+	return nil
+}

pkg/config/clustering.go

@@ -9,10 +9,12 @@
 package config
 
 import (
+	"fmt"
 	"os"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/openconfig/gnmic/pkg/api/types"
 )
 
 const (
@@ -32,6 +34,7 @@ type clustering struct {
 	LeaderWaitTimer         time.Duration          `mapstructure:"leader-wait-timer,omitempty" json:"leader-wait-timer,omitempty" yaml:"leader-wait-timer,omitempty"`
 	Tags                    []string               `mapstructure:"tags,omitempty" json:"tags,omitempty" yaml:"tags,omitempty"`
 	Locker                  map[string]interface{} `mapstructure:"locker,omitempty" json:"locker,omitempty" yaml:"locker,omitempty"`
+	TLS                     *types.TLSConfig       `mapstructure:"tls,omitempty" json:"tls,omitempty" yaml:"tls,omitempty"`
 }
 
 func (c *Config) GetClustering() error {
@@ -50,6 +53,16 @@ func (c *Config) GetClustering() error {
 	for i := range c.Clustering.Tags {
 		c.Clustering.Tags[i] = os.ExpandEnv(c.Clustering.Tags[i])
 	}
+	if c.FileConfig.IsSet("clustering/tls") {
+		c.Clustering.TLS = new(types.TLSConfig)
+		c.Clustering.TLS.CaFile = os.ExpandEnv(c.FileConfig.GetString("clustering/tls/ca-file"))
+		c.Clustering.TLS.CertFile = os.ExpandEnv(c.FileConfig.GetString("clustering/tls/cert-file"))
+		c.Clustering.TLS.KeyFile = os.ExpandEnv(c.FileConfig.GetString("clustering/tls/key-file"))
+		c.Clustering.TLS.SkipVerify = os.ExpandEnv(c.FileConfig.GetString("clustering/tls/skip-verify")) == trueString
+		if err := c.APIServer.TLS.Validate(); err != nil {
+			return fmt.Errorf("clustering TLS config error: %w", err)
+		}
+	}
 	c.setClusteringDefaults()
 	return c.getLocker()
 }