diff --git a/pkg/encryption/cipher.go b/pkg/encryption/cipher.go
new file mode 100644
index 0000000000..63d64fb0a3
--- /dev/null
+++ b/pkg/encryption/cipher.go
@@ -0,0 +1,107 @@
+// 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 (
+	"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
+}
diff --git a/pkg/encryption/cipher_test.go b/pkg/encryption/cipher_test.go
new file mode 100644
index 0000000000..e980428d77
--- /dev/null
+++ b/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)
+}
diff --git a/pkg/encryption/data_key_id.go b/pkg/encryption/data_key_id.go
new file mode 100644
index 0000000000..8df36878bb
--- /dev/null
+++ b/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 cerrors "github.com/pingcap/ticdc/pkg/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{}, cerrors.ErrInvalidDataKeyID.GenWithStackByArgs("data key ID must be exactly 3 bytes")
+	}
+	var id DataKeyID
+	copy(id[:], s)
+	return id, nil
+}
diff --git a/pkg/encryption/data_key_id_24be.go b/pkg/encryption/data_key_id_24be.go
new file mode 100644
index 0000000000..18a9cad6e6
--- /dev/null
+++ b/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
+}
diff --git a/pkg/encryption/format.go b/pkg/encryption/format.go
new file mode 100644
index 0000000000..db0e4d5392
--- /dev/null
+++ b/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")
+}
diff --git a/pkg/encryption/format_test.go b/pkg/encryption/format_test.go
new file mode 100644
index 0000000000..f37a310b82
--- /dev/null
+++ b/pkg/encryption/format_test.go
@@ -0,0 +1,198 @@
+// 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"
+
+	cerrors "github.com/pingcap/ticdc/pkg/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEncodeEncryptedDataInvalidKey(t *testing.T) {
+	// Key ID must be exactly 3 bytes
+	_, err := EncodeEncryptedData([]byte("payload"), 0x01, "ab")
+	require.Error(t, err)
+	require.True(t, cerrors.ErrInvalidDataKeyID.Equal(err))
+
+	_, err = EncodeEncryptedData([]byte("payload"), 0x01, "abcd")
+	require.Error(t, err)
+	require.True(t, cerrors.ErrInvalidDataKeyID.Equal(err))
+}
+
+func TestEncodeEncryptedDataInvalidVersion(t *testing.T) {
+	// Version cannot be 0 for encrypted data
+	_, err := EncodeEncryptedData([]byte("payload"), VersionUnencrypted, "abc")
+	require.Error(t, err)
+}
+
+func TestEncodeDecodeEncryptedData(t *testing.T) {
+	data := []byte("payload")
+	keyID := "abc" // 3 bytes
+	version := byte(0x01)
+
+	encoded, err := EncodeEncryptedData(data, version, keyID)
+	require.NoError(t, err)
+	require.True(t, IsEncrypted(encoded))
+
+	// Verify version byte is set correctly
+	require.Equal(t, version, encoded[0])
+
+	decodedVersion, decodedKeyID, body, err := DecodeEncryptedData(encoded)
+	require.NoError(t, err)
+	require.Equal(t, version, decodedVersion)
+	require.Equal(t, keyID, decodedKeyID)
+	require.Equal(t, data, body)
+}
+
+func TestEncodeDecodeWithDifferentVersions(t *testing.T) {
+	data := []byte("payload")
+	keyID := "xyz"
+
+	// Test with different version values that might come from TiKV
+	versions := []byte{0x01, 0x02, 0x10, 0xFF}
+
+	for _, version := range versions {
+		encoded, err := EncodeEncryptedData(data, version, keyID)
+		require.NoError(t, err)
+		require.True(t, IsEncrypted(encoded))
+		require.Equal(t, version, GetVersion(encoded))
+
+		decodedVersion, decodedKeyID, body, err := DecodeEncryptedData(encoded)
+		require.NoError(t, err)
+		require.Equal(t, version, decodedVersion)
+		require.Equal(t, keyID, decodedKeyID)
+		require.Equal(t, data, body)
+	}
+}
+
+func TestEncodeUnencryptedData(t *testing.T) {
+	raw := []byte("plain")
+	encoded := EncodeUnencryptedData(raw)
+
+	// Unencrypted data with header should NOT be detected as encrypted
+	// because the version byte is VersionUnencrypted (0x00)
+	require.False(t, IsEncrypted(encoded))
+	require.Equal(t, VersionUnencrypted, encoded[0])
+
+	decoded, err := DecodeUnencryptedData(encoded)
+	require.NoError(t, err)
+	require.Equal(t, raw, decoded)
+}
+
+func TestIsEncryptedWithLegacyData(t *testing.T) {
+	// Legacy unencrypted data (no header) should not be detected as encrypted
+	// because it's too short for the header
+	shortData := []byte("abc")
+	require.False(t, IsEncrypted(shortData))
+
+	// Data with version=0 (first byte is 0x00) is not encrypted
+	unencryptedWithHeader := []byte{0x00, 0x01, 0x02, 0x03, 0x04, 0x05}
+	require.False(t, IsEncrypted(unencryptedWithHeader))
+}
+
+func TestIsEncryptedWithVersionByte(t *testing.T) {
+	// Data with non-zero version byte should be detected as encrypted
+	encryptedData := []byte{0x01, 'a', 'b', 'c', 'd', 'a', 't', 'a'}
+	require.True(t, IsEncrypted(encryptedData))
+
+	// Data with different version values
+	for _, v := range []byte{0x01, 0x02, 0x10, 0xFF} {
+		data := []byte{v, 'a', 'b', 'c', 'd', 'a', 't', 'a'}
+		require.True(t, IsEncrypted(data))
+	}
+
+	// Data too short should not be detected as encrypted
+	shortData := []byte{0x01, 'a', 'b'}
+	require.False(t, IsEncrypted(shortData))
+}
+
+func TestIsEncryptedWithVersion(t *testing.T) {
+	data := []byte{0x05, 'a', 'b', 'c', 'd', 'a', 't', 'a'}
+
+	// Should match when version matches
+	require.True(t, IsEncryptedWithVersion(data, 0x05))
+
+	// Should not match when version doesn't match
+	require.False(t, IsEncryptedWithVersion(data, 0x01))
+	require.False(t, IsEncryptedWithVersion(data, 0x00))
+}
+
+func TestGetVersion(t *testing.T) {
+	// Normal data
+	data := []byte{0x05, 'a', 'b', 'c', 'd', 'a', 't', 'a'}
+	require.Equal(t, byte(0x05), GetVersion(data))
+
+	// Short data returns 0
+	shortData := []byte{0x05, 'a', 'b'}
+	require.Equal(t, byte(0x00), GetVersion(shortData))
+}
+
+func TestDecodeUnencryptedDataBackwardCompatibility(t *testing.T) {
+	// Legacy data without header should be returned as-is
+	// Use data that is too short to have a header (length < 4)
+	legacyData := []byte("legacy")
+	decoded, err := DecodeUnencryptedData(legacyData)
+	require.NoError(t, err)
+	require.Equal(t, legacyData, decoded)
+
+	// Also test with data that has non-zero DataKeyID pattern
+	// This can't be confused with new-format encrypted data (which would have non-zero key ID)
+	// and can't be confused with new-format unencrypted (which has zero key ID)
+	legacyData2 := []byte{0x00, 0x01, 0x02, 0x03, 0x04, 0x05}
+	decoded2, err := DecodeUnencryptedData(legacyData2)
+	require.NoError(t, err)
+	require.Equal(t, legacyData2, decoded2)
+}
+
+func TestDecodeUnencryptedDataWithEncryptedData(t *testing.T) {
+	// For backward compatibility, DecodeUnencryptedData treats any format as legacy unencrypted data
+	// and returns the data as-is. It does not return an error even for encrypted-looking data.
+	// The caller is responsible for ensuring data is not actually encrypted.
+	encryptedData := []byte{0x01, 'a', 'b', 'c', 'd', 'a', 't', 'a'}
+	decoded, err := DecodeUnencryptedData(encryptedData)
+	require.NoError(t, err)
+	require.Equal(t, encryptedData, decoded)
+}
+
+func TestExtractDataKeyID(t *testing.T) {
+	data := []byte("payload")
+	keyID := "xyz"
+	version := byte(0x01)
+
+	encoded, err := EncodeEncryptedData(data, version, keyID)
+	require.NoError(t, err)
+
+	extractedKeyID, err := ExtractDataKeyID(encoded)
+	require.NoError(t, err)
+	require.Equal(t, keyID, extractedKeyID)
+}
+
+func TestExtractDataKeyIDFromUnencryptedData(t *testing.T) {
+	// Trying to extract key ID from unencrypted data should return error
+	unencryptedData := EncodeUnencryptedData([]byte("plain"))
+	_, err := ExtractDataKeyID(unencryptedData)
+	require.Error(t, err)
+
+	// Trying to extract key ID from legacy data that is too short should return error
+	legacyData := []byte("abc") // 3 bytes < 4
+	_, err = ExtractDataKeyID(legacyData)
+	require.Error(t, err)
+
+	// Legacy data with non-zero bytes in positions 1-3 should also return error
+	// (This can't be confused with new-format encrypted data which has non-zero key ID)
+	legacyData2 := []byte{0x00, 0x01, 0x02, 0x03, 0x04, 0x05}
+	_, err = ExtractDataKeyID(legacyData2)
+	require.Error(t, err)
+}
diff --git a/pkg/encryption/types.go b/pkg/encryption/types.go
new file mode 100644
index 0000000000..d3b88640f3
--- /dev/null
+++ b/pkg/encryption/types.go
@@ -0,0 +1,44 @@
+// 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
+
+// EncryptionMeta is aligned with kvproto `keyspace_encryptionpb.EncryptionMeta`.
+type EncryptionMeta struct {
+	KeyspaceId uint32              `json:"keyspace_id,omitempty"`
+	Current    *EncryptionEpoch    `json:"current,omitempty"`
+	MasterKey  *MasterKey          `json:"master_key,omitempty"`
+	DataKeys   map[uint32]*DataKey `json:"data_keys,omitempty"`
+	History    []*EncryptionEpoch  `json:"history,omitempty"`
+}
+
+// EncryptionEpoch is aligned with kvproto `keyspace_encryptionpb.EncryptionEpoch`.
+type EncryptionEpoch struct {
+	FileId    uint64 `json:"file_id,omitempty"`
+	DataKeyId uint32 `json:"data_key_id,omitempty"`
+	CreatedAt uint64 `json:"created_at,omitempty"`
+}
+
+// MasterKey is aligned with kvproto `keyspace_encryptionpb.MasterKey`.
+type MasterKey struct {
+	Vendor     string `json:"vendor,omitempty"`
+	CmekId     string `json:"cmek_id,omitempty"`
+	Region     string `json:"region,omitempty"`
+	Endpoint   string `json:"endpoint,omitempty"`
+	Ciphertext []byte `json:"ciphertext,omitempty"`
+}
+
+// DataKey is aligned with kvproto `keyspace_encryptionpb.DataKey`.
+type DataKey struct {
+	Ciphertext []byte `json:"ciphertext,omitempty"`
+}
diff --git a/pkg/errors/error.go b/pkg/errors/error.go
index 3ad854eee1..3d0ce1c0cb 100644
--- a/pkg/errors/error.go
+++ b/pkg/errors/error.go
@@ -821,6 +821,37 @@ var (
 		"unimplemented IOType: %d",
 		errors.RFCCodeText("CDC:ErrUnimplementedIOType"),
 	)
+
+	// encryption related errors
+	ErrEncryptionMetaNotFound = errors.Normalize(
+		"encryption meta not found",
+		errors.RFCCodeText("CDC:ErrEncryptionMetaNotFound"),
+	)
+
+	ErrUnsupportedEncryptionAlgorithm = errors.Normalize(
+		"unsupported encryption algorithm: %s",
+		errors.RFCCodeText("CDC:ErrUnsupportedEncryptionAlgorithm"),
+	)
+
+	ErrEncryptionFailed = errors.Normalize(
+		"encryption failed: %s",
+		errors.RFCCodeText("CDC:ErrEncryptionFailed"),
+	)
+
+	ErrDecryptionFailed = errors.Normalize(
+		"decryption failed: %s",
+		errors.RFCCodeText("CDC:ErrDecryptionFailed"),
+	)
+
+	ErrInvalidDataKeyID = errors.Normalize(
+		"invalid data key ID: %s",
+		errors.RFCCodeText("CDC:ErrInvalidDataKeyID"),
+	)
+
+	ErrDataKeyNotFound = errors.Normalize(
+		"data key not found: %s",
+		errors.RFCCodeText("CDC:ErrDataKeyNotFound"),
+	)
 )
 
 // ErrorType defines the type of application errors