Add cert hot reload and cert file locking

Adds file-watching certificate hot reload, cross-replica ACME
certificate lock coordination via flock (Unix) and Kubernetes lease
objects.
This commit is contained in:
Viktor Liu
2026-02-09 22:55:12 +08:00
parent be5f30225a
commit fd442138e6
14 changed files with 1606 additions and 23 deletions
+279
View File
@@ -0,0 +1,279 @@
// Package certwatch watches TLS certificate files on disk and provides
// a hot-reloading GetCertificate callback for tls.Config.
package certwatch
import (
"context"
"crypto/tls"
"crypto/x509"
"fmt"
"path/filepath"
"sync"
"time"
"github.com/fsnotify/fsnotify"
log "github.com/sirupsen/logrus"
)
const (
defaultPollInterval = 30 * time.Second
debounceDelay = 500 * time.Millisecond
)
// Watcher monitors TLS certificate files on disk and caches the loaded
// certificate in memory. It detects changes via fsnotify (with a polling
// fallback for filesystems like NFS that lack inotify support) and
// reloads the certificate pair automatically.
type Watcher struct {
certPath string
keyPath string
mu sync.RWMutex
cert *tls.Certificate
leaf *x509.Certificate
pollInterval time.Duration
logger *log.Logger
}
// NewWatcher creates a Watcher that monitors the given cert and key files.
// It performs an initial load of the certificate and returns an error
// if the initial load fails.
func NewWatcher(certPath, keyPath string, logger *log.Logger) (*Watcher, error) {
if logger == nil {
logger = log.StandardLogger()
}
w := &Watcher{
certPath: certPath,
keyPath: keyPath,
pollInterval: defaultPollInterval,
logger: logger,
}
if err := w.reload(); err != nil {
return nil, fmt.Errorf("initial certificate load: %w", err)
}
return w, nil
}
// GetCertificate returns the current in-memory certificate.
// It is safe for concurrent use and compatible with tls.Config.GetCertificate.
func (w *Watcher) GetCertificate(_ *tls.ClientHelloInfo) (*tls.Certificate, error) {
w.mu.RLock()
defer w.mu.RUnlock()
return w.cert, nil
}
// Watch starts watching for certificate file changes. It blocks until
// ctx is cancelled. It uses fsnotify for immediate detection and falls
// back to polling if fsnotify is unavailable (e.g. on NFS).
// Even with fsnotify active, a periodic poll runs as a safety net.
func (w *Watcher) Watch(ctx context.Context) {
// Watch the parent directory rather than individual files. Some volume
// mounts use an atomic symlink swap (..data -> timestamped dir), so
// watching the parent directory catches the link replacement.
certDir := filepath.Dir(w.certPath)
keyDir := filepath.Dir(w.keyPath)
watcher, err := fsnotify.NewWatcher()
if err != nil {
w.logger.Warnf("fsnotify unavailable, using polling only: %v", err)
w.pollLoop(ctx)
return
}
defer func() {
if err := watcher.Close(); err != nil {
w.logger.Debugf("close fsnotify watcher: %v", err)
}
}()
if err := watcher.Add(certDir); err != nil {
w.logger.Warnf("fsnotify watch on %s failed, using polling only: %v", certDir, err)
w.pollLoop(ctx)
return
}
if keyDir != certDir {
if err := watcher.Add(keyDir); err != nil {
w.logger.Warnf("fsnotify watch on %s failed: %v", keyDir, err)
}
}
w.logger.Infof("watching certificate files in %s", certDir)
w.fsnotifyLoop(ctx, watcher)
}
func (w *Watcher) fsnotifyLoop(ctx context.Context, watcher *fsnotify.Watcher) {
certBase := filepath.Base(w.certPath)
keyBase := filepath.Base(w.keyPath)
var debounce *time.Timer
defer func() {
if debounce != nil {
debounce.Stop()
}
}()
// Periodic poll as a safety net for missed fsnotify events.
pollTicker := time.NewTicker(w.pollInterval)
defer pollTicker.Stop()
for {
select {
case <-ctx.Done():
return
case event, ok := <-watcher.Events:
if !ok {
return
}
base := filepath.Base(event.Name)
if !isRelevantFile(base, certBase, keyBase) {
w.logger.Debugf("fsnotify: ignoring event %s on %s", event.Op, event.Name)
continue
}
if !event.Has(fsnotify.Create) && !event.Has(fsnotify.Write) && !event.Has(fsnotify.Rename) {
w.logger.Debugf("fsnotify: ignoring op %s on %s", event.Op, base)
continue
}
w.logger.Debugf("fsnotify: detected %s on %s, scheduling reload", event.Op, base)
// Debounce: cert-manager may write cert and key as separate
// operations. Wait briefly to load both at once.
if debounce != nil {
debounce.Stop()
}
debounce = time.AfterFunc(debounceDelay, func() {
if ctx.Err() != nil {
return
}
w.tryReload()
})
case err, ok := <-watcher.Errors:
if !ok {
return
}
w.logger.Warnf("fsnotify error: %v", err)
case <-pollTicker.C:
w.tryReload()
}
}
}
func (w *Watcher) pollLoop(ctx context.Context) {
ticker := time.NewTicker(w.pollInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
w.tryReload()
}
}
}
// reload loads the certificate from disk and updates the in-memory cache.
func (w *Watcher) reload() error {
cert, err := tls.LoadX509KeyPair(w.certPath, w.keyPath)
if err != nil {
return err
}
// Parse the leaf for comparison on subsequent reloads.
if cert.Leaf == nil && len(cert.Certificate) > 0 {
leaf, err := x509.ParseCertificate(cert.Certificate[0])
if err != nil {
return fmt.Errorf("parse leaf certificate: %w", err)
}
cert.Leaf = leaf
}
w.mu.Lock()
w.cert = &cert
w.leaf = cert.Leaf
w.mu.Unlock()
w.logCertDetails("loaded certificate", cert.Leaf)
return nil
}
// tryReload attempts to reload the certificate. It skips the update
// if the certificate on disk is identical to the one in memory (same
// serial number and issuer) to avoid redundant log noise.
func (w *Watcher) tryReload() {
cert, err := tls.LoadX509KeyPair(w.certPath, w.keyPath)
if err != nil {
w.logger.Warnf("reload certificate: %v", err)
return
}
if cert.Leaf == nil && len(cert.Certificate) > 0 {
leaf, err := x509.ParseCertificate(cert.Certificate[0])
if err != nil {
w.logger.Warnf("parse reloaded leaf certificate: %v", err)
return
}
cert.Leaf = leaf
}
w.mu.Lock()
if w.leaf != nil && cert.Leaf != nil &&
w.leaf.SerialNumber.Cmp(cert.Leaf.SerialNumber) == 0 &&
w.leaf.Issuer.CommonName == cert.Leaf.Issuer.CommonName {
w.mu.Unlock()
return
}
prev := w.leaf
w.cert = &cert
w.leaf = cert.Leaf
w.mu.Unlock()
w.logCertChange(prev, cert.Leaf)
}
func (w *Watcher) logCertDetails(msg string, leaf *x509.Certificate) {
if leaf == nil {
w.logger.Info(msg)
return
}
w.logger.Infof("%s: subject=%q serial=%s SANs=%v notAfter=%s",
msg,
leaf.Subject.CommonName,
leaf.SerialNumber.Text(16),
leaf.DNSNames,
leaf.NotAfter.UTC().Format(time.RFC3339),
)
}
func (w *Watcher) logCertChange(prev, next *x509.Certificate) {
if prev == nil || next == nil {
w.logCertDetails("certificate reloaded from disk", next)
return
}
w.logger.Infof("certificate reloaded from disk: subject=%q -> %q serial=%s -> %s notAfter=%s -> %s",
prev.Subject.CommonName, next.Subject.CommonName,
prev.SerialNumber.Text(16), next.SerialNumber.Text(16),
prev.NotAfter.UTC().Format(time.RFC3339), next.NotAfter.UTC().Format(time.RFC3339),
)
}
// isRelevantFile returns true if the changed file name is one we care about.
// This includes the cert/key files themselves and the ..data symlink used
// by atomic volume mounts.
func isRelevantFile(changed, certBase, keyBase string) bool {
return changed == certBase || changed == keyBase || changed == "..data"
}
+292
View File
@@ -0,0 +1,292 @@
package certwatch
import (
"context"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"encoding/pem"
"math/big"
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func generateSelfSignedCert(t *testing.T, serial int64) (certPEM, keyPEM []byte) {
t.Helper()
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
require.NoError(t, err)
template := &x509.Certificate{
SerialNumber: big.NewInt(serial),
Subject: pkix.Name{CommonName: "test"},
NotBefore: time.Now().Add(-time.Hour),
NotAfter: time.Now().Add(24 * time.Hour),
}
certDER, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
require.NoError(t, err)
certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER})
keyDER, err := x509.MarshalECPrivateKey(key)
require.NoError(t, err)
keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})
return certPEM, keyPEM
}
func writeCert(t *testing.T, dir string, certPEM, keyPEM []byte) {
t.Helper()
require.NoError(t, os.WriteFile(filepath.Join(dir, "tls.crt"), certPEM, 0o600))
require.NoError(t, os.WriteFile(filepath.Join(dir, "tls.key"), keyPEM, 0o600))
}
func TestNewWatcher(t *testing.T) {
dir := t.TempDir()
certPEM, keyPEM := generateSelfSignedCert(t, 1)
writeCert(t, dir, certPEM, keyPEM)
w, err := NewWatcher(
filepath.Join(dir, "tls.crt"),
filepath.Join(dir, "tls.key"),
nil,
)
require.NoError(t, err)
cert, err := w.GetCertificate(nil)
require.NoError(t, err)
require.NotNil(t, cert)
assert.Equal(t, int64(1), cert.Leaf.SerialNumber.Int64())
}
func TestNewWatcherMissingFiles(t *testing.T) {
dir := t.TempDir()
_, err := NewWatcher(
filepath.Join(dir, "tls.crt"),
filepath.Join(dir, "tls.key"),
nil,
)
assert.Error(t, err)
}
func TestReload(t *testing.T) {
dir := t.TempDir()
certPEM1, keyPEM1 := generateSelfSignedCert(t, 100)
writeCert(t, dir, certPEM1, keyPEM1)
w, err := NewWatcher(
filepath.Join(dir, "tls.crt"),
filepath.Join(dir, "tls.key"),
nil,
)
require.NoError(t, err)
cert1, err := w.GetCertificate(nil)
require.NoError(t, err)
assert.Equal(t, int64(100), cert1.Leaf.SerialNumber.Int64())
// Write a new cert with a different serial.
certPEM2, keyPEM2 := generateSelfSignedCert(t, 200)
writeCert(t, dir, certPEM2, keyPEM2)
// Manually trigger reload.
w.tryReload()
cert2, err := w.GetCertificate(nil)
require.NoError(t, err)
assert.Equal(t, int64(200), cert2.Leaf.SerialNumber.Int64())
}
func TestTryReloadSkipsUnchanged(t *testing.T) {
dir := t.TempDir()
certPEM, keyPEM := generateSelfSignedCert(t, 42)
writeCert(t, dir, certPEM, keyPEM)
w, err := NewWatcher(
filepath.Join(dir, "tls.crt"),
filepath.Join(dir, "tls.key"),
nil,
)
require.NoError(t, err)
cert1, err := w.GetCertificate(nil)
require.NoError(t, err)
// Reload with same cert - pointer should remain the same.
w.tryReload()
cert2, err := w.GetCertificate(nil)
require.NoError(t, err)
assert.Same(t, cert1, cert2, "cert pointer should not change when content is the same")
}
func TestWatchDetectsChanges(t *testing.T) {
dir := t.TempDir()
certPEM1, keyPEM1 := generateSelfSignedCert(t, 1)
writeCert(t, dir, certPEM1, keyPEM1)
w, err := NewWatcher(
filepath.Join(dir, "tls.crt"),
filepath.Join(dir, "tls.key"),
nil,
)
require.NoError(t, err)
// Use a short poll interval for the test.
w.pollInterval = 100 * time.Millisecond
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go w.Watch(ctx)
// Write new cert.
certPEM2, keyPEM2 := generateSelfSignedCert(t, 999)
writeCert(t, dir, certPEM2, keyPEM2)
// Wait for the watcher to pick it up.
require.Eventually(t, func() bool {
cert, err := w.GetCertificate(nil)
if err != nil {
return false
}
return cert.Leaf.SerialNumber.Int64() == 999
}, 5*time.Second, 50*time.Millisecond, "watcher should detect cert change")
}
func TestIsRelevantFile(t *testing.T) {
assert.True(t, isRelevantFile("tls.crt", "tls.crt", "tls.key"))
assert.True(t, isRelevantFile("tls.key", "tls.crt", "tls.key"))
assert.True(t, isRelevantFile("..data", "tls.crt", "tls.key"))
assert.False(t, isRelevantFile("other.txt", "tls.crt", "tls.key"))
}
// TestWatchSymlinkRotation simulates Kubernetes secret volume updates where
// the data directory is atomically swapped via a ..data symlink.
func TestWatchSymlinkRotation(t *testing.T) {
base := t.TempDir()
// Create initial target directory with certs.
dir1 := filepath.Join(base, "dir1")
require.NoError(t, os.Mkdir(dir1, 0o755))
certPEM1, keyPEM1 := generateSelfSignedCert(t, 1)
require.NoError(t, os.WriteFile(filepath.Join(dir1, "tls.crt"), certPEM1, 0o600))
require.NoError(t, os.WriteFile(filepath.Join(dir1, "tls.key"), keyPEM1, 0o600))
// Create ..data symlink pointing to dir1.
dataLink := filepath.Join(base, "..data")
require.NoError(t, os.Symlink(dir1, dataLink))
// Create tls.crt and tls.key as symlinks to ..data/{file}.
certLink := filepath.Join(base, "tls.crt")
keyLink := filepath.Join(base, "tls.key")
require.NoError(t, os.Symlink(filepath.Join(dataLink, "tls.crt"), certLink))
require.NoError(t, os.Symlink(filepath.Join(dataLink, "tls.key"), keyLink))
w, err := NewWatcher(certLink, keyLink, nil)
require.NoError(t, err)
cert, err := w.GetCertificate(nil)
require.NoError(t, err)
assert.Equal(t, int64(1), cert.Leaf.SerialNumber.Int64())
w.pollInterval = 100 * time.Millisecond
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go w.Watch(ctx)
// Simulate k8s atomic rotation: create dir2, swap ..data symlink.
dir2 := filepath.Join(base, "dir2")
require.NoError(t, os.Mkdir(dir2, 0o755))
certPEM2, keyPEM2 := generateSelfSignedCert(t, 777)
require.NoError(t, os.WriteFile(filepath.Join(dir2, "tls.crt"), certPEM2, 0o600))
require.NoError(t, os.WriteFile(filepath.Join(dir2, "tls.key"), keyPEM2, 0o600))
// Atomic swap: create temp link, then rename over ..data.
tmpLink := filepath.Join(base, "..data_tmp")
require.NoError(t, os.Symlink(dir2, tmpLink))
require.NoError(t, os.Rename(tmpLink, dataLink))
require.Eventually(t, func() bool {
cert, err := w.GetCertificate(nil)
if err != nil {
return false
}
return cert.Leaf.SerialNumber.Int64() == 777
}, 5*time.Second, 50*time.Millisecond, "watcher should detect symlink rotation")
}
// TestPollLoopDetectsChanges verifies the poll-only fallback path works.
func TestPollLoopDetectsChanges(t *testing.T) {
dir := t.TempDir()
certPEM1, keyPEM1 := generateSelfSignedCert(t, 1)
writeCert(t, dir, certPEM1, keyPEM1)
w, err := NewWatcher(
filepath.Join(dir, "tls.crt"),
filepath.Join(dir, "tls.key"),
nil,
)
require.NoError(t, err)
w.pollInterval = 100 * time.Millisecond
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Directly use pollLoop to test the fallback path.
go w.pollLoop(ctx)
certPEM2, keyPEM2 := generateSelfSignedCert(t, 555)
writeCert(t, dir, certPEM2, keyPEM2)
require.Eventually(t, func() bool {
cert, err := w.GetCertificate(nil)
if err != nil {
return false
}
return cert.Leaf.SerialNumber.Int64() == 555
}, 5*time.Second, 50*time.Millisecond, "poll loop should detect cert change")
}
func TestGetCertificateConcurrency(t *testing.T) {
dir := t.TempDir()
certPEM, keyPEM := generateSelfSignedCert(t, 1)
writeCert(t, dir, certPEM, keyPEM)
w, err := NewWatcher(
filepath.Join(dir, "tls.crt"),
filepath.Join(dir, "tls.key"),
nil,
)
require.NoError(t, err)
// Hammer GetCertificate concurrently while reloading.
done := make(chan struct{})
go func() {
for i := 0; i < 100; i++ {
w.tryReload()
}
close(done)
}()
for i := 0; i < 1000; i++ {
cert, err := w.GetCertificate(&tls.ClientHelloInfo{})
assert.NoError(t, err)
assert.NotNil(t, cert)
}
<-done
}