INFOPLAT 2963 beholder rotating auth headers loop impl

pkg/loop/config.go

@@ -57,6 +57,7 @@ const (
 	envTelemetryTraceSampleRatio          = "CL_TELEMETRY_TRACE_SAMPLE_RATIO"
 	envTelemetryAuthHeader                = "CL_TELEMETRY_AUTH_HEADER"
 	envTelemetryAuthPubKeyHex             = "CL_TELEMETRY_AUTH_PUB_KEY_HEX"
+	envTelemetryAuthHeadersTTL            = "CL_TELEMETRY_AUTH_HEADERS_TTL"
 	envTelemetryEmitterBatchProcessor     = "CL_TELEMETRY_EMITTER_BATCH_PROCESSOR"
 	envTelemetryEmitterExportTimeout      = "CL_TELEMETRY_EMITTER_EXPORT_TIMEOUT"
 	envTelemetryEmitterExportInterval     = "CL_TELEMETRY_EMITTER_EXPORT_INTERVAL"
@@ -114,6 +115,7 @@ type EnvConfig struct {
 	TelemetryTraceSampleRatio          float64
 	TelemetryAuthHeaders               map[string]string
 	TelemetryAuthPubKeyHex             string
+	TelemetryAuthHeadersTTL            time.Duration
 	TelemetryEmitterBatchProcessor     bool
 	TelemetryEmitterExportTimeout      time.Duration
 	TelemetryEmitterExportInterval     time.Duration
@@ -184,6 +186,7 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 		add(envTelemetryAuthHeader+k, v)
 	}
 	add(envTelemetryAuthPubKeyHex, e.TelemetryAuthPubKeyHex)
+	add(envTelemetryAuthHeadersTTL, e.TelemetryAuthHeadersTTL.String())
 	add(envTelemetryEmitterBatchProcessor, strconv.FormatBool(e.TelemetryEmitterBatchProcessor))
 	add(envTelemetryEmitterExportTimeout, e.TelemetryEmitterExportTimeout.String())
 	add(envTelemetryEmitterExportInterval, e.TelemetryEmitterExportInterval.String())
@@ -331,6 +334,10 @@ func (e *EnvConfig) parse() error {
 		e.TelemetryTraceSampleRatio = getFloat64OrZero(envTelemetryTraceSampleRatio)
 		e.TelemetryAuthHeaders = getMap(envTelemetryAuthHeader)
 		e.TelemetryAuthPubKeyHex = os.Getenv(envTelemetryAuthPubKeyHex)
+		e.TelemetryAuthHeadersTTL, err = getDuration(envTelemetryAuthHeadersTTL)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envTelemetryAuthHeadersTTL, err)
+		}
 		e.TelemetryEmitterBatchProcessor, err = getBool(envTelemetryEmitterBatchProcessor)
 		if err != nil {
 			return fmt.Errorf("failed to parse %s: %w", envTelemetryEmitterBatchProcessor, err)

pkg/loop/lazy_keystore_signer.go

@@ -0,0 +1,49 @@
+package loop
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/types/keystore"
+)
+
+// lazyKeystoreSigner is a thread-safe wrapper that allows the keystore
+// to be set after the signer is created. This enables beholder to start
+// with rotating auth configured, but the actual keystore can be injected later.
+type lazyKeystoreSigner struct {
+	mu       sync.RWMutex
+	keystore keystore.Keystore
+}
+
+func newLazyKeystoreSigner() *lazyKeystoreSigner {
+	return &lazyKeystoreSigner{}
+}
+
+// Sign implements the beholder.Signer interface
+func (l *lazyKeystoreSigner) Sign(ctx context.Context, keyID []byte, data []byte) ([]byte, error) {
+	l.mu.RLock()
+	ks := l.keystore
+	l.mu.RUnlock()
+
+	if ks == nil {
+		return nil, fmt.Errorf("keystore not yet available for signing")
+	}
+
+	return ks.Sign(ctx, keyID, data)
+}
+
+// SetKeystore updates the underlying keystore. This is thread-safe and can be
+// called at any time, even after beholder has been initialized.
+func (l *lazyKeystoreSigner) SetKeystore(ks keystore.Keystore) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.keystore = ks
+}
+
+// HasKeystore returns true if a keystore has been set
+func (l *lazyKeystoreSigner) HasKeystore() bool {
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+	return l.keystore != nil
+}

pkg/loop/server.go

@@ -21,6 +21,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
 	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil/pg"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/keystore"
 )
 
 // NewStartedServer returns a started Server.
