Reload client CAs in server TLS settings on file change (open-telemet…

…ry#6708)

Reload client CAs in server TLS settings when file has been modified

Link to tracking Issue: open-telemetry#6524
---------

Co-authored-by: Alex Boten <alex@boten.ca>
Co-authored-by: Ben B. <bongartz@klimlive.de>

.chloggen/reload-clientca-on-file-change.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: configtls
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Reload mTLS ClientCA certificates on file change
+
+# One or more tracking issues or pull requests related to the change
+issues: [6524]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

cmd/otelcorecol/go.mod

@@ -29,6 +29,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
+	github.com/fsnotify/fsnotify v1.4.9 // indirect
 	github.com/go-kit/log v0.2.1 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect

config/configgrpc/configgrpc_test.go

@@ -453,7 +453,7 @@ func TestGRPCServerSettingsError(t *testing.T) {
 			},
 		},
 		{
-			err: "^failed to load TLS config: failed to load client CA CertPool: failed to load CA /doesnt/exist:",
+			err: "^failed to load client CA CertPool: failed to load CA /doesnt/exist:",
 			settings: GRPCServerSettings{
 				NetAddr: confignet.NetAddr{
 					Endpoint:  "127.0.0.1:1234",

config/confighttp/confighttp_test.go

@@ -379,7 +379,7 @@ func TestHTTPServerSettingsError(t *testing.T) {
 			},
 		},
 		{
-			err: "^failed to load TLS config: failed to load client CA CertPool: failed to load CA /doesnt/exist:",
+			err: "failed to load client CA CertPool: failed to load CA /doesnt/exist:",
 			settings: HTTPServerSettings{
 				Endpoint: "localhost:0",
 				TLSSetting: &configtls.TLSServerSetting{

config/configtls/clientcasfilereloader.go

@@ -0,0 +1,152 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package configtls // import "go.opentelemetry.io/collector/config/configtls"
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"sync"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+type clientCAsFileReloader struct {
+	clientCAsFile   string
+	certPool        *x509.CertPool
+	lastReloadError error
+	lock            sync.RWMutex
+	loader          clientCAsFileLoader
+	watcher         *fsnotify.Watcher
+	shutdownCH      chan bool
+}
+
+type clientCAsFileLoader interface {
+	loadClientCAFile() (*x509.CertPool, error)
+}
+
+func newClientCAsReloader(clientCAsFile string, loader clientCAsFileLoader) (*clientCAsFileReloader, error) {
+	certPool, err := loader.loadClientCAFile()
+	if err != nil {
+		return nil, fmt.Errorf("failed to load client CA CertPool: %w", err)
+	}
+
+	reloader := &clientCAsFileReloader{
+		clientCAsFile: clientCAsFile,
+		certPool:      certPool,
+		loader:        loader,
+		shutdownCH:    nil,
+		watcher:       nil,
+	}
+
+	return reloader, nil
+}
+
+func (r *clientCAsFileReloader) getClientConfig(original *tls.Config) (*tls.Config, error) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	return &tls.Config{
+		RootCAs:              original.RootCAs,
+		GetCertificate:       original.GetCertificate,
+		GetClientCertificate: original.GetClientCertificate,
+		MinVersion:           original.MinVersion,
+		MaxVersion:           original.MaxVersion,
+		NextProtos:           original.NextProtos,
+		ClientCAs:            r.certPool,
+		ClientAuth:           tls.RequireAndVerifyClientCert,
+	}, nil
+}
+
+func (r *clientCAsFileReloader) reload() {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	certPool, err := r.loader.loadClientCAFile()
+	if err != nil {
+		r.lastReloadError = err
+	} else {
+		r.certPool = certPool
+		r.lastReloadError = nil
+	}
+}
+
+func (r *clientCAsFileReloader) getLastError() error {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	return r.lastReloadError
+}
+
+func (r *clientCAsFileReloader) startWatching() error {
+	if r.shutdownCH != nil {
+		return fmt.Errorf("client CA file watcher already started")
+	}
+
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return fmt.Errorf("failed to create watcher to reload client CA CertPool: %w", err)
+	}
+	r.watcher = watcher
+
+	err = watcher.Add(r.clientCAsFile)
+	if err != nil {
+		return fmt.Errorf("failed to add client CA file to watcher: %w", err)
+	}
+
+	r.shutdownCH = make(chan bool)
+	go r.handleWatcherEvents()
+
+	return nil
+}
+
+func (r *clientCAsFileReloader) handleWatcherEvents() {
+	defer r.watcher.Close()
+	for {
+		select {
+		case _, ok := <-r.shutdownCH:
+			_ = ok
+			return
+		case event, ok := <-r.watcher.Events:
+			if !ok {
+				continue
+			}
+			// NOTE: k8s configmaps uses symlinks, we need this workaround.
+			// original configmap file is removed.
+			// SEE: https://martensson.io/go-fsnotify-and-kubernetes-configmaps/
+			if event.Op == fsnotify.Remove || event.Op == fsnotify.Chmod {
+				// remove the watcher since the file is removed
+				if err := r.watcher.Remove(event.Name); err != nil {
+					r.lastReloadError = err
+				}
+				// add a new watcher pointing to the new symlink/file
+				if err := r.watcher.Add(r.clientCAsFile); err != nil {
+					r.lastReloadError = err
+				}
+				r.reload()
+			}
+			if event.Op == fsnotify.Write {
+				r.reload()
+			}
+		}
+	}
+}
+
+func (r *clientCAsFileReloader) shutdown() error {
+	if r.shutdownCH == nil {
+		return fmt.Errorf("client CAs file watcher is not running")
+	}
+	r.shutdownCH <- true
+	close(r.shutdownCH)
+	r.shutdownCH = nil
+	return nil
+}

config/configtls/clientcasfilereloader_test.go

@@ -0,0 +1,119 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package configtls
+
+import (
+	"crypto/x509"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCannotShutdownIfNotWatching(t *testing.T) {
+	reloader, _, _ := createReloader(t)
+	err := reloader.shutdown()
+	assert.Error(t, err)
+}
+
+func TestCannotStartIfAlreadyWatching(t *testing.T) {
+	reloader, _, _ := createReloader(t)
+
+	err := reloader.startWatching()
+	assert.NoError(t, err)
+
+	err = reloader.startWatching()
+	assert.Error(t, err)
+
+	err = reloader.shutdown()
+	assert.NoError(t, err)
+}
+
+func TestClosingWatcherDoesntBreakReloader(t *testing.T) {
+	reloader, loader, _ := createReloader(t)
+
+	err := reloader.startWatching()
+	assert.NoError(t, err)
+
+	assert.Equal(t, 1, loader.reloadNumber())
+
+	err = reloader.watcher.Close()
+	assert.NoError(t, err)
+
+	err = reloader.shutdown()
+	assert.NoError(t, err)
+}
+
+func TestErrorRecordedIfFileDeleted(t *testing.T) {
+	reloader, loader, filePath := createReloader(t)
+
+	err := reloader.startWatching()
+	assert.NoError(t, err)
+
+	assert.Equal(t, 1, loader.reloadNumber())
+
+	loader.returnErrorOnSubsequentCalls("test error on reload")
+
+	err = os.WriteFile(filePath, []byte("some_data"), 0600)
+	assert.NoError(t, err)
+
+	assert.Eventually(t, func() bool {
+		return loader.reloadNumber() > 1 && reloader.getLastError() != nil
+	}, 5*time.Second, 10*time.Millisecond)
+
+	lastErr := reloader.getLastError()
+	assert.Equal(t, "test error on reload", fmt.Sprint(lastErr))
+
+	err = reloader.shutdown()
+	assert.NoError(t, err)
+}
+
+func createReloader(t *testing.T) (*clientCAsFileReloader, *testLoader, string) {
+	tmpClientCAsFilePath := createTempFile(t)
+	loader := &testLoader{}
+	reloader, _ := newClientCAsReloader(tmpClientCAsFilePath, loader)
+	return reloader, loader, tmpClientCAsFilePath
+}
+
+func createTempFile(t *testing.T) string {
+	tmpCa, err := os.CreateTemp("", "clientCAs.crt")
+	assert.NoError(t, err)
+	tmpCaPath, err := filepath.Abs(tmpCa.Name())
+	assert.NoError(t, err)
+	assert.NoError(t, tmpCa.Close())
+	return tmpCaPath
+}
+
+type testLoader struct {
+	err     error
+	counter atomic.Uint32
+}
+
+func (r *testLoader) loadClientCAFile() (*x509.CertPool, error) {
+	r.counter.Add(1)
+	return nil, r.err
+}
+
+func (r *testLoader) returnErrorOnSubsequentCalls(msg string) {
+	r.err = fmt.Errorf(msg)
+}
+
+func (r *testLoader) reloadNumber() int {
+	return int(r.counter.Load())
+}

config/configtls/configtls.go

@@ -97,6 +97,10 @@ type TLSServerSetting struct {
 	// This sets the ClientCAs and ClientAuth to RequireAndVerifyClientCert in the TLSConfig. Please refer to
 	// https://godoc.org/crypto/tls#Config for more information. (optional)
 	ClientCAFile string `mapstructure:"client_ca_file"`
+
+	// Reload the ClientCAs file when it is modified
+	// (optional, default false)
+	ReloadClientCAFile bool `mapstructure:"client_ca_file_reload"`
 }
 
 // certReloader is a wrapper object for certificate reloading
@@ -236,16 +240,27 @@ func (c TLSServerSetting) LoadTLSConfig() (*tls.Config, error) {
 		return nil, fmt.Errorf("failed to load TLS config: %w", err)
 	}
 	if c.ClientCAFile != "" {
-		certPool, err := c.loadCert(c.ClientCAFile)
+		reloader, err := newClientCAsReloader(c.ClientCAFile, &c)
 		if err != nil {
-			return nil, fmt.Errorf("failed to load TLS config: failed to load client CA CertPool: %w", err)
+			return nil, err
+		}
+		if c.ReloadClientCAFile {
+			err = reloader.startWatching()
+			if err != nil {
+				return nil, err
+			}
+			tlsCfg.GetConfigForClient = func(t *tls.ClientHelloInfo) (*tls.Config, error) { return reloader.getClientConfig(tlsCfg) }
 		}
-		tlsCfg.ClientCAs = certPool
+		tlsCfg.ClientCAs = reloader.certPool
 		tlsCfg.ClientAuth = tls.RequireAndVerifyClientCert
 	}
 	return tlsCfg, nil
 }
 
+func (c TLSServerSetting) loadClientCAFile() (*x509.CertPool, error) {
+	return c.loadCert(c.ClientCAFile)
+}
+
 func convertVersion(v string, defaultVersion uint16) (uint16, error) {
 	// Use a default that is explicitly defined
 	if v == "" {