Add proxy <-> management authentication

2026-10-07 05:59:06 +02:00 · 2026-02-08 14:33:27 +08:00
parent 2f263bf7e6
commit 0a3a9f977d
17 changed files with 1256 additions and 17 deletions
@@ -269,19 +269,27 @@ func (s *ProxyServiceServer) sender(conn *proxyConnection, errChan chan<- error)
 func (s *ProxyServiceServer) SendAccessLog(ctx context.Context, req *proto.SendAccessLogRequest) (*proto.SendAccessLogResponse, error) {
 	accessLog := req.GetLog()

-	log.WithFields(log.Fields{
+	fields := log.Fields{
 		"reverse_proxy_id": accessLog.GetServiceId(),
 		"account_id":       accessLog.GetAccountId(),
 		"host":             accessLog.GetHost(),
-		"path":             accessLog.GetPath(),
-		"method":           accessLog.GetMethod(),
-		"response_code":    accessLog.GetResponseCode(),
-		"duration_ms":      accessLog.GetDurationMs(),
 		"source_ip":        accessLog.GetSourceIp(),
-		"auth_mechanism":   accessLog.GetAuthMechanism(),
-		"user_id":          accessLog.GetUserId(),
-		"auth_success":     accessLog.GetAuthSuccess(),
-	}).Debug("Access log from proxy")
+	}
+	if mechanism := accessLog.GetAuthMechanism(); mechanism != "" {
+		fields["auth_mechanism"] = mechanism
+	}
+	if userID := accessLog.GetUserId(); userID != "" {
+		fields["user_id"] = userID
+	}
+	if !accessLog.GetAuthSuccess() {
+		fields["auth_success"] = false
+	}
+	log.WithFields(fields).Debugf("%s %s %d (%dms)",
+		accessLog.GetMethod(),
+		accessLog.GetPath(),
+		accessLog.GetResponseCode(),
+		accessLog.GetDurationMs(),
+	)

 	logEntry := &accesslogs.AccessLogEntry{}
 	logEntry.FromProto(accessLog)
@@ -0,0 +1,234 @@
+package grpc
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/types"
+)
+
+const (
+	// lastUsedUpdateInterval is the minimum interval between last_used updates for the same token.
+	lastUsedUpdateInterval = time.Minute
+	// lastUsedCleanupInterval is how often stale lastUsed entries are removed.
+	lastUsedCleanupInterval = 2 * time.Minute
+)
+
+type proxyTokenContextKey struct{}
+
+// ProxyTokenContextKey is the typed key used to store validated token info in context.
+var ProxyTokenContextKey = proxyTokenContextKey{}
+
+// proxyTokenID identifies a proxy access token by its database ID.
+type proxyTokenID = string
+
+// proxyTokenStore defines the store interface needed for token validation
+type proxyTokenStore interface {
+	GetProxyAccessTokenByHashedToken(ctx context.Context, lockStrength store.LockingStrength, hashedToken types.HashedProxyToken) (*types.ProxyAccessToken, error)
+	MarkProxyAccessTokenUsed(ctx context.Context, tokenID string) error
+}
+
+// proxyAuthInterceptor holds state for proxy authentication interceptors.
+type proxyAuthInterceptor struct {
+	store          proxyTokenStore
+	failureLimiter *authFailureLimiter
+
+	// lastUsedMu protects lastUsedTimes
+	lastUsedMu    sync.Mutex
+	lastUsedTimes map[proxyTokenID]time.Time
+	cancel        context.CancelFunc
+}
+
+func newProxyAuthInterceptor(tokenStore proxyTokenStore) *proxyAuthInterceptor {
+	ctx, cancel := context.WithCancel(context.Background())
+	i := &proxyAuthInterceptor{
+		store:          tokenStore,
+		failureLimiter: newAuthFailureLimiter(),
+		lastUsedTimes:  make(map[proxyTokenID]time.Time),
+		cancel:         cancel,
+	}
+	go i.lastUsedCleanupLoop(ctx)
+	return i
+}
+
+// NewProxyAuthInterceptors creates gRPC unary and stream interceptors that validate proxy access tokens.
+// They only intercept ProxyService methods. Both interceptors share state for last-used and failure rate limiting.
+// The returned close function must be called on shutdown to stop background goroutines.
+func NewProxyAuthInterceptors(tokenStore proxyTokenStore) (grpc.UnaryServerInterceptor, grpc.StreamServerInterceptor, func()) {
+	interceptor := newProxyAuthInterceptor(tokenStore)
+
+	unary := func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
+		if !strings.HasPrefix(info.FullMethod, "/management.ProxyService/") {
+			return handler(ctx, req)
+		}
+
+		token, err := interceptor.validateProxyToken(ctx)
+		if err != nil {
+			// Log auth failures explicitly; gRPC doesn't log these by default.
+			log.WithContext(ctx).Warnf("proxy auth failed: %v", err)
+			return nil, err
+		}
+
+		ctx = context.WithValue(ctx, ProxyTokenContextKey, token)
+		return handler(ctx, req)
+	}
+
+	stream := func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		if !strings.HasPrefix(info.FullMethod, "/management.ProxyService/") {
+			return handler(srv, ss)
+		}
+
+		token, err := interceptor.validateProxyToken(ss.Context())
+		if err != nil {
+			// Log auth failures explicitly; gRPC doesn't log these by default.
+			log.WithContext(ss.Context()).Warnf("proxy auth failed: %v", err)
+			return err
+		}
+
+		ctx := context.WithValue(ss.Context(), ProxyTokenContextKey, token)
+		wrapped := &wrappedServerStream{
+			ServerStream: ss,
+			ctx:          ctx,
+		}
+
+		return handler(srv, wrapped)
+	}
+
+	return unary, stream, interceptor.close
+}
+
+func (i *proxyAuthInterceptor) validateProxyToken(ctx context.Context) (*types.ProxyAccessToken, error) {
+	clientIP := peerIPFromContext(ctx)
+
+	if clientIP != "" && i.failureLimiter.isLimited(clientIP) {
+		return nil, status.Errorf(codes.ResourceExhausted, "too many failed authentication attempts")
+	}
+
+	token, err := i.doValidateProxyToken(ctx)
+	if err != nil {
+		if clientIP != "" {
+			i.failureLimiter.recordFailure(clientIP)
+		}
+		return nil, err
+	}
+
+	i.maybeUpdateLastUsed(ctx, token.ID)
+
+	return token, nil
+}
+
+func (i *proxyAuthInterceptor) doValidateProxyToken(ctx context.Context) (*types.ProxyAccessToken, error) {
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+	}
+
+	authValues := md.Get("authorization")
+	if len(authValues) == 0 {
+		return nil, status.Errorf(codes.Unauthenticated, "missing authorization header")
+	}
+
+	authValue := authValues[0]
+	if !strings.HasPrefix(authValue, "Bearer ") {
+		return nil, status.Errorf(codes.Unauthenticated, "invalid authorization format")
+	}
+
+	plainToken := types.PlainProxyToken(strings.TrimPrefix(authValue, "Bearer "))
+
+	if err := plainToken.Validate(); err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "invalid token format")
+	}
+
+	token, err := i.store.GetProxyAccessTokenByHashedToken(ctx, store.LockingStrengthNone, plainToken.Hash())
+	if err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "invalid token")
+	}
+
+	// TODO: Enforce AccountID scope for "bring your own proxy" feature.
+	// Currently tokens are management-wide; AccountID field is reserved for future use.
+
+	if !token.IsValid() {
+		return nil, status.Errorf(codes.Unauthenticated, "token expired or revoked")
+	}
+
+	return token, nil
+}
+
+// maybeUpdateLastUsed updates the last_used timestamp if enough time has passed since the last update.
+func (i *proxyAuthInterceptor) maybeUpdateLastUsed(ctx context.Context, tokenID string) {
+	now := time.Now()
+
+	i.lastUsedMu.Lock()
+	lastUpdate, exists := i.lastUsedTimes[tokenID]
+	if exists && now.Sub(lastUpdate) < lastUsedUpdateInterval {
+		i.lastUsedMu.Unlock()
+		return
+	}
+	i.lastUsedTimes[tokenID] = now
+	i.lastUsedMu.Unlock()
+
+	if err := i.store.MarkProxyAccessTokenUsed(ctx, tokenID); err != nil {
+		log.WithContext(ctx).Debugf("failed to mark proxy token as used: %v", err)
+	}
+}
+
+func (i *proxyAuthInterceptor) lastUsedCleanupLoop(ctx context.Context) {
+	ticker := time.NewTicker(lastUsedCleanupInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			i.cleanupStaleLastUsed()
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// cleanupStaleLastUsed removes entries older than 2x the update interval.
+func (i *proxyAuthInterceptor) cleanupStaleLastUsed() {
+	i.lastUsedMu.Lock()
+	defer i.lastUsedMu.Unlock()
+
+	now := time.Now()
+	staleThreshold := 2 * lastUsedUpdateInterval
+	for id, lastUpdate := range i.lastUsedTimes {
+		if now.Sub(lastUpdate) > staleThreshold {
+			delete(i.lastUsedTimes, id)
+		}
+	}
+}
+
+func (i *proxyAuthInterceptor) close() {
+	i.cancel()
+	i.failureLimiter.stop()
+}
+
+// GetProxyTokenFromContext retrieves the validated proxy token from the context
+func GetProxyTokenFromContext(ctx context.Context) *types.ProxyAccessToken {
+	token, ok := ctx.Value(ProxyTokenContextKey).(*types.ProxyAccessToken)
+	if !ok {
+		return nil
+	}
+	return token
+}
+
+// wrappedServerStream wraps a grpc.ServerStream to provide a custom context
+type wrappedServerStream struct {
+	grpc.ServerStream
+	ctx context.Context
+}
+
+func (w *wrappedServerStream) Context() context.Context {
+	return w.ctx
+}
@@ -0,0 +1,134 @@
+package grpc
+
+import (
+	"context"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
+	"golang.org/x/time/rate"
+	"google.golang.org/grpc/peer"
+)
+
+const (
+	// proxyAuthFailureBurst is the maximum number of failed attempts before rate limiting kicks in.
+	proxyAuthFailureBurst = 5
+	// proxyAuthLimiterCleanup is how often stale limiters are removed.
+	proxyAuthLimiterCleanup = 5 * time.Minute
+	// proxyAuthLimiterTTL is how long a limiter is kept after the last failure.
+	proxyAuthLimiterTTL = 15 * time.Minute
+)
+
+// defaultProxyAuthFailureRate is the token replenishment rate for failed auth attempts.
+// One token every 12 seconds = 5 per minute.
+var defaultProxyAuthFailureRate = rate.Every(12 * time.Second)
+
+// clientIP identifies a client by its IP address for rate limiting purposes.
+type clientIP = string
+
+type limiterEntry struct {
+	limiter    *rate.Limiter
+	lastAccess time.Time
+}
+
+// authFailureLimiter tracks per-IP rate limits for failed proxy authentication attempts.
+type authFailureLimiter struct {
+	mu          sync.Mutex
+	limiters    map[clientIP]*limiterEntry
+	failureRate rate.Limit
+	cancel      context.CancelFunc
+}
+
+func newAuthFailureLimiter() *authFailureLimiter {
+	return newAuthFailureLimiterWithRate(defaultProxyAuthFailureRate)
+}
+
+func newAuthFailureLimiterWithRate(failureRate rate.Limit) *authFailureLimiter {
+	ctx, cancel := context.WithCancel(context.Background())
+	l := &authFailureLimiter{
+		limiters:    make(map[clientIP]*limiterEntry),
+		failureRate: failureRate,
+		cancel:      cancel,
+	}
+	go l.cleanupLoop(ctx)
+	return l
+}
+
+// isLimited returns true if the given IP has exhausted its failure budget.
+func (l *authFailureLimiter) isLimited(ip clientIP) bool {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	entry, exists := l.limiters[ip]
+	if !exists {
+		return false
+	}
+
+	return entry.limiter.Tokens() < 1
+}
+
+// recordFailure consumes a token from the rate limiter for the given IP.
+func (l *authFailureLimiter) recordFailure(ip clientIP) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	now := time.Now()
+	entry, exists := l.limiters[ip]
+	if !exists {
+		entry = &limiterEntry{
+			limiter: rate.NewLimiter(l.failureRate, proxyAuthFailureBurst),
+		}
+		l.limiters[ip] = entry
+	}
+	entry.lastAccess = now
+	entry.limiter.Allow()
+}
+
+func (l *authFailureLimiter) cleanupLoop(ctx context.Context) {
+	ticker := time.NewTicker(proxyAuthLimiterCleanup)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			l.cleanup()
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (l *authFailureLimiter) cleanup() {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	now := time.Now()
+	for ip, entry := range l.limiters {
+		if now.Sub(entry.lastAccess) > proxyAuthLimiterTTL {
+			delete(l.limiters, ip)
+		}
+	}
+}
+
+func (l *authFailureLimiter) stop() {
+	l.cancel()
+}
+
+// peerIPFromContext extracts the client IP from the gRPC context.
+// Uses realip (from trusted proxy headers) first, falls back to the transport peer address.
+func peerIPFromContext(ctx context.Context) clientIP {
+	if addr, ok := realip.FromContext(ctx); ok {
+		return addr.String()
+	}
+
+	if p, ok := peer.FromContext(ctx); ok {
+		host, _, err := net.SplitHostPort(p.Addr.String())
+		if err != nil {
+			return p.Addr.String()
+		}
+		return host
+	}
+
+	return ""
+}
@@ -0,0 +1,98 @@
+package grpc
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/time/rate"
+)
+
+func TestAuthFailureLimiter_NotLimitedInitially(t *testing.T) {
+	l := newAuthFailureLimiter()
+	defer l.stop()
+
+	assert.False(t, l.isLimited("192.168.1.1"), "new IP should not be rate limited")
+}
+
+func TestAuthFailureLimiter_LimitedAfterBurst(t *testing.T) {
+	l := newAuthFailureLimiter()
+	defer l.stop()
+
+	ip := "192.168.1.1"
+	for i := 0; i < proxyAuthFailureBurst; i++ {
+		l.recordFailure(ip)
+	}
+
+	assert.True(t, l.isLimited(ip), "IP should be limited after exhausting burst")
+}
+
+func TestAuthFailureLimiter_DifferentIPsIndependent(t *testing.T) {
+	l := newAuthFailureLimiter()
+	defer l.stop()
+
+	for i := 0; i < proxyAuthFailureBurst; i++ {
+		l.recordFailure("192.168.1.1")
+	}
+
+	assert.True(t, l.isLimited("192.168.1.1"))
+	assert.False(t, l.isLimited("192.168.1.2"), "different IP should not be affected")
+}
+
+func TestAuthFailureLimiter_RecoveryOverTime(t *testing.T) {
+	l := newAuthFailureLimiterWithRate(rate.Limit(100)) // 100 tokens/sec for fast recovery
+	defer l.stop()
+
+	ip := "10.0.0.1"
+
+	// Exhaust burst
+	for i := 0; i < proxyAuthFailureBurst; i++ {
+		l.recordFailure(ip)
+	}
+	require.True(t, l.isLimited(ip))
+
+	// Wait for token replenishment
+	time.Sleep(50 * time.Millisecond)
+
+	assert.False(t, l.isLimited(ip), "should recover after tokens replenish")
+}
+
+func TestAuthFailureLimiter_Cleanup(t *testing.T) {
+	l := newAuthFailureLimiter()
+	defer l.stop()
+
+	l.recordFailure("10.0.0.1")
+
+	l.mu.Lock()
+	require.Len(t, l.limiters, 1)
+	// Backdate the entry so it looks stale
+	l.limiters["10.0.0.1"].lastAccess = time.Now().Add(-proxyAuthLimiterTTL - time.Minute)
+	l.mu.Unlock()
+
+	l.cleanup()
+
+	l.mu.Lock()
+	assert.Empty(t, l.limiters, "stale entries should be cleaned up")
+	l.mu.Unlock()
+}
+
+func TestAuthFailureLimiter_CleanupKeepsFresh(t *testing.T) {
+	l := newAuthFailureLimiter()
+	defer l.stop()
+
+	l.recordFailure("10.0.0.1")
+	l.recordFailure("10.0.0.2")
+
+	l.mu.Lock()
+	// Only backdate one entry
+	l.limiters["10.0.0.1"].lastAccess = time.Now().Add(-proxyAuthLimiterTTL - time.Minute)
+	l.mu.Unlock()
+
+	l.cleanup()
+
+	l.mu.Lock()
+	assert.Len(t, l.limiters, 1, "only stale entries should be removed")
+	assert.Contains(t, l.limiters, "10.0.0.2")
+	l.mu.Unlock()
+}