@@ -78,6 +79,10 @@ type Server struct {
 	checker         *services.HealthChecker
 	LimitsFactory   limits.Factory
 	otelViews       []sdkmetric.View
+	// Keystore is an optional keystore that can be used for beholder auth signing
+	Keystore keystore.Keystore
+	// lazySigner allows keystore to be injected after beholder initialization
+	lazySigner *lazyKeystoreSigner
 }
 
 func newServer(loggerName string, otelViews []sdkmetric.View) (*Server, error) {
@@ -133,8 +138,6 @@ func (s *Server) start() error {
 			OtelExporterGRPCEndpoint:       s.EnvConfig.TelemetryEndpoint,
 			ResourceAttributes:             append(attributes, s.EnvConfig.TelemetryAttributes.AsStringAttributes()...),
 			TraceSampleRatio:               s.EnvConfig.TelemetryTraceSampleRatio,
-			AuthHeaders:                    s.EnvConfig.TelemetryAuthHeaders,
-			AuthPublicKeyHex:               s.EnvConfig.TelemetryAuthPubKeyHex,
 			EmitterBatchProcessor:          s.EnvConfig.TelemetryEmitterBatchProcessor,
 			EmitterExportTimeout:           s.EnvConfig.TelemetryEmitterExportTimeout,
 			EmitterExportInterval:          s.EnvConfig.TelemetryEmitterExportInterval,
@@ -147,6 +150,36 @@ func (s *Server) start() error {
 			ChipIngressInsecureConnection:  s.EnvConfig.ChipIngressInsecureConnection,
 		}
 
+		// Configure beholder auth with lazy keystore injection support
+		if s.EnvConfig.TelemetryAuthPubKeyHex != "" && s.EnvConfig.TelemetryAuthHeadersTTL > 0 {
+			// Use lazy signer pattern - allows keystore to be injected after startup
+			s.lazySigner = newLazyKeystoreSigner()
+
+			// If keystore is already available, set it immediately
+			if s.Keystore != nil {
+				s.lazySigner.SetKeystore(s.Keystore)
+				s.Logger.Infow("Using keystore for beholder rotating auth", "ttl", s.EnvConfig.TelemetryAuthHeadersTTL)
+			} else {
+				s.Logger.Infow("Beholder rotating auth configured, waiting for keystore injection", "ttl", s.EnvConfig.TelemetryAuthHeadersTTL)
+			}
+
+			beholderCfg.AuthKeySigner = s.lazySigner
+			beholderCfg.AuthPublicKeyHex = s.EnvConfig.TelemetryAuthPubKeyHex
+			beholderCfg.AuthHeadersTTL = s.EnvConfig.TelemetryAuthHeadersTTL
+		} else {
+			// Fall back to static auth headers if rotating auth not configured
+			if len(s.EnvConfig.TelemetryAuthHeaders) > 0 {
+				s.Logger.Info("Using static auth headers for beholder")
+				beholderCfg.AuthHeaders = s.EnvConfig.TelemetryAuthHeaders
+				beholderCfg.AuthPublicKeyHex = s.EnvConfig.TelemetryAuthPubKeyHex
+			} else if s.EnvConfig.TelemetryAuthPubKeyHex != "" {
+				// Static auth with pub key only
+				s.Logger.Info("Using static auth with public key for beholder")
+				beholderCfg.AuthHeaders = s.EnvConfig.TelemetryAuthHeaders
+				beholderCfg.AuthPublicKeyHex = s.EnvConfig.TelemetryAuthPubKeyHex
+			}
+		}
+
 		// note: due to the OTEL specification, all histogram buckets
 		// must be defined when the beholder client is created
 		beholderCfg.MetricViews = append(beholderCfg.MetricViews, s.otelViews...)
@@ -237,6 +270,23 @@ func (s *Server) MustRegister(c services.HealthReporter) {
 
 func (s *Server) Register(c services.HealthReporter) error { return s.checker.Register(c) }
 
+// UpdateKeystore injects or updates the keystore used for beholder auth signing.
+// This can be called at any time, even after the server has started and beholder
+// has been initialized. The next time beholder needs to sign (after TTL expires),
+// it will use the new keystore.
+func (s *Server) UpdateKeystore(ks keystore.Keystore) {
+	s.Keystore = ks
+	if s.lazySigner != nil {
+		wasSet := s.lazySigner.HasKeystore()
+		s.lazySigner.SetKeystore(ks)
+		if wasSet {
+			s.Logger.Info("Keystore updated for beholder rotating auth")
+		} else {
+			s.Logger.Info("Keystore injected for beholder rotating auth")
+		}
+	}
+}
+
 // Stop closes resources and flushes logs.
 func (s *Server) Stop() {
 	if s.dbStatsReporter != nil {