INFOPLAT 2962 rotating beholder headers

go.mod

@@ -37,7 +37,7 @@ require (
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.67
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.4
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.6
 	github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250722225531-876fd6b94976
 	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20250911124514-5874cc6d62b2
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b

go.sum

@@ -326,8 +326,8 @@ github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMB
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/smartcontractkit/chain-selectors v1.0.67 h1:gxTqP/JC40KDe3DE1SIsIKSTKTZEPyEU1YufO1admnw=
 github.com/smartcontractkit/chain-selectors v1.0.67/go.mod h1:xsKM0aN3YGcQKTPRPDDtPx2l4mlTN1Djmg0VVXV40b8=
-github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.4 h1:hvqATtrZ0iMRTI80cpBot/3JFbjz2j+2tvpfooVhRHw=
-github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.4/go.mod h1:eKGyfTKzr0/PeR7qKN4l2FcW9p+HzyKUwAfGhm/5YZc=
+github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.6 h1:INTd6uKc/QO11B0Vx7Ze17xgW3bqYbWuQcBQa9ixicQ=
+github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.6/go.mod h1:eKGyfTKzr0/PeR7qKN4l2FcW9p+HzyKUwAfGhm/5YZc=
 github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250722225531-876fd6b94976 h1:mF3FiDUoV0QbJcks9R2y7ydqntNL1Z0VCPBJgx/Ms+0=
 github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20250722225531-876fd6b94976/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA=
 github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20250911124514-5874cc6d62b2 h1:1/KdO5AbUr3CmpLjMPuJXPo2wHMbfB8mldKLsg7D4M8=

pkg/beholder/auth.go

@@ -1,30 +1,157 @@
 package beholder
 
 import (
+	"context"
 	"crypto"
 	"crypto/ed25519"
 	"crypto/rand"
+	"encoding/binary"
 	"fmt"
+	"maps"
+	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 )
 
 // authHeaderKey is the name of the header that the node authenticator will use to send the auth token
 var authHeaderKey = "X-Beholder-Node-Auth-Token"
 
 // authHeaderVersion is the version of the auth header format
 var authHeaderVersion = "1"
+var authHeaderV2 = "2"
 
-type staticAuthHeaderProvider struct {
-	headers map[string]string
+type HeaderProvider interface {
+	Headers(ctx context.Context) (map[string]string, error)
 }
 
-func (p *staticAuthHeaderProvider) GetHeaders() map[string]string {
-	return p.headers
+type PerRPCCredentialsProvider interface {
+	Credentials() credentials.PerRPCCredentials
 }
 
+type Auth interface {
+	PerRPCCredentialsProvider
+	HeaderProvider
+}
+
+type Signer interface {
+	Sign(ctx context.Context, keyID []byte, data []byte) ([]byte, error)
+}
+
+type staticAuth struct {
+	headers                  map[string]string
+	requireTransportSecurity bool
+}
+
+func (p *staticAuth) Headers(_ context.Context) (map[string]string, error) {
+	return p.headers, nil
+}
+
+func (p *staticAuth) Credentials() credentials.PerRPCCredentials {
+	return p
+}
+
+func (p *staticAuth) GetRequestMetadata(ctx context.Context, _ ...string) (map[string]string, error) {
+	return p.Headers(ctx)
+}
+
+func (p *staticAuth) RequireTransportSecurity() bool {
+	return p.requireTransportSecurity
+}
+
+func NewStaticAuth(headers map[string]string, requireTransportSecurity bool) Auth {
+	return &staticAuth{headers, requireTransportSecurity}
+}
+
+// Deprecated: use NewStaticAuth instead
 func NewStaticAuthHeaderProvider(headers map[string]string) chipingress.HeaderProvider {
-	return &staticAuthHeaderProvider{headers: headers}
+	return &staticAuth{headers: headers}
+}
+
+type rotatingAuth struct {
+	csaPubKey                ed25519.PublicKey
+	signer                   Signer
+	signerTimeout            time.Duration
+	headers                  atomic.Value // stores map[string]string
+	ttl                      time.Duration
+	lastUpdatedNanos         atomic.Int64
+	requireTransportSecurity bool
+	mu                       sync.Mutex
+}
+
+func NewRotatingAuth(csaPubKey ed25519.PublicKey, signer Signer, ttl time.Duration, requireTransportSecurity bool) Auth {
+	r := &rotatingAuth{
+		csaPubKey:                csaPubKey,
+		signer:                   signer,
+		signerTimeout:            time.Second * 5,
+		ttl:                      ttl,
+		lastUpdatedNanos:         atomic.Int64{},
+		requireTransportSecurity: requireTransportSecurity,
+	}
+	r.headers.Store(make(map[string]string))
+	return r
+}
+
+func (r *rotatingAuth) Headers(ctx context.Context) (map[string]string, error) {
+
+	// Return a copy of the headers to avoid concurrent read/write to the map by callers
+	returnHeader := make(map[string]string)
+	lastUpdated := time.Unix(0, r.lastUpdatedNanos.Load())
+
+	if time.Since(lastUpdated) > r.ttl {
+
+		r.mu.Lock()
+		defer r.mu.Unlock()
+
+		// Multiple concurrent calls (after the first) will block waiting for the lock.
+		// First will get the lock and update headers + lastUpdated, double check since potentially another goroutine has already
+		// updated the headers and lastUpdated while waiting for the lock.
+		lastUpdated = time.Unix(0, r.lastUpdatedNanos.Load())
+		if time.Since(lastUpdated) < r.ttl {
+			maps.Copy(returnHeader, r.headers.Load().(map[string]string))
+			return returnHeader, nil
+		}
+
+		// Append the bytes of the public key with bytes of the timestamp to create the message to sign
+		ts := time.Now()
+		tsBytes := make([]byte, 8)
+		binary.BigEndian.PutUint64(tsBytes, uint64(ts.UnixNano()))
+		msgBytes := append(r.csaPubKey, tsBytes...)
+
+		ctxWithTimeout, cancel := context.WithTimeout(ctx, r.signerTimeout)
+		defer cancel()
+
+		// Sign(public key bytes + timestamp bytes)
+		signature, err := r.signer.Sign(ctxWithTimeout, r.csaPubKey, msgBytes)
+		if err != nil {
+			return nil, fmt.Errorf("beholder: failed to sign auth header: %w", err)
+		}
+
+		newHeaders := make(map[string]string)
+		newHeaders[authHeaderKey] = fmt.Sprintf("%s:%x:%d:%x", authHeaderV2, r.csaPubKey, ts.UnixNano(), signature)
+
+		r.headers.Store(newHeaders)
+		r.lastUpdatedNanos.Store(ts.UnixNano())
+	}
+
+	maps.Copy(returnHeader, r.headers.Load().(map[string]string))
+
+	return returnHeader, nil
+}
+
+func (a *rotatingAuth) Credentials() credentials.PerRPCCredentials {
+	return a
+}
+
+func (a *rotatingAuth) GetRequestMetadata(ctx context.Context, _ ...string) (map[string]string, error) {
+	return a.Headers(ctx)
+}
+
+func (a *rotatingAuth) RequireTransportSecurity() bool {
+	return a.requireTransportSecurity
 }
 
 // BuildAuthHeaders creates the auth header value to be included on requests.
@@ -58,3 +185,7 @@ func NewAuthHeaders(ed25519Signer crypto.Signer) (map[string]string, error) {
 
 	return map[string]string{authHeaderKey: headerValue}, nil
 }
+
+func authDialOpt(auth PerRPCCredentialsProvider) grpc.DialOption {
+	return grpc.WithPerRPCCredentials(auth.Credentials())
+}