auth: implement re-authentication in case of rotated certificates

This commit introduces mutual auth re-authentication. Whenever an
authhandler is emiting CertificateRotatedEvents, an authentication will
be triggered.

Signed-off-by: Marco Hofstetter <marco.hofstetter@isovalent.com>

pkg/auth/cell.go

@@ -113,6 +113,8 @@ func newManager(params authManagerParams) error {
 		return fmt.Errorf("failed to register signal authentication job: %w", err)
 	}
 
+	registerReAuthenticationJob(jobGroup, mgr, params.AuthHandlers)
+
 	mapGC := newAuthMapGC(mapCache, params.IPCache)
 
 	registerGCJobs(jobGroup, mapGC, params)
@@ -122,6 +124,14 @@ func newManager(params authManagerParams) error {
 	return nil
 }
 
+func registerReAuthenticationJob(jobGroup job.Group, mgr *authManager, authHandlers []authHandler) {
+	for _, ah := range authHandlers {
+		if ah != nil && ah.subscribeToRotatedIdentities() != nil {
+			jobGroup.Add(job.Observer("auth re-authentication", mgr.handleCertificateRotationEvent, stream.FromChannel(ah.subscribeToRotatedIdentities())))
+		}
+	}
+}
+
 func registerSignalAuthenticationJob(jobGroup job.Group, mgr *authManager, sm signal.SignalManager, config config) error {
 	var signalChannel = make(chan signalAuthKey, config.MeshAuthQueueSize)
 

pkg/auth/manager.go

@@ -29,8 +29,9 @@ type authManager struct {
 	authHandlers map[policy.AuthType]authHandler
 	authmap      authMap
 
-	mutex   lock.Mutex
-	pending map[authKey]struct{}
+	mutex                    lock.Mutex
+	pending                  map[authKey]struct{}
+	handleAuthenticationFunc func(a *authManager, k authKey, reAuth bool)
 }
 
 // ipCache is the set of interactions the auth manager performs with the IPCache
@@ -56,11 +57,7 @@ type authResponse struct {
 	expirationTime time.Time
 }
 
-func newAuthManager(
-	authHandlers []authHandler,
-	authmap authMap,
-	ipCache ipCache,
-) (*authManager, error) {
+func newAuthManager(authHandlers []authHandler, authmap authMap, ipCache ipCache) (*authManager, error) {
 	ahs := map[policy.AuthType]authHandler{}
 	for _, ah := range authHandlers {
 		if ah == nil {
@@ -73,10 +70,11 @@ func newAuthManager(
 	}
 
 	return &authManager{
-		authHandlers: ahs,
-		authmap:      authmap,
-		ipCache:      ipCache,
-		pending:      make(map[authKey]struct{}),
+		authHandlers:             ahs,
+		authmap:                  authmap,
+		ipCache:                  ipCache,
+		pending:                  make(map[authKey]struct{}),
+		handleAuthenticationFunc: handleAuthentication,
 	}, nil
 }
 
@@ -89,25 +87,50 @@ func (a *authManager) handleAuthRequest(_ context.Context, key signalAuthKey) er
 		authType:       policy.AuthType(key.AuthType),
 	}
 
+	log.Debugf("auth: Handle authentication request for key %s", k)
+
+	a.handleAuthenticationFunc(a, k, false)
+
+	return nil
+}
+
+func (a *authManager) handleCertificateRotationEvent(_ context.Context, event certs.CertificateRotationEvent) error {
+	log.Debugf("auth: Handle certificate rotation event for identity %s", event.Identity)
+
+	all, err := a.authmap.All()
+	if err != nil {
+		return fmt.Errorf("failed to get all auth map entries: %w", err)
+	}
+
+	for k := range all {
+		if k.localIdentity == event.Identity || k.remoteIdentity == event.Identity {
+			a.handleAuthenticationFunc(a, k, true)
+		}
+	}
+
+	return nil
+}
+
+func handleAuthentication(a *authManager, k authKey, reAuth bool) {
 	if a.markPendingAuth(k) {
 		go func(key authKey) {
 			defer a.clearPendingAuth(key)
 
-			// Check if the auth is actually required, as we might have
-			// updated the authmap since the datapath issued the auth
-			// required signal.
-			if i, err := a.authmap.Get(key); err == nil && i.expiration.After(time.Now()) {
-				log.Debugf("auth: Already authenticated, skipped authentication for key %v", key)
-				return
+			if !reAuth {
+				// Check if the auth is actually required, as we might have
+				// updated the authmap since the datapath issued the auth
+				// required signal.
+				if i, err := a.authmap.Get(key); err == nil && i.expiration.After(time.Now()) {
+					log.Debugf("auth: Already authenticated, skipped authentication for key %v", key)
+					return
+				}
 			}
 
 			if err := a.authenticate(key); err != nil {
 				log.WithError(err).Warningf("auth: Failed to authenticate request for key %v", key)
 			}
 		}(k)
 	}
-
-	return nil
 }
 
 // markPendingAuth checks if there is a pending authentication for the given key.

pkg/auth/manager_test.go

@@ -4,14 +4,17 @@
 package auth
 
 import (
+	"context"
 	"errors"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/exp/maps"
 
 	"github.com/cilium/cilium/pkg/auth/certs"
+	"github.com/cilium/cilium/pkg/identity"
 	"github.com/cilium/cilium/pkg/policy"
 )
 
@@ -105,6 +108,71 @@ func Test_authManager_authenticate(t *testing.T) {
 	}
 }
 
+func Test_authManager_handleAuthRequest(t *testing.T) {
+	authHandlers := []authHandler{
+		&alwaysPassAuthHandler{},
+	}
+
+	am, err := newAuthManager(authHandlers, nil, nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, am)
+
+	handleAuthCalled := false
+	am.handleAuthenticationFunc = func(_ *authManager, k authKey, reAuth bool) {
+		handleAuthCalled = true
+		assert.False(t, reAuth)
+		assert.Equal(t, authKey{localIdentity: 1, remoteIdentity: 2, remoteNodeID: 0, authType: 100}, k)
+	}
+
+	err = am.handleAuthRequest(context.Background(), signalAuthKey{LocalIdentity: 1, RemoteIdentity: 2, RemoteNodeID: 0, AuthType: 100, Pad: 0})
+	assert.NoError(t, err)
+	assert.True(t, handleAuthCalled)
+}
+
+func Test_authManager_handleCertificateRotationEvent_Error(t *testing.T) {
+	authHandlers := []authHandler{
+		&alwaysPassAuthHandler{},
+	}
+	aMap := &fakeAuthMap{
+		failGet: true,
+	}
+
+	am, err := newAuthManager(authHandlers, aMap, nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, am)
+
+	err = am.handleCertificateRotationEvent(context.Background(), certs.CertificateRotationEvent{Identity: identity.NumericIdentity(10)})
+	assert.ErrorContains(t, err, "failed to get all auth map entries: failed to list entries")
+}
+
+func Test_authManager_handleCertificateRotationEvent(t *testing.T) {
+	authHandlers := []authHandler{
+		&alwaysPassAuthHandler{},
+	}
+	aMap := &fakeAuthMap{
+		entries: map[authKey]authInfo{
+			{localIdentity: 1, remoteIdentity: 2, remoteNodeID: 1, authType: 100}: {expiration: time.Now()},
+			{localIdentity: 2, remoteIdentity: 3, remoteNodeID: 1, authType: 100}: {expiration: time.Now()},
+			{localIdentity: 3, remoteIdentity: 4, remoteNodeID: 1, authType: 100}: {expiration: time.Now()},
+		},
+	}
+
+	am, err := newAuthManager(authHandlers, aMap, nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, am)
+
+	handleAuthCalled := false
+	am.handleAuthenticationFunc = func(_ *authManager, k authKey, reAuth bool) {
+		handleAuthCalled = true
+		assert.True(t, reAuth)
+		assert.True(t, k.localIdentity == 2 || k.remoteIdentity == 2)
+	}
+
+	err = am.handleCertificateRotationEvent(context.Background(), certs.CertificateRotationEvent{Identity: identity.NumericIdentity(2)})
+	assert.NoError(t, err)
+	assert.True(t, handleAuthCalled)
+}
+
 // Fake IPCache
 type fakeIPCache struct {
 	nodeIdMappings map[uint16]string
@@ -151,6 +219,7 @@ func (r *fakeAuthHandler) subscribeToRotatedIdentities() <-chan certs.Certificat
 type fakeAuthMap struct {
 	entries    map[authKey]authInfo
 	failDelete bool
+	failGet    bool
 }
 
 func (r *fakeAuthMap) Delete(key authKey) error {
@@ -173,10 +242,18 @@ func (r *fakeAuthMap) DeleteIf(predicate func(key authKey, info authInfo) bool)
 }
 
 func (r *fakeAuthMap) All() (map[authKey]authInfo, error) {
+	if r.failGet {
+		return nil, errors.New("failed to list entries")
+	}
+
 	return r.entries, nil
 }
 
 func (r *fakeAuthMap) Get(key authKey) (authInfo, error) {
+	if r.failGet {
+		return authInfo{}, errors.New("failed to get entry")
+	}
+
 	v, ok := r.entries[key]
 	if !ok {
 		return authInfo{}, errors.New("authinfo not available")