From 0e0782cf0bc388e7a1b664486c40c7e60ed53e64 Mon Sep 17 00:00:00 2001 From: Dennis Lanov Date: Thu, 30 Jul 2026 14:48:08 -0500 Subject: [PATCH] collector: honor clustering TLS for cluster API clients Collector-mode cluster dispatch used plain HTTP clients, so leader-to-member API calls ignored clustering TLS settings and could not use private CAs or present client certificates. Build the shared dispatch client from the configured clustering TLS settings after environment expansion. Preserve the existing timeout and default transport behavior when TLS is not configured. Signed-off-by: Dennis Lanov --- pkg/collector/managers/cluster/api_client.go | 34 ++++ .../managers/cluster/api_client_test.go | 189 ++++++++++++++++++ pkg/collector/managers/cluster/assigner.go | 14 +- .../managers/cluster/cluster_manager.go | 9 +- 4 files changed, 240 insertions(+), 6 deletions(-) create mode 100644 pkg/collector/managers/cluster/api_client.go create mode 100644 pkg/collector/managers/cluster/api_client_test.go diff --git a/pkg/collector/managers/cluster/api_client.go b/pkg/collector/managers/cluster/api_client.go new file mode 100644 index 00000000..d533dc13 --- /dev/null +++ b/pkg/collector/managers/cluster/api_client.go @@ -0,0 +1,34 @@ +package cluster_manager + +import ( + "net/http" + "time" + + "github.com/openconfig/gnmic/pkg/api/utils" + "github.com/openconfig/gnmic/pkg/config" +) + +const apiClientTimeout = 10 * time.Second + +// newAPIClient builds the HTTP client used for leader to member API calls, +// honoring the clustering TLS configuration. +func newAPIClient(clusteringConfig *config.Clustering) (*http.Client, error) { + if clusteringConfig == nil || clusteringConfig.TLS == nil { + return &http.Client{Timeout: apiClientTimeout}, nil + } + tlsConfig, err := utils.NewTLSConfig( + clusteringConfig.TLS.CaFile, + clusteringConfig.TLS.CertFile, + clusteringConfig.TLS.KeyFile, "", + clusteringConfig.TLS.SkipVerify, + false) + if err != nil { + return nil, err + } + tr := http.DefaultTransport.(*http.Transport).Clone() + tr.TLSClientConfig = tlsConfig + return &http.Client{ + Timeout: apiClientTimeout, + Transport: tr, + }, nil +} diff --git a/pkg/collector/managers/cluster/api_client_test.go b/pkg/collector/managers/cluster/api_client_test.go new file mode 100644 index 00000000..d5f48fda --- /dev/null +++ b/pkg/collector/managers/cluster/api_client_test.go @@ -0,0 +1,189 @@ +package cluster_manager + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "github.com/openconfig/gnmic/pkg/api/types" + "github.com/openconfig/gnmic/pkg/config" +) + +func TestNewAPIClient_noTLSConfig(t *testing.T) { + for name, cfg := range map[string]*config.Clustering{ + "nil clustering": nil, + "nil tls": {ClusterName: "lab"}, + } { + t.Run(name, func(t *testing.T) { + client, err := newAPIClient(cfg) + if err != nil { + t.Fatalf("newAPIClient: %v", err) + } + if client.Timeout != apiClientTimeout { + t.Fatalf("timeout = %s, want %s", client.Timeout, apiClientTimeout) + } + if client.Transport != nil { + t.Fatalf("expected default transport, got %#v", client.Transport) + } + }) + } +} + +func TestNewAPIClient_invalidCert(t *testing.T) { + _, err := newAPIClient(&config.Clustering{ + TLS: &types.TLSConfig{ + CertFile: filepath.Join(t.TempDir(), "missing.pem"), + KeyFile: filepath.Join(t.TempDir(), "missing-key.pem"), + }, + }) + if err == nil { + t.Fatal("expected error for missing cert files") + } +} + +func TestNewAPIClient_mTLS(t *testing.T) { + caFile, certFile, keyFile, serverCert, caPool := writeTestCerts(t) + + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + srv.TLS = &tls.Config{ + Certificates: []tls.Certificate{serverCert}, + ClientCAs: caPool, + ClientAuth: tls.RequireAndVerifyClientCert, + } + srv.StartTLS() + t.Cleanup(srv.Close) + + client, err := newAPIClient(&config.Clustering{ + TLS: &types.TLSConfig{ + CaFile: caFile, + CertFile: certFile, + KeyFile: keyFile, + }, + }) + if err != nil { + t.Fatalf("newAPIClient: %v", err) + } + resp, err := client.Get(srv.URL) + if err != nil { + t.Fatalf("request with clustering TLS certs failed: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d", resp.StatusCode) + } + + // without a client certificate the member API must reject the request + noCertClient, err := newAPIClient(&config.Clustering{ + TLS: &types.TLSConfig{CaFile: caFile}, + }) + if err != nil { + t.Fatalf("newAPIClient: %v", err) + } + resp, err = noCertClient.Get(srv.URL) + if err == nil { + resp.Body.Close() + t.Fatal("expected request without client certificate to fail") + } +} + +// writeTestCerts generates a CA, a server certificate and a client certificate, +// writes the CA and client pair to disk and returns their paths together with +// the server certificate and a pool containing the CA. +func writeTestCerts(t *testing.T) (caFile, certFile, keyFile string, serverCert tls.Certificate, caPool *x509.CertPool) { + t.Helper() + dir := t.TempDir() + + caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + caTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + BasicConstraintsValid: true, + IsCA: true, + } + caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caKey.PublicKey, caKey) + if err != nil { + t.Fatal(err) + } + caCert, err := x509.ParseCertificate(caDER) + if err != nil { + t.Fatal(err) + } + + newCert := func(template *x509.Certificate) (certPEM, keyPEM []byte) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + der, err := x509.CreateCertificate(rand.Reader, template, caCert, &key.PublicKey, caKey) + if err != nil { + t.Fatal(err) + } + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + t.Fatal(err) + } + certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}) + return certPEM, keyPEM + } + + serverCertPEM, serverKeyPEM := newCert(&x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: pkix.Name{CommonName: "server"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}, + }) + clientCertPEM, clientKeyPEM := newCert(&x509.Certificate{ + SerialNumber: big.NewInt(3), + Subject: pkix.Name{CommonName: "client"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + }) + + serverCert, err = tls.X509KeyPair(serverCertPEM, serverKeyPEM) + if err != nil { + t.Fatal(err) + } + caPool = x509.NewCertPool() + caPool.AddCert(caCert) + + caFile = filepath.Join(dir, "ca.pem") + certFile = filepath.Join(dir, "client-cert.pem") + keyFile = filepath.Join(dir, "client-key.pem") + caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER}) + for f, b := range map[string][]byte{ + caFile: caPEM, + certFile: clientCertPEM, + keyFile: clientKeyPEM, + } { + if err := os.WriteFile(f, b, 0o600); err != nil { + t.Fatal(err) + } + } + return caFile, certFile, keyFile, serverCert, caPool +} diff --git a/pkg/collector/managers/cluster/assigner.go b/pkg/collector/managers/cluster/assigner.go index a0ac5170..cf1c50ca 100644 --- a/pkg/collector/managers/cluster/assigner.go +++ b/pkg/collector/managers/cluster/assigner.go @@ -8,7 +8,6 @@ import ( "io" "log/slog" "net/http" - "time" apiconst "github.com/openconfig/gnmic/pkg/collector/api/const" collstore "github.com/openconfig/gnmic/pkg/collector/store" @@ -42,12 +41,19 @@ type restAssigner struct { } func NewAssigner(store *collstore.Store) Assigner { + return newAssigner(store, nil) +} + +func newAssigner(store *collstore.Store, client *http.Client) Assigner { + if client == nil { + client = &http.Client{ + Timeout: apiClientTimeout, + } + } return &restAssigner{ store: store, logger: slog.With("component", "assignment-pusher"), - client: &http.Client{ - Timeout: 10 * time.Second, - }, + client: client, } } diff --git a/pkg/collector/managers/cluster/cluster_manager.go b/pkg/collector/managers/cluster/cluster_manager.go index 8df80fe9..bf936481 100644 --- a/pkg/collector/managers/cluster/cluster_manager.go +++ b/pkg/collector/managers/cluster/cluster_manager.go @@ -66,7 +66,7 @@ func NewClusterManager(store *collstore.Store) *ClusterManager { locker: nil, lockCheckLimiter: make(chan struct{}, 64), // TODO: make this configurable rebalancingSem: semaphore.NewWeighted(1), - apiClient: &http.Client{Timeout: 10 * time.Second}, // TODO: + apiClient: &http.Client{Timeout: apiClientTimeout}, } } @@ -94,6 +94,11 @@ func (c *ClusterManager) Start(ctx context.Context, locker lockers.Locker, wg *s c.clusteringConfig = clustering env.ExpandClusterEnv(c.clusteringConfig) + c.apiClient, err = newAPIClient(c.clusteringConfig) + if err != nil { + return fmt.Errorf("failed to create cluster API client: %w", err) + } + apiConfig, ok, err := c.store.Config.Get("api-server", "api-server") if err != nil { return err @@ -118,7 +123,7 @@ func (c *ClusterManager) Start(ctx context.Context, locker lockers.Locker, wg *s return err } c.membership = NewMembership(c.locker, clustering, c.logger) - c.assigner = NewAssigner(c.store) + c.assigner = newAssigner(c.store, c.apiClient) // start registration to register the api service wg.Add(1)