feat(encryption): add encryption primitives and wire format

Signed-off-by: tenfyzhong <tenfy@tenfy.cn>

Author: tenfyzhong

pkg/encryption/cipher.go

@@ -0,0 +1,107 @@
+// Copyright 2025 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 (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+
+	cerrors "github.com/pingcap/ticdc/pkg/errors"
+)
+
+// Cipher is the interface for encryption/decryption operations
+type Cipher interface {
+	// Encrypt encrypts data using the provided key and IV
+	Encrypt(data, key, iv []byte) ([]byte, error)
+
+	// Decrypt decrypts data using the provided key and IV
+	Decrypt(data, key, iv []byte) ([]byte, error)
+
+	// IVSize returns the required IV size in bytes
+	IVSize() int
+}
+
+// AES256CTRCipher implements AES-CTR encryption for AES key sizes.
+type AES256CTRCipher struct{}
+
+// NewAES256CTRCipher creates a new AES-CTR cipher.
+func NewAES256CTRCipher() *AES256CTRCipher {
+	return &AES256CTRCipher{}
+}
+
+// IVSize returns the IV size for AES-CTR (16 bytes)
+func (c *AES256CTRCipher) IVSize() int {
+	return aes.BlockSize
+}
+
+func isValidAESKeySize(key []byte) bool {
+	switch len(key) {
+	case 16, 24, 32:
+		return true
+	default:
+		return false
+	}
+}
+
+// Encrypt encrypts data using AES-CTR.
+func (c *AES256CTRCipher) Encrypt(data, key, iv []byte) ([]byte, error) {
+	if !isValidAESKeySize(key) {
+		return nil, cerrors.ErrEncryptionFailed.GenWithStackByArgs("key must be 16, 24, or 32 bytes for AES-CTR")
+	}
+	if len(iv) != c.IVSize() {
+		return nil, cerrors.ErrEncryptionFailed.GenWithStackByArgs("IV must be 16 bytes")
+	}
+
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, cerrors.ErrEncryptionFailed.Wrap(err)
+	}
+
+	stream := cipher.NewCTR(block, iv)
+	ciphertext := make([]byte, len(data))
+	stream.XORKeyStream(ciphertext, data)
+
+	return ciphertext, nil
+}
+
+// Decrypt decrypts data using AES-CTR.
+func (c *AES256CTRCipher) Decrypt(data, key, iv []byte) ([]byte, error) {
+	if !isValidAESKeySize(key) {
+		return nil, cerrors.ErrDecryptionFailed.GenWithStackByArgs("key must be 16, 24, or 32 bytes for AES-CTR")
+	}
+	if len(iv) != c.IVSize() {
+		return nil, cerrors.ErrDecryptionFailed.GenWithStackByArgs("IV must be 16 bytes")
+	}
+
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, cerrors.ErrDecryptionFailed.Wrap(err)
+	}
+
+	stream := cipher.NewCTR(block, iv)
+	plaintext := make([]byte, len(data))
+	stream.XORKeyStream(plaintext, data)
+
+	return plaintext, nil
+}
+
+// GenerateIV generates a random IV of the specified size
+func GenerateIV(size int) ([]byte, error) {
+	iv := make([]byte, size)
+	if _, err := rand.Read(iv); err != nil {
+		return nil, cerrors.ErrEncryptionFailed.Wrap(err)
+	}
+	return iv, nil
+}

pkg/encryption/cipher_test.go

@@ -0,0 +1,35 @@
+// Copyright 2025 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 (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAES256CTREncryptDecrypt(t *testing.T) {
+	key := []byte("0123456789abcdef0123456789abcdef") // 32 bytes
+	iv := []byte("1234567890abcdef")                  // 16 bytes
+	plain := []byte("hello world")
+
+	cipherImpl := NewAES256CTRCipher()
+	encrypted, err := cipherImpl.Encrypt(plain, key, iv)
+	require.NoError(t, err)
+	require.NotEqual(t, plain, encrypted)
+
+	decrypted, err := cipherImpl.Decrypt(encrypted, key, iv)
+	require.NoError(t, err)
+	require.Equal(t, plain, decrypted)
+}

pkg/encryption/data_key_id.go

@@ -0,0 +1,34 @@
+// 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 "github.com/pingcap/errors"
+
+// DataKeyID represents a 3-byte data key identifier in the encryption header.
+type DataKeyID [3]byte
+
+// ToString converts DataKeyID to string.
+func (id DataKeyID) ToString() string {
+	return string(id[:])
+}
+
+// DataKeyIDFromString creates DataKeyID from string (must be 3 bytes).
+func DataKeyIDFromString(s string) (DataKeyID, error) {
+	if len(s) != 3 {
+		return DataKeyID{}, errors.New("data key ID must be exactly 3 bytes")
+	}
+	var id DataKeyID
+	copy(id[:], s)
+	return id, nil
+}

pkg/encryption/data_key_id_24be.go

@@ -0,0 +1,32 @@
+// 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 cerrors "github.com/pingcap/ticdc/pkg/errors"
+
+func encodeDataKeyID24BE(id uint32) (string, error) {
+	if id > 0xFFFFFF {
+		return "", cerrors.ErrInvalidDataKeyID.GenWithStackByArgs("data key ID exceeds 24-bit range")
+	}
+	b := [3]byte{byte(id >> 16), byte(id >> 8), byte(id)}
+	return string(b[:]), nil
+}
+
+func decodeDataKeyID24BE(id string) (uint32, error) {
+	if len(id) != 3 {
+		return 0, cerrors.ErrInvalidDataKeyID.GenWithStackByArgs("data key ID must be 3 bytes")
+	}
+	b := []byte(id)
+	return uint32(b[0])<<16 | uint32(b[1])<<8 | uint32(b[2]), nil
+}

pkg/encryption/format.go

@@ -0,0 +1,157 @@
+// Copyright 2025 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 (
+	cerrors "github.com/pingcap/ticdc/pkg/errors"
+)
+
+const (
+	// EncryptionHeaderSize is the size of encryption header (4 bytes)
+	// Format: [version(1 byte)][dataKeyID(3 bytes)]
+	EncryptionHeaderSize = 4
+
+	// VersionUnencrypted indicates data is not encrypted
+	VersionUnencrypted byte = 0x00
+)
+
+// EncryptionHeader represents the 4-byte encryption header
+// Format: [version(1 byte)][dataKeyID(3 bytes)]
+type EncryptionHeader struct {
+	Version   byte
+	DataKeyID [3]byte
+}
+
+// EncodeEncryptedData encodes data with encryption header
+// Format: [version(1)][dataKeyID(3)][encryptedData]
+// The version byte comes from the encryption metadata obtained from TiKV
+func EncodeEncryptedData(data []byte, version byte, dataKeyID string) ([]byte, error) {
+	if len(dataKeyID) != 3 {
+		return nil, cerrors.ErrInvalidDataKeyID.GenWithStackByArgs("data key ID must be 3 bytes")
+	}
+
+	if version == VersionUnencrypted {
+		return nil, cerrors.ErrEncryptionFailed.GenWithStackByArgs("version cannot be 0 for encrypted data")
+	}
+
+	result := make([]byte, EncryptionHeaderSize+len(data))
+	result[0] = version
+	copy(result[1:4], dataKeyID)
+	copy(result[4:], data)
+
+	return result, nil
+}
+
+// DecodeEncryptedData decodes data and extracts encryption header
+// Returns: (version, dataKeyID, encryptedData, error)
+func DecodeEncryptedData(data []byte) (byte, string, []byte, error) {
+	if len(data) < EncryptionHeaderSize {
+		return 0, "", nil, cerrors.ErrDecodeFailed.GenWithStackByArgs("data too short for encryption header")
+	}
+
+	version := data[0]
+	var dataKeyID [3]byte
+	copy(dataKeyID[:], data[1:4])
+	encryptedData := data[4:]
+
+	return version, string(dataKeyID[:]), encryptedData, nil
+}
+
+// IsEncrypted checks if data is encrypted by examining the version byte
+// Data is considered encrypted if version != 0 (VersionUnencrypted)
+// The caller should validate that the version matches expected versions from TiKV metadata
+func IsEncrypted(data []byte) bool {
+	if len(data) < EncryptionHeaderSize {
+		return false
+	}
+	return data[0] != VersionUnencrypted
+}
+
+// IsEncryptedWithVersion checks if data is encrypted with a specific version
+// This is useful when you know the expected version from TiKV metadata
+func IsEncryptedWithVersion(data []byte, expectedVersion byte) bool {
+	if len(data) < EncryptionHeaderSize {
+		return false
+	}
+	return data[0] == expectedVersion
+}
+
+// GetVersion extracts the version byte from data
+// Returns 0 if data is too short
+func GetVersion(data []byte) byte {
+	if len(data) < EncryptionHeaderSize {
+		return 0
+	}
+	return data[0]
+}
+
+// EncodeUnencryptedData encodes unencrypted data with version=0 header
+// This creates a unified format where all new data has the 4-byte header
+func EncodeUnencryptedData(data []byte) []byte {
+	result := make([]byte, EncryptionHeaderSize+len(data))
+	result[0] = VersionUnencrypted
+	// DataKeyID is zero for unencrypted data (3 bytes)
+	result[1] = 0
+	result[2] = 0
+	result[3] = 0
+	copy(result[4:], data)
+	return result
+}
+
+// DecodeUnencryptedData decodes unencrypted data (removes header if present)
+func DecodeUnencryptedData(data []byte) ([]byte, error) {
+	if len(data) < EncryptionHeaderSize {
+		// No header, return as-is (backward compatibility)
+		return data, nil
+	}
+
+	version := data[0]
+	dataKeyID1, dataKeyID2, dataKeyID3 := data[1], data[2], data[3]
+	dataKeyIDIsZero := dataKeyID1 == 0 && dataKeyID2 == 0 && dataKeyID3 == 0
+
+	if version == VersionUnencrypted && dataKeyIDIsZero {
+		// New-format unencrypted data with header, remove header
+		return data[4:], nil
+	}
+
+	// For backward compatibility, treat any other format as legacy unencrypted data
+	// This includes:
+	// - Legacy data without header (any pattern)
+	// - Data that might look like encrypted but is actually legacy
+	// The caller is responsible for ensuring data is not actually encrypted
+	return data, nil
+}
+
+// ExtractDataKeyID extracts the data key ID from encrypted data
+func ExtractDataKeyID(data []byte) (string, error) {
+	if len(data) < EncryptionHeaderSize {
+		return "", cerrors.ErrDecodeFailed.GenWithStackByArgs("data too short")
+	}
+
+	version := data[0]
+	dataKeyID1, dataKeyID2, dataKeyID3 := data[1], data[2], data[3]
+	dataKeyIDIsZero := dataKeyID1 == 0 && dataKeyID2 == 0 && dataKeyID3 == 0
+
+	// Only extract key ID from data that definitively looks like new-format encrypted:
+	// - version != 0 (encrypted data has non-zero version)
+	// - DataKeyID is non-zero (encrypted data always has a valid key ID)
+	if version != VersionUnencrypted && !dataKeyIDIsZero {
+		var keyID [3]byte
+		copy(keyID[:], data[1:4])
+		return string(keyID[:]), nil
+	}
+
+	// Otherwise, this is not encrypted data (legacy data or new-format unencrypted)
+	return "", cerrors.ErrDecodeFailed.GenWithStackByArgs("data is not encrypted")
+}