Implemented a comprehensive secure authentication mechanism for inter-node cluster communication with the following features: 1. Global Cluster Secret (GCS) - Auto-generated cryptographically secure random secret (256-bit) - Configurable via YAML config file - Shared across all cluster nodes for authentication 2. Cluster Authentication Middleware - Validates X-Cluster-Secret and X-Node-ID headers - Applied to all cluster endpoints (/members/*, /merkle_tree/*, /kv_range) - Comprehensive logging of authentication attempts 3. Authenticated HTTP Client - Custom HTTP client with cluster auth headers - TLS support with configurable certificate verification - Protocol-aware (http/https based on TLS settings) 4. Secure Bootstrap Endpoint - New /auth/cluster-bootstrap endpoint - Protected by JWT authentication with admin scope - Allows new nodes to securely obtain cluster secret 5. Updated Cluster Communication - All gossip protocol requests include auth headers - All Merkle tree sync requests include auth headers - All data replication requests include auth headers 6. Configuration - cluster_secret: Shared secret (auto-generated if not provided) - cluster_tls_enabled: Enable TLS for inter-node communication - cluster_tls_cert_file: Path to TLS certificate - cluster_tls_key_file: Path to TLS private key - cluster_tls_skip_verify: Skip TLS verification (testing only) This implementation addresses the security vulnerability of unprotected cluster endpoints and provides a flexible, secure approach to protecting internal cluster communication while allowing for automated node bootstrapping. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
113 lines
2.8 KiB
Go
113 lines
2.8 KiB
Go
package storage
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
badger "github.com/dgraph-io/badger/v4"
|
|
"github.com/sirupsen/logrus"
|
|
|
|
"kvs/types"
|
|
)
|
|
|
|
// StorageService handles all BadgerDB operations and data management
|
|
type StorageService struct {
|
|
db *badger.DB
|
|
config *types.Config
|
|
compressionSvc *CompressionService
|
|
logger *logrus.Logger
|
|
}
|
|
|
|
// NewStorageService creates a new storage service
|
|
func NewStorageService(db *badger.DB, config *types.Config, logger *logrus.Logger) (*StorageService, error) {
|
|
var compressionSvc *CompressionService
|
|
var err error
|
|
|
|
// Initialize compression if enabled
|
|
if config.CompressionEnabled {
|
|
compressionSvc, err = NewCompressionService()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to initialize compression: %v", err)
|
|
}
|
|
}
|
|
|
|
return &StorageService{
|
|
db: db,
|
|
config: config,
|
|
compressionSvc: compressionSvc,
|
|
logger: logger,
|
|
}, nil
|
|
}
|
|
|
|
// Close closes the storage service and its resources
|
|
func (s *StorageService) Close() {
|
|
if s.compressionSvc != nil {
|
|
s.compressionSvc.Close()
|
|
}
|
|
}
|
|
|
|
// StoreWithTTL stores data with optional TTL and compression
|
|
func (s *StorageService) StoreWithTTL(txn *badger.Txn, key []byte, data []byte, ttl time.Duration) error {
|
|
var finalData []byte
|
|
var err error
|
|
|
|
// Compress data if compression is enabled
|
|
if s.config.CompressionEnabled && s.compressionSvc != nil {
|
|
finalData, err = s.compressionSvc.CompressData(data)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to compress data: %v", err)
|
|
}
|
|
} else {
|
|
finalData = data
|
|
}
|
|
|
|
entry := badger.NewEntry(key, finalData)
|
|
|
|
// Apply TTL if specified
|
|
if ttl > 0 {
|
|
entry = entry.WithTTL(ttl)
|
|
}
|
|
|
|
return txn.SetEntry(entry)
|
|
}
|
|
|
|
// RetrieveWithDecompression retrieves and decompresses data from BadgerDB
|
|
func (s *StorageService) RetrieveWithDecompression(txn *badger.Txn, key []byte) ([]byte, error) {
|
|
item, err := txn.Get(key)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var compressedData []byte
|
|
err = item.Value(func(val []byte) error {
|
|
compressedData = append(compressedData, val...)
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Decompress data if compression is enabled
|
|
if s.config.CompressionEnabled && s.compressionSvc != nil {
|
|
return s.compressionSvc.DecompressData(compressedData)
|
|
}
|
|
|
|
return compressedData, nil
|
|
}
|
|
|
|
// CompressData compresses data using the compression service
|
|
func (s *StorageService) CompressData(data []byte) ([]byte, error) {
|
|
if !s.config.CompressionEnabled || s.compressionSvc == nil {
|
|
return data, nil
|
|
}
|
|
return s.compressionSvc.CompressData(data)
|
|
}
|
|
|
|
// DecompressData decompresses data using the compression service
|
|
func (s *StorageService) DecompressData(compressedData []byte) ([]byte, error) {
|
|
if !s.config.CompressionEnabled || s.compressionSvc == nil {
|
|
return compressedData, nil
|
|
}
|
|
return s.compressionSvc.DecompressData(compressedData)
|
|
}
|