diff --git a/AGENTS.md b/AGENTS.md index d4c1eb49fe..470f8569e6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,7 +26,7 @@ - Functions: use camel case and **do not** include `_` (e.g. `getPartitionNum`, not `get_partition_num`). - Variables: use lowerCamelCase (e.g. `flushInterval`, not `flush_interval`). - Logging: structured logs via `github.com/pingcap/log` + `zap` fields; message strings should **not** include function names and should avoid `-` (use spaces instead). -- Errors: when an error comes from a third party/library call, wrap it immediately with `errors.Trace(err)` or `errors.WrapError(...)` to attach a stack trace; upstream callers should propagate wrapped errors without wrapping again. +- Errors: when an error comes from a third party/library call, wrap it immediately with `errors.Trace(err)` or `errors.WrapError(...)` to attach a stack trace; upstream callers should propagate wrapped errors without wrapping again. Avoid using `errors.New` to create error objects; instead, utilize the predefined objects available in the `cerrors` package. ## Testing Guidelines diff --git a/pkg/encryption/encryption_manager.go b/pkg/encryption/encryption_manager.go new file mode 100644 index 0000000000..4fea198626 --- /dev/null +++ b/pkg/encryption/encryption_manager.go @@ -0,0 +1,199 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package encryption + +import ( + "context" + + "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/config" + cerrors "github.com/pingcap/ticdc/pkg/errors" + "go.uber.org/zap" +) + +// EncryptionManager is the main interface for encryption/decryption operations +type EncryptionManager interface { + // EncryptData encrypts data for a keyspace + // Returns encrypted data with header, or original data if encryption is not enabled + EncryptData(ctx context.Context, keyspaceID uint32, data []byte) ([]byte, error) + + // DecryptData decrypts data for a keyspace + // Automatically detects if data is encrypted and handles accordingly + DecryptData(ctx context.Context, keyspaceID uint32, encryptedData []byte) ([]byte, error) +} + +type encryptionManager struct { + metaManager EncryptionMetaManager +} + +// NewEncryptionManager creates a new encryption manager +func NewEncryptionManager(metaManager EncryptionMetaManager) EncryptionManager { + return &encryptionManager{ + metaManager: metaManager, + } +} + +// EncryptData encrypts data for a keyspace +func (m *encryptionManager) EncryptData(ctx context.Context, keyspaceID uint32, data []byte) ([]byte, error) { + allowDegrade := true + serverCfg := config.GetGlobalServerConfig() + if serverCfg != nil && serverCfg.Encryption != nil { + allowDegrade = serverCfg.Encryption.AllowDegradeOnError + } + + // Get current data key, key ID and version together to avoid mismatch when keys rotate. + dataKey, currentDataKeyID, version, err := m.metaManager.GetCurrentDataKey(ctx, keyspaceID) + if err != nil { + if allowDegrade { + log.Warn("failed to get current data key, degrade to plaintext", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + return data, nil + } + log.Error("failed to get current data key", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + return nil, cerrors.ErrEncryptionFailed.Wrap(err) + } + + if len(dataKey) == 0 { + log.Debug("encryption not enabled for keyspace", + zap.Uint32("keyspaceID", keyspaceID)) + return data, nil + } + + cipherImpl := NewAES256CTRCipher() + + // Generate IV + iv, err := GenerateIV(cipherImpl.IVSize()) + if err != nil { + log.Error("failed to generate IV", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + return nil, cerrors.ErrEncryptionFailed.Wrap(err) + } + + // Encrypt data + encryptedData, err := cipherImpl.Encrypt(data, dataKey, iv) + if err != nil { + log.Error("failed to encrypt data", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + return nil, cerrors.ErrEncryptionFailed.Wrap(err) + } + + // Prepend IV to encrypted data + encryptedWithIV := make([]byte, len(iv)+len(encryptedData)) + copy(encryptedWithIV, iv) + copy(encryptedWithIV[len(iv):], encryptedData) + + // Encode with encryption header + result, err := EncodeEncryptedData(encryptedWithIV, version, currentDataKeyID) + if err != nil { + log.Error("failed to encode encrypted data", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint8("version", version), + zap.Binary("dataKeyID", []byte(currentDataKeyID)), + zap.Error(err)) + return nil, cerrors.ErrEncryptionFailed.Wrap(err) + } + + log.Debug("data encrypted successfully", + zap.Uint32("keyspaceID", keyspaceID), + zap.String("dataKeyID", currentDataKeyID), + zap.Int("originalSize", len(data)), + zap.Int("encryptedSize", len(result))) + + return result, nil +} + +// DecryptData decrypts data for a keyspace +func (m *encryptionManager) DecryptData(ctx context.Context, keyspaceID uint32, encryptedData []byte) ([]byte, error) { + // Check if data is encrypted + if !IsEncrypted(encryptedData) { + // Data is not encrypted, return as-is (backward compatibility) + log.Debug("data is not encrypted", + zap.Uint32("keyspaceID", keyspaceID)) + return encryptedData, nil + } + + // Decode encryption header + version, dataKeyID, dataWithIV, err := DecodeEncryptedData(encryptedData) + if err != nil { + log.Warn("failed to decode encrypted data header", + zap.Uint32("keyspaceID", keyspaceID), + zap.Int("encryptedSize", len(encryptedData)), + zap.Error(err)) + return nil, cerrors.ErrDecryptionFailed.Wrap(err) + } + + if version == VersionUnencrypted { + // Should not happen if IsEncrypted returned true, but handle it anyway + return dataWithIV, nil + } + + dataKey, err := m.metaManager.GetDataKey(ctx, keyspaceID, dataKeyID) + if err != nil { + log.Warn("failed to get data key for decryption", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint8("version", version), + zap.Binary("dataKeyID", []byte(dataKeyID)), + zap.Error(err)) + return nil, cerrors.ErrDecryptionFailed.Wrap(err) + } + + if len(dataKey) == 0 { + log.Warn("data key is empty for decryption", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint8("version", version), + zap.Binary("dataKeyID", []byte(dataKeyID))) + return nil, cerrors.ErrDecryptionFailed.GenWithStackByArgs("data key is empty") + } + + cipherImpl := NewAES256CTRCipher() + + // Extract IV from the beginning of data + ivSize := cipherImpl.IVSize() + if len(dataWithIV) < ivSize { + log.Warn("encrypted data too short for IV", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint8("version", version), + zap.Binary("dataKeyID", []byte(dataKeyID)), + zap.Int("dataWithIVSize", len(dataWithIV)), + zap.Int("expectedIVSize", ivSize)) + return nil, cerrors.ErrDecryptionFailed.GenWithStackByArgs("data too short for IV") + } + + iv := dataWithIV[:ivSize] + encryptedDataOnly := dataWithIV[ivSize:] + + // Decrypt data + plaintext, err := cipherImpl.Decrypt(encryptedDataOnly, dataKey, iv) + if err != nil { + log.Warn("failed to decrypt data", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint8("version", version), + zap.Binary("dataKeyID", []byte(dataKeyID)), + zap.Error(err)) + return nil, cerrors.ErrDecryptionFailed.Wrap(err) + } + + log.Debug("data decrypted successfully", + zap.Uint32("keyspaceID", keyspaceID), + zap.String("dataKeyID", dataKeyID), + zap.Int("encryptedSize", len(encryptedData)), + zap.Int("plaintextSize", len(plaintext))) + + return plaintext, nil +} diff --git a/pkg/encryption/encryption_manager_test.go b/pkg/encryption/encryption_manager_test.go new file mode 100644 index 0000000000..9d87895c89 --- /dev/null +++ b/pkg/encryption/encryption_manager_test.go @@ -0,0 +1,154 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package encryption + +import ( + "bytes" + "context" + "testing" + + "github.com/pingcap/ticdc/pkg/config" + cerrors "github.com/pingcap/ticdc/pkg/errors" + "github.com/stretchr/testify/require" +) + +type mockMetaManager struct { + currentKey []byte + currentKeyID string + version byte + currentKeyErr error + dataKeys map[string][]byte +} + +func (m *mockMetaManager) IsEncryptionEnabled(ctx context.Context, keyspaceID uint32) bool { + return true +} + +func (m *mockMetaManager) GetCurrentDataKey(ctx context.Context, keyspaceID uint32) ([]byte, string, byte, error) { + return m.currentKey, m.currentKeyID, m.version, m.currentKeyErr +} + +func (m *mockMetaManager) GetDataKey(ctx context.Context, keyspaceID uint32, dataKeyID string) ([]byte, error) { + if m.dataKeys == nil { + if m.currentKeyID == dataKeyID && len(m.currentKey) > 0 { + return m.currentKey, nil + } + return nil, cerrors.ErrDataKeyNotFound.GenWithStackByArgs("data key not found") + } + key, ok := m.dataKeys[dataKeyID] + if !ok { + return nil, cerrors.ErrDataKeyNotFound.GenWithStackByArgs("data key not found") + } + return key, nil +} + +func (m *mockMetaManager) Start(ctx context.Context) error { return nil } +func (m *mockMetaManager) Stop() {} + +func setAllowDegradeOnError(t *testing.T, allow bool) func() { + t.Helper() + original := config.GetGlobalServerConfig().Clone() + updated := original.Clone() + updated.Encryption.AllowDegradeOnError = allow + config.StoreGlobalServerConfig(updated) + return func() { + config.StoreGlobalServerConfig(original) + } +} + +func TestEncryptDataAllowDegradeOnError(t *testing.T) { + restore := setAllowDegradeOnError(t, true) + defer restore() + + meta := &mockMetaManager{ + currentKeyErr: cerrors.ErrEncryptionFailed.GenWithStackByArgs("boom"), + } + manager := NewEncryptionManager(meta) + input := []byte("payload") + + output, err := manager.EncryptData(context.Background(), 1, input) + require.NoError(t, err) + require.Equal(t, input, output) +} + +func TestEncryptDataDisallowDegradeOnError(t *testing.T) { + restore := setAllowDegradeOnError(t, false) + defer restore() + + meta := &mockMetaManager{ + currentKeyErr: cerrors.ErrEncryptionFailed.GenWithStackByArgs("boom"), + } + manager := NewEncryptionManager(meta) + _, err := manager.EncryptData(context.Background(), 1, []byte("payload")) + require.Error(t, err) +} + +func TestEncryptDataDisabledSkipsEncryption(t *testing.T) { + restore := setAllowDegradeOnError(t, false) + defer restore() + + meta := &mockMetaManager{} + manager := NewEncryptionManager(meta) + input := []byte("payload") + + output, err := manager.EncryptData(context.Background(), 1, input) + require.NoError(t, err) + require.Equal(t, input, output) +} + +func TestEncryptDecryptRoundTrip(t *testing.T) { + restore := setAllowDegradeOnError(t, false) + defer restore() + + key := bytes.Repeat([]byte{0x11}, 32) + meta := &mockMetaManager{ + currentKey: key, + currentKeyID: "K01", + version: 0x01, + } + manager := NewEncryptionManager(meta) + + input := []byte("round-trip-payload") + encrypted, err := manager.EncryptData(context.Background(), 1, input) + require.NoError(t, err) + require.NotEqual(t, input, encrypted) + require.True(t, IsEncrypted(encrypted)) + + decrypted, err := manager.DecryptData(context.Background(), 1, encrypted) + require.NoError(t, err) + require.Equal(t, input, decrypted) +} + +func TestEncryptDecryptRoundTripWithAES128Key(t *testing.T) { + restore := setAllowDegradeOnError(t, false) + defer restore() + + key := bytes.Repeat([]byte{0x22}, 16) + meta := &mockMetaManager{ + currentKey: key, + currentKeyID: "K02", + version: 0x01, + } + manager := NewEncryptionManager(meta) + + input := []byte("round-trip-with-16-byte-key") + encrypted, err := manager.EncryptData(context.Background(), 1, input) + require.NoError(t, err) + require.NotEqual(t, input, encrypted) + require.True(t, IsEncrypted(encrypted)) + + decrypted, err := manager.DecryptData(context.Background(), 1, encrypted) + require.NoError(t, err) + require.Equal(t, input, decrypted) +} diff --git a/pkg/encryption/manager.go b/pkg/encryption/manager.go new file mode 100644 index 0000000000..d5c8cd2d66 --- /dev/null +++ b/pkg/encryption/manager.go @@ -0,0 +1,532 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "sync" + "time" + + "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/encryption/kms" + cerrors "github.com/pingcap/ticdc/pkg/errors" + "go.uber.org/zap" +) + +// EncryptionMetaManager manages encryption metadata for keyspaces +type EncryptionMetaManager interface { + // IsEncryptionEnabled checks if encryption is enabled for a keyspace + IsEncryptionEnabled(ctx context.Context, keyspaceID uint32) bool + + // GetCurrentDataKey gets the current data key for a keyspace. + // It returns the plaintext data key, the 3-byte data key ID used in the encryption header, + // and the encryption format version (derived from current.data_key_id & 0xFF). + // + // Returning key and key ID together avoids mismatches if TiKV rotates keys between calls. + GetCurrentDataKey(ctx context.Context, keyspaceID uint32) (dataKey []byte, dataKeyID string, version byte, err error) + + // GetDataKey gets a data key by ID. + GetDataKey(ctx context.Context, keyspaceID uint32, dataKeyID string) ([]byte, error) + + // Start starts the background refresh goroutine + Start(ctx context.Context) error + + // Stop stops the background refresh goroutine + Stop() +} + +type encryptionMetaManager struct { + tikvClient TiKVEncryptionClient + kmsClient kms.KMSClient + + metaCache map[uint32]*cachedMeta + metaMu sync.RWMutex + dataKeyCache map[uint32]map[string]*cachedDataKey + dataKeyMu sync.RWMutex + + ttl time.Duration + refreshInterval time.Duration + stopCh chan struct{} + stopOnce sync.Once + wg sync.WaitGroup +} + +type cachedMeta struct { + meta *EncryptionMeta + timestamp time.Time +} + +type cachedDataKey struct { + key []byte + timestamp time.Time +} + +// NewEncryptionMetaManager creates a new encryption meta manager +func NewEncryptionMetaManager(tikvClient TiKVEncryptionClient, kmsClient kms.KMSClient) EncryptionMetaManager { + return &encryptionMetaManager{ + tikvClient: tikvClient, + kmsClient: kmsClient, + metaCache: make(map[uint32]*cachedMeta), + dataKeyCache: make(map[uint32]map[string]*cachedDataKey), + ttl: 1 * time.Hour, // Default TTL: 1 hour + refreshInterval: 1 * time.Hour, // Default refresh interval: 1 hour + stopCh: make(chan struct{}), + } +} + +// IsEncryptionEnabled checks if encryption is enabled for a keyspace +func (m *encryptionMetaManager) IsEncryptionEnabled(ctx context.Context, keyspaceID uint32) bool { + meta, err := m.getMeta(ctx, keyspaceID) + if err != nil { + log.Warn("failed to get encryption meta", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + // If we can't get meta, encryption is not enabled + return false + } + return meta != nil +} + +// GetCurrentDataKey gets the current data key for a keyspace +func (m *encryptionMetaManager) GetCurrentDataKey(ctx context.Context, keyspaceID uint32) ([]byte, string, byte, error) { + meta, err := m.getMeta(ctx, keyspaceID) + if err != nil { + log.Warn("failed to get encryption meta for current data key", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + return nil, "", 0, err + } + + if meta == nil { + return nil, "", 0, nil + } + + if meta.Current == nil || meta.Current.DataKeyId == 0 { + log.Warn("encryption meta current data key ID is empty", + zap.Uint32("keyspaceID", keyspaceID)) + return nil, "", 0, cerrors.ErrDataKeyNotFound.GenWithStackByArgs("current data key ID is empty") + } + + currentKeyID, err := encodeDataKeyID24BE(meta.Current.DataKeyId) + if err != nil { + log.Warn("failed to encode current data key ID", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("dataKeyID", meta.Current.DataKeyId), + zap.Error(err)) + return nil, "", 0, err + } + + version := byte(meta.Current.DataKeyId & 0xFF) + if version == VersionUnencrypted { + log.Warn("invalid encryption meta version derived from current data key ID", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("dataKeyID", meta.Current.DataKeyId)) + return nil, "", 0, cerrors.ErrEncryptionFailed.GenWithStackByArgs("version must be non-zero") + } + + // Check cache first. + m.dataKeyMu.RLock() + if keyspaceCache, ok := m.dataKeyCache[keyspaceID]; ok { + if cached, ok := keyspaceCache[currentKeyID]; ok { + if time.Since(cached.timestamp) < m.ttl { + key := make([]byte, len(cached.key)) + copy(key, cached.key) + m.dataKeyMu.RUnlock() + return key, currentKeyID, version, nil + } + } + } + m.dataKeyMu.RUnlock() + + dataKey, ok := meta.DataKeys[meta.Current.DataKeyId] + if !ok { + log.Warn("current data key not found in encryption meta", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("dataKeyID", meta.Current.DataKeyId)) + return nil, "", 0, cerrors.ErrDataKeyNotFound.GenWithStackByArgs("current data key not found") + } + + plaintextKey, err := m.decryptDataKey(ctx, meta.MasterKey, dataKey.Ciphertext) + if err != nil { + log.Warn("failed to decrypt current data key", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("dataKeyID", meta.Current.DataKeyId), + zap.Binary("dataKeyIDBytes", []byte(currentKeyID)), + zap.String("kmsVendor", safeKMSVendor(meta.MasterKey)), + zap.String("cmekID", safeCMEKID(meta.MasterKey)), + zap.Error(err)) + return nil, "", 0, err + } + + m.dataKeyMu.Lock() + if m.dataKeyCache[keyspaceID] == nil { + m.dataKeyCache[keyspaceID] = make(map[string]*cachedDataKey) + } + m.dataKeyCache[keyspaceID][currentKeyID] = &cachedDataKey{ + key: plaintextKey, + timestamp: time.Now(), + } + m.dataKeyMu.Unlock() + + return plaintextKey, currentKeyID, version, nil +} + +// GetDataKey gets a data key by ID. +func (m *encryptionMetaManager) GetDataKey(ctx context.Context, keyspaceID uint32, dataKeyID string) ([]byte, error) { + // Check cache first + m.dataKeyMu.RLock() + if keyspaceCache, ok := m.dataKeyCache[keyspaceID]; ok { + if cached, ok := keyspaceCache[dataKeyID]; ok { + // Check if cache is still valid + if time.Since(cached.timestamp) < m.ttl { + key := make([]byte, len(cached.key)) + copy(key, cached.key) + m.dataKeyMu.RUnlock() + return key, nil + } + } + } + m.dataKeyMu.RUnlock() + + // Get meta to find the data key + meta, err := m.getMeta(ctx, keyspaceID) + if err != nil { + log.Warn("failed to get encryption meta for data key", + zap.Uint32("keyspaceID", keyspaceID), + zap.Binary("dataKeyID", []byte(dataKeyID)), + zap.Error(err)) + return nil, err + } + + if meta == nil { + log.Warn("encryption not enabled when looking up data key", + zap.Uint32("keyspaceID", keyspaceID), + zap.Binary("dataKeyID", []byte(dataKeyID))) + return nil, cerrors.ErrDataKeyNotFound.GenWithStackByArgs("encryption not enabled") + } + + id, err := decodeDataKeyID24BE(dataKeyID) + if err != nil { + log.Warn("failed to decode data key ID", + zap.Uint32("keyspaceID", keyspaceID), + zap.Binary("dataKeyID", []byte(dataKeyID)), + zap.Error(err)) + return nil, err + } + + dataKey, ok := meta.DataKeys[id] + if !ok { + log.Warn("data key not found in encryption meta", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("dataKeyID", id), + zap.Binary("dataKeyIDBytes", []byte(dataKeyID))) + return nil, cerrors.ErrDataKeyNotFound.GenWithStackByArgs("data key not found: " + dataKeyID) + } + + // Decrypt the data key using master key + plaintextKey, err := m.decryptDataKey(ctx, meta.MasterKey, dataKey.Ciphertext) + if err != nil { + log.Warn("failed to decrypt data key", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("dataKeyID", id), + zap.Binary("dataKeyIDBytes", []byte(dataKeyID)), + zap.String("kmsVendor", safeKMSVendor(meta.MasterKey)), + zap.String("cmekID", safeCMEKID(meta.MasterKey)), + zap.Error(err)) + return nil, err + } + + // Cache the decrypted key + m.dataKeyMu.Lock() + if m.dataKeyCache[keyspaceID] == nil { + m.dataKeyCache[keyspaceID] = make(map[string]*cachedDataKey) + } + m.dataKeyCache[keyspaceID][dataKeyID] = &cachedDataKey{ + key: plaintextKey, + timestamp: time.Now(), + } + m.dataKeyMu.Unlock() + + return plaintextKey, nil +} + +// getMeta gets encryption metadata, with caching +func (m *encryptionMetaManager) getMeta(ctx context.Context, keyspaceID uint32) (*EncryptionMeta, error) { + // Check cache first + m.metaMu.RLock() + if cached, ok := m.metaCache[keyspaceID]; ok { + // Check if cache is still valid + if time.Since(cached.timestamp) < m.ttl { + meta := cached.meta + m.metaMu.RUnlock() + if meta == nil { + log.Debug("using cached empty encryption meta", + zap.Uint32("keyspaceID", keyspaceID)) + } else { + log.Debug("using cached encryption meta", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("metaKeyspaceID", meta.KeyspaceId), + zap.Uint32("currentDataKeyID", meta.Current.DataKeyId), + zap.Uint8("version", byte(meta.Current.DataKeyId&0xFF)), + zap.Int("dataKeyCount", len(meta.DataKeys))) + } + return meta, nil + } + } + m.metaMu.RUnlock() + + // Fetch from TiKV + meta, err := m.tikvClient.GetKeyspaceEncryptionMeta(ctx, keyspaceID) + if err != nil { + // If we get ErrEncryptionMetaNotFound, cache nil to avoid repeated lookups + if cerrors.ErrEncryptionMetaNotFound.Equal(err) { + log.Info("encryption meta not found for keyspace", + zap.Uint32("keyspaceID", keyspaceID)) + m.metaMu.Lock() + m.metaCache[keyspaceID] = &cachedMeta{ + meta: nil, + timestamp: time.Now(), + } + m.metaMu.Unlock() + return nil, nil + } + log.Warn("failed to fetch encryption meta from TiKV", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + return nil, err + } + + // Cache the result (including nil if enabled=false) + m.metaMu.Lock() + m.metaCache[keyspaceID] = &cachedMeta{ + meta: meta, + timestamp: time.Now(), + } + m.metaMu.Unlock() + + if meta == nil { + log.Info("encryption meta loaded as empty", + zap.Uint32("keyspaceID", keyspaceID)) + return nil, nil + } + + log.Info("encryption meta loaded", + zap.Uint32("keyspaceID", keyspaceID), + zap.Uint32("metaKeyspaceID", meta.KeyspaceId), + zap.Uint32("currentDataKeyID", meta.Current.DataKeyId), + zap.Uint8("version", byte(meta.Current.DataKeyId&0xFF)), + zap.Int("dataKeyCount", len(meta.DataKeys)), + zap.Int("historyCount", len(meta.History)), + zap.String("kmsVendor", safeKMSVendor(meta.MasterKey)), + zap.String("cmekID", safeCMEKID(meta.MasterKey))) + + return meta, nil +} + +// decryptDataKey decrypts a data key using the master key +func (m *encryptionMetaManager) decryptDataKey(ctx context.Context, masterKey *MasterKey, dataKeyCiphertext []byte) ([]byte, error) { + if masterKey == nil { + log.Warn("failed to decrypt data key: master key is nil") + return nil, cerrors.ErrDecodeFailed.GenWithStackByArgs("master key is nil") + } + + // Decrypt master key from KMS + masterKeyPlaintext, err := m.kmsClient.DecryptMasterKey( + ctx, + masterKey.Ciphertext, + masterKey.CmekId, + masterKey.Vendor, + masterKey.Region, + masterKey.Endpoint, + ) + if err != nil { + log.Warn("failed to decrypt master key via KMS", + zap.String("kmsVendor", masterKey.Vendor), + zap.String("cmekID", masterKey.CmekId), + zap.String("region", masterKey.Region), + zap.String("endpoint", masterKey.Endpoint), + zap.Error(err)) + return nil, cerrors.ErrDecodeFailed.Wrap(err) + } + + if len(masterKeyPlaintext) != 32 { + log.Warn("invalid master key plaintext length", + zap.Int("length", len(masterKeyPlaintext))) + return nil, cerrors.ErrDecodeFailed.GenWithStackByArgs("master key plaintext must be 32 bytes") + } + + // Decrypt data key using master key (AES-CTR). + block, err := aes.NewCipher(masterKeyPlaintext) + if err != nil { + log.Warn("failed to initialize AES cipher for data key decryption", + zap.Error(err)) + return nil, cerrors.ErrDecodeFailed.Wrap(err) + } + + // New format: [iv(16)][ciphertext(payload)]. + if len(dataKeyCiphertext) > aes.BlockSize { + key, method, hasMethod, err := decryptDataKeyPayload( + block, dataKeyCiphertext[:aes.BlockSize], dataKeyCiphertext[aes.BlockSize:], true, + ) + if err == nil { + fields := []zap.Field{ + zap.String("format", "iv_prefixed"), + zap.Int("ciphertextLen", len(dataKeyCiphertext)), + zap.Int("keyLen", len(key)), + } + if hasMethod { + fields = append(fields, zap.Uint8("method", method)) + } + log.Debug("decoded data key ciphertext", fields...) + return key, nil + } + log.Debug("failed to decode iv prefixed data key ciphertext, fallback to legacy format", + zap.Int("ciphertextLen", len(dataKeyCiphertext)), + zap.Error(err)) + } + + // Legacy format: ciphertext only, decrypted with zero IV. + key, method, hasMethod, err := decryptDataKeyPayload( + block, make([]byte, aes.BlockSize), dataKeyCiphertext, false, + ) + if err != nil { + log.Warn("invalid data key ciphertext", + zap.Int("length", len(dataKeyCiphertext)), + zap.Error(err)) + return nil, cerrors.ErrDecodeFailed.Wrap(err) + } + fields := []zap.Field{ + zap.String("format", "legacy_zero_iv"), + zap.Int("ciphertextLen", len(dataKeyCiphertext)), + zap.Int("keyLen", len(key)), + } + if hasMethod { + fields = append(fields, zap.Uint8("method", method)) + } + log.Debug("decoded data key ciphertext", fields...) + return key, nil +} + +func decryptDataKeyPayload(block cipher.Block, iv []byte, ciphertext []byte, requireMethodPrefix bool) ([]byte, byte, bool, error) { + if len(iv) != aes.BlockSize { + return nil, 0, false, cerrors.ErrDecodeFailed.GenWithStackByArgs("iv must be 16 bytes") + } + stream := cipher.NewCTR(block, iv) + plaintext := make([]byte, len(ciphertext)) + stream.XORKeyStream(plaintext, ciphertext) + + key, method, hasMethod, err := parsePlaintextDataKey(plaintext, requireMethodPrefix) + if err != nil { + return nil, 0, false, err + } + return key, method, hasMethod, nil +} + +func parsePlaintextDataKey(plaintext []byte, requireMethodPrefix bool) ([]byte, byte, bool, error) { + if requireMethodPrefix { + switch len(plaintext) { + case 17, 25, 33: + method := plaintext[0] + key := make([]byte, len(plaintext)-1) + copy(key, plaintext[1:]) + return key, method, true, nil + default: + return nil, 0, false, cerrors.ErrDecodeFailed.GenWithStackByArgs("invalid data key plaintext length") + } + } + + switch len(plaintext) { + case 16, 24, 32: + key := make([]byte, len(plaintext)) + copy(key, plaintext) + return key, 0, false, nil + default: + return nil, 0, false, cerrors.ErrDecodeFailed.GenWithStackByArgs("invalid data key plaintext length") + } +} + +func safeKMSVendor(masterKey *MasterKey) string { + if masterKey == nil { + return "" + } + return masterKey.Vendor +} + +func safeCMEKID(masterKey *MasterKey) string { + if masterKey == nil { + return "" + } + return masterKey.CmekId +} + +// Start starts the background refresh goroutine +func (m *encryptionMetaManager) Start(ctx context.Context) error { + m.wg.Add(1) + go m.refreshLoop(ctx) + return nil +} + +func (m *encryptionMetaManager) Stop() { + m.stopOnce.Do(func() { + close(m.stopCh) + }) + m.wg.Wait() +} + +func (m *encryptionMetaManager) Close() { + m.Stop() +} + +// refreshLoop periodically refreshes encryption metadata +func (m *encryptionMetaManager) refreshLoop(ctx context.Context) { + defer m.wg.Done() + + ticker := time.NewTicker(m.refreshInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-m.stopCh: + return + case <-ticker.C: + m.refreshAll(ctx) + } + } +} + +func (m *encryptionMetaManager) refreshAll(ctx context.Context) { + m.metaMu.RLock() + keyspaceIDs := make([]uint32, 0, len(m.metaCache)) + for keyspaceID := range m.metaCache { + keyspaceIDs = append(keyspaceIDs, keyspaceID) + } + m.metaMu.RUnlock() + + for _, keyspaceID := range keyspaceIDs { + m.metaMu.Lock() + delete(m.metaCache, keyspaceID) + m.metaMu.Unlock() + + _, err := m.getMeta(ctx, keyspaceID) + if err != nil { + log.Warn("failed to refresh encryption meta", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + } + } +} diff --git a/pkg/encryption/manager_test.go b/pkg/encryption/manager_test.go new file mode 100644 index 0000000000..53c9e5bedf --- /dev/null +++ b/pkg/encryption/manager_test.go @@ -0,0 +1,148 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "testing" + + "github.com/pingcap/ticdc/pkg/encryption/kms" + "github.com/stretchr/testify/require" +) + +type staticKMSClient struct { + plaintext []byte +} + +func (c *staticKMSClient) DecryptMasterKey(ctx context.Context, ciphertext []byte, keyID string, vendor string, region string, endpoint string) ([]byte, error) { + return c.plaintext, nil +} + +var _ kms.KMSClient = (*staticKMSClient)(nil) + +type staticTiKVEncryptionClient struct { + meta *EncryptionMeta + err error +} + +func (c *staticTiKVEncryptionClient) GetKeyspaceEncryptionMeta(ctx context.Context, keyspaceID uint32) (*EncryptionMeta, error) { + return c.meta, c.err +} + +func TestEncryptionMetaManagerDecryptDataKeyUsesZeroIV(t *testing.T) { + t.Parallel() + + masterKeyPlaintext := make([]byte, 32) + for i := range masterKeyPlaintext { + masterKeyPlaintext[i] = byte(i + 1) + } + + dataKeyPlaintext := make([]byte, 32) + for i := range dataKeyPlaintext { + dataKeyPlaintext[i] = byte(0xA0 + i) + } + + block, err := aes.NewCipher(masterKeyPlaintext) + require.NoError(t, err) + iv := make([]byte, aes.BlockSize) + stream := cipher.NewCTR(block, iv) + dataKeyCiphertext := make([]byte, len(dataKeyPlaintext)) + stream.XORKeyStream(dataKeyCiphertext, dataKeyPlaintext) + + dataKeyID := uint32(0x4b3031) // "K01" + + meta := &EncryptionMeta{ + KeyspaceId: 1, + Current: &EncryptionEpoch{ + FileId: 1, + DataKeyId: dataKeyID, + CreatedAt: 0, + }, + MasterKey: &MasterKey{ + Vendor: "aws-kms", + CmekId: "cmek-1", + Region: "us-west-1", + Ciphertext: []byte{1, 2, 3}, + }, + DataKeys: map[uint32]*DataKey{ + dataKeyID: {Ciphertext: dataKeyCiphertext}, + }, + } + + tikvClient := &staticTiKVEncryptionClient{meta: meta} + kmsClient := &staticKMSClient{plaintext: masterKeyPlaintext} + mgr := NewEncryptionMetaManager(tikvClient, kmsClient) + + gotKey, err := mgr.GetDataKey(context.Background(), 1, "K01") + require.NoError(t, err) + require.Equal(t, dataKeyPlaintext, gotKey) +} + +func TestEncryptionMetaManagerDecryptDataKeySupportsIVPrefixedCiphertext(t *testing.T) { + t.Parallel() + + masterKeyPlaintext := make([]byte, 32) + for i := range masterKeyPlaintext { + masterKeyPlaintext[i] = byte(i + 1) + } + + // TiKV may store data key payload as: [method(1)][key(16/24/32)]. + // method byte is not guaranteed to be within a small enum range. + method := byte(0xF5) + dataKeyPlaintext := make([]byte, 16) + for i := range dataKeyPlaintext { + dataKeyPlaintext[i] = byte(0xB0 + i) + } + payload := append([]byte{method}, dataKeyPlaintext...) + + block, err := aes.NewCipher(masterKeyPlaintext) + require.NoError(t, err) + iv := []byte("1234567890abcdef") + stream := cipher.NewCTR(block, iv) + payloadCiphertext := make([]byte, len(payload)) + stream.XORKeyStream(payloadCiphertext, payload) + + // New format: [iv(16)][ciphertext(payload)]. + dataKeyCiphertext := append(append([]byte{}, iv...), payloadCiphertext...) + + dataKeyID := uint32(0x4b3131) // "K11" + + meta := &EncryptionMeta{ + KeyspaceId: 1, + Current: &EncryptionEpoch{ + FileId: 1, + DataKeyId: dataKeyID, + CreatedAt: 0, + }, + MasterKey: &MasterKey{ + Vendor: "aws-kms", + CmekId: "cmek-1", + Region: "us-west-1", + Ciphertext: []byte{1, 2, 3}, + }, + DataKeys: map[uint32]*DataKey{ + dataKeyID: {Ciphertext: dataKeyCiphertext}, + }, + } + + tikvClient := &staticTiKVEncryptionClient{meta: meta} + kmsClient := &staticKMSClient{plaintext: masterKeyPlaintext} + mgr := NewEncryptionMetaManager(tikvClient, kmsClient) + + gotKey, err := mgr.GetDataKey(context.Background(), 1, "K11") + require.NoError(t, err) + require.Equal(t, dataKeyPlaintext, gotKey) +} diff --git a/pkg/encryption/mock_tikv_client.go b/pkg/encryption/mock_tikv_client.go new file mode 100644 index 0000000000..7dd005176e --- /dev/null +++ b/pkg/encryption/mock_tikv_client.go @@ -0,0 +1,171 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + + "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/encryption/kms" + cerrors "github.com/pingcap/ticdc/pkg/errors" + "go.uber.org/zap" +) + +// TiKVEncryptionClient is the interface for getting encryption metadata from TiKV +type TiKVEncryptionClient interface { + // GetKeyspaceEncryptionMeta gets the encryption metadata for a keyspace + GetKeyspaceEncryptionMeta(ctx context.Context, keyspaceID uint32) (*EncryptionMeta, error) +} + +// MockTiKVClient is a mock implementation of TiKVEncryptionClient for development and testing +type MockTiKVClient struct { + // metaMap stores mock encryption metadata by keyspace ID + metaMap map[uint32]*EncryptionMeta + // notFoundKeyspaces stores keyspace IDs that should return ErrEncryptionMetaNotFound + notFoundKeyspaces map[uint32]bool +} + +// NewMockTiKVClient creates a new mock TiKV client +func NewMockTiKVClient() *MockTiKVClient { + client := &MockTiKVClient{ + metaMap: make(map[uint32]*EncryptionMeta), + notFoundKeyspaces: make(map[uint32]bool), + } + + // Initialize with some default mock data for testing + client.initDefaultMockData() + + return client +} + +// initDefaultMockData initializes default mock encryption metadata +func (c *MockTiKVClient) initDefaultMockData() { + // Create a mock keyspace with encryption enabled + // Data key IDs carry the encryption format version in their low 8 bits. + mockDataKeyID1 := uint32(0x010001) + mockDataKeyID2 := uint32(0x020001) + + // Generate mock master key plaintext (32 bytes for AES-256) + masterKeyPlaintext := make([]byte, 32) + if _, err := rand.Read(masterKeyPlaintext); err != nil { + log.Panic("failed to generate random master key plaintext", zap.Error(err)) + } + + // Generate mock data key plaintext (32 bytes for AES-256) + dataKey1Plaintext := make([]byte, 32) + if _, err := rand.Read(dataKey1Plaintext); err != nil { + log.Panic("failed to generate random data key plaintext", zap.Error(err)) + } + dataKey2Plaintext := make([]byte, 32) + if _, err := rand.Read(dataKey2Plaintext); err != nil { + log.Panic("failed to generate random data key plaintext", zap.Error(err)) + } + + // Encrypt data keys using master key (AES-256-CTR with zero IV) + block, err := aes.NewCipher(masterKeyPlaintext) + if err != nil { + log.Panic("failed to create AES cipher for data key encryption", zap.Error(err)) + } + iv := make([]byte, aes.BlockSize) + stream := cipher.NewCTR(block, iv) + + dataKey1Ciphertext := make([]byte, len(dataKey1Plaintext)) + stream.XORKeyStream(dataKey1Ciphertext, dataKey1Plaintext) + + // Reset stream by creating a new CTR stream to ensure deterministic encryption per key. + stream = cipher.NewCTR(block, iv) + dataKey2Ciphertext := make([]byte, len(dataKey2Plaintext)) + stream.XORKeyStream(dataKey2Ciphertext, dataKey2Plaintext) + + // Encrypt master key plaintext via mock KMS to generate a realistic ciphertext. + kmsClient := kms.NewMockKMSClient() + masterKeyCiphertext, err := kmsClient.EncryptMasterKey(masterKeyPlaintext) + if err != nil { + log.Panic("failed to encrypt master key via mock KMS", zap.Error(err)) + } + + meta := &EncryptionMeta{ + KeyspaceId: 1, + Current: &EncryptionEpoch{ + FileId: 1, + DataKeyId: mockDataKeyID2, + CreatedAt: 0, + }, + MasterKey: &MasterKey{ + Vendor: "aws-kms", + CmekId: "foobar1", + Region: "us-west-1", + Ciphertext: masterKeyCiphertext, + }, + DataKeys: map[uint32]*DataKey{ + mockDataKeyID1: {Ciphertext: dataKey1Ciphertext}, + mockDataKeyID2: {Ciphertext: dataKey2Ciphertext}, + }, + History: nil, + } + + // Use keyspace ID 1 as default enabled keyspace + c.metaMap[1] = meta + + // Create a mock keyspace with encryption disabled. + c.notFoundKeyspaces[2] = true +} + +// GetKeyspaceEncryptionMeta gets the encryption metadata for a keyspace +func (c *MockTiKVClient) GetKeyspaceEncryptionMeta(ctx context.Context, keyspaceID uint32) (*EncryptionMeta, error) { + // Check if this keyspace should return not found error + if c.notFoundKeyspaces[keyspaceID] { + log.Debug("mock TiKV client: encryption meta not found", + zap.Uint32("keyspaceID", keyspaceID)) + return nil, cerrors.ErrEncryptionMetaNotFound + } + + // Return mock metadata if available + if meta, ok := c.metaMap[keyspaceID]; ok { + log.Debug("mock TiKV client: returning encryption meta", + zap.Uint32("keyspaceID", keyspaceID), + zap.Bool("enabled", meta != nil)) + return meta, nil + } + + // Default behavior: return not found for unknown keyspaces + // This simulates classic architecture or unconfigured encryption + log.Debug("mock TiKV client: encryption meta not found (unknown keyspace)", + zap.Uint32("keyspaceID", keyspaceID)) + return nil, cerrors.ErrEncryptionMetaNotFound +} + +// SetKeyspaceMeta sets mock encryption metadata for a keyspace (for testing) +func (c *MockTiKVClient) SetKeyspaceMeta(keyspaceID uint32, meta *EncryptionMeta) { + c.metaMap[keyspaceID] = meta +} + +// SetKeyspaceNotFound sets a keyspace to return cerrors.ErrEncryptionMetaNotFound (for testing) +func (c *MockTiKVClient) SetKeyspaceNotFound(keyspaceID uint32) { + c.notFoundKeyspaces[keyspaceID] = true +} + +// ClearKeyspaceNotFound clears the not found flag for a keyspace (for testing) +func (c *MockTiKVClient) ClearKeyspaceNotFound(keyspaceID uint32) { + delete(c.notFoundKeyspaces, keyspaceID) +} + +// GetKeyspaceMeta gets the stored mock metadata (for testing) +func (c *MockTiKVClient) GetKeyspaceMeta(keyspaceID uint32) (*EncryptionMeta, bool) { + meta, ok := c.metaMap[keyspaceID] + return meta, ok +} diff --git a/pkg/encryption/mock_tikv_client_test.go b/pkg/encryption/mock_tikv_client_test.go new file mode 100644 index 0000000000..8333c27598 --- /dev/null +++ b/pkg/encryption/mock_tikv_client_test.go @@ -0,0 +1,40 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package encryption + +import ( + "context" + "testing" + + cerrors "github.com/pingcap/ticdc/pkg/errors" + "github.com/stretchr/testify/require" +) + +func TestMockTiKVClientNotFound(t *testing.T) { + cli := NewMockTiKVClient() + meta, err := cli.GetKeyspaceEncryptionMeta(context.Background(), 999) + require.Nil(t, meta) + require.True(t, cerrors.ErrEncryptionMetaNotFound.Equal(err)) +} + +func TestMockTiKVClientEnabledKeyspace(t *testing.T) { + cli := NewMockTiKVClient() + meta, err := cli.GetKeyspaceEncryptionMeta(context.Background(), 1) + require.NoError(t, err) + require.NotNil(t, meta) + require.Equal(t, uint32(1), meta.KeyspaceId) + require.NotNil(t, meta.Current) + require.NotZero(t, meta.Current.DataKeyId) + require.NotEmpty(t, meta.DataKeys) +} diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go index 9794b50b80..16b74d92fb 100644 --- a/pkg/encryption/tikv_http_client.go +++ b/pkg/encryption/tikv_http_client.go @@ -40,12 +40,6 @@ type tikvEncryptionHTTPClient struct { httpTimeout time.Duration } -// TiKVEncryptionClient fetches keyspace-level encryption metadata from TiKV. -// It is consumed by the encryption meta manager in follow-up PRs. -type TiKVEncryptionClient interface { - GetKeyspaceEncryptionMeta(ctx context.Context, keyspaceID uint32) (*EncryptionMeta, error) -} - func NewTiKVEncryptionHTTPClient(pdClient pd.Client, credential *security.Credential) (TiKVEncryptionClient, error) { httpClient, err := httputil.NewClient(credential) if err != nil { @@ -291,7 +285,7 @@ func decodeEncryptionMetaResponseFromProtobuf(body []byte) (*encryptionMetaRespo return nil, errors.Trace(err) } if metaPB.Current == nil && metaPB.MasterKey == nil && len(metaPB.DataKeys) == 0 && len(metaPB.History) == 0 && metaPB.KeyspaceId == 0 { - return nil, errors.New("protobuf payload does not contain encryption meta fields") + return nil, cerrors.ErrDecodeFailed.GenWithStackByArgs("protobuf payload does not contain encryption meta fields") } return metaPB.toEncryptionMetaResponse(), nil } @@ -410,17 +404,3 @@ func truncateBytesForLog(b []byte, max int) string { } return fmt.Sprintf("%s...(truncated, %d bytes total)", string(b[:max]), len(b)) } - -func safeKMSVendor(masterKey *MasterKey) string { - if masterKey == nil { - return "" - } - return masterKey.Vendor -} - -func safeCMEKID(masterKey *MasterKey) string { - if masterKey == nil { - return "" - } - return masterKey.CmekId -}