From 2cc7c8ae455f7a64e726637096e78916880c19b8 Mon Sep 17 00:00:00 2001 From: tenfyzhong Date: Mon, 23 Mar 2026 10:56:58 +0800 Subject: [PATCH 1/6] feat(encryption): add tikv encryption metadata http client Signed-off-by: tenfyzhong --- pkg/encryption/json_types.go | 62 ++++ pkg/encryption/tikv_http_client.go | 422 ++++++++++++++++++++++++ pkg/encryption/tikv_http_client_test.go | 312 ++++++++++++++++++ 3 files changed, 796 insertions(+) create mode 100644 pkg/encryption/json_types.go create mode 100644 pkg/encryption/tikv_http_client.go create mode 100644 pkg/encryption/tikv_http_client_test.go diff --git a/pkg/encryption/json_types.go b/pkg/encryption/json_types.go new file mode 100644 index 0000000000..b468846921 --- /dev/null +++ b/pkg/encryption/json_types.go @@ -0,0 +1,62 @@ +// 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 ( + "encoding/base64" + "encoding/json" + "fmt" +) + +// ByteArray supports decoding either: +// - a JSON string (base64-encoded bytes), or +// - a JSON array of uint8 values (TiKV status API style). +type ByteArray []byte + +func (b *ByteArray) UnmarshalJSON(data []byte) error { + if len(data) == 0 || string(data) == "null" { + *b = nil + return nil + } + + switch data[0] { + case '"': + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + decoded, err := base64.StdEncoding.DecodeString(s) + if err != nil { + return err + } + *b = decoded + return nil + case '[': + var ints []int + if err := json.Unmarshal(data, &ints); err != nil { + return err + } + out := make([]byte, len(ints)) + for i, v := range ints { + if v < 0 || v > 255 { + return fmt.Errorf("byte value out of range: %d", v) + } + out[i] = byte(v) + } + *b = out + return nil + default: + return fmt.Errorf("unsupported JSON type for bytes: %s", string(data)) + } +} diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go new file mode 100644 index 0000000000..96421dbc35 --- /dev/null +++ b/pkg/encryption/tikv_http_client.go @@ -0,0 +1,422 @@ +// 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" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + oldproto "github.com/gogo/protobuf/proto" + "github.com/pingcap/errors" + "github.com/pingcap/log" + cerrors "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/httputil" + "github.com/pingcap/ticdc/pkg/security" + "github.com/pingcap/tidb/pkg/util/engine" + pd "github.com/tikv/pd/client" + "github.com/tikv/pd/client/opt" + "go.uber.org/zap" +) + +type tikvEncryptionHTTPClient struct { + pdClient pd.Client + httpClient *httputil.Client + httpScheme string + httpTimeout time.Duration +} + +func NewTiKVEncryptionHTTPClient(pdClient pd.Client, credential *security.Credential) (TiKVEncryptionClient, error) { + httpClient, err := httputil.NewClient(credential) + if err != nil { + return nil, err + } + + httpScheme := "http" + if credential != nil && credential.IsTLSEnabled() { + httpScheme = "https" + } + + return &tikvEncryptionHTTPClient{ + pdClient: pdClient, + httpClient: httpClient, + httpScheme: httpScheme, + httpTimeout: 5 * time.Second, + }, nil +} + +func (c *tikvEncryptionHTTPClient) GetKeyspaceEncryptionMeta(ctx context.Context, keyspaceID uint32) (*EncryptionMeta, error) { + stores, err := c.pdClient.GetAllStores(ctx, opt.WithExcludeTombstone()) + if err != nil { + log.Warn("failed to list TiKV stores", + zap.Uint32("keyspaceID", keyspaceID), + zap.Error(err)) + return nil, errors.Trace(err) + } + + var lastErr error + for _, store := range stores { + if engine.IsTiFlash(store) { + continue + } + + statusAddr := store.GetStatusAddress() + if statusAddr == "" { + continue + } + + meta, err := c.getEncryptionMetaFromStore(ctx, store.GetId(), statusAddr, keyspaceID) + if err == nil { + return meta, nil + } + if cerrors.ErrEncryptionMetaNotFound.Equal(err) { + lastErr = err + continue + } + lastErr = err + } + + if lastErr == nil { + lastErr = cerrors.ErrEncryptionMetaNotFound + } + return nil, lastErr +} + +func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Context, storeID uint64, statusAddr string, keyspaceID uint32) (*EncryptionMeta, error) { + storeURL := c.buildStatusURL(statusAddr, keyspaceID) + contentType := "" + + reqCtx, cancel := context.WithTimeout(ctx, c.httpTimeout) + defer cancel() + + resp, err := c.httpClient.Get(reqCtx, storeURL) + if err != nil { + log.Warn("failed to fetch encryption meta from TiKV store", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL), + zap.Error(err)) + return nil, errors.Trace(err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + log.Warn("failed to read encryption meta response body", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL), + zap.Error(err)) + return nil, errors.Trace(err) + } + contentType = resp.Header.Get("Content-Type") + + if resp.StatusCode == http.StatusNotFound { + log.Debug("encryption meta not found on TiKV store", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL)) + return nil, cerrors.ErrEncryptionMetaNotFound + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + log.Warn("unexpected encryption meta response status", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL), + zap.Int("statusCode", resp.StatusCode), + zap.Int("bodySize", len(body)), + zap.String("body", truncateBytesForLog(body, 256))) + return nil, errors.Errorf("[%d] %s", resp.StatusCode, body) + } + + metaResp := &encryptionMetaResponse{} + responseFormat := "json" + if jsonErr := json.Unmarshal(body, metaResp); jsonErr != nil { + log.Debug("failed to decode encryption meta response as json, fallback to protobuf", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL), + zap.String("contentType", contentType), + zap.Int("bodySize", len(body)), + zap.Error(jsonErr)) + + metaResp, err = decodeEncryptionMetaResponseFromProtobuf(body) + if err != nil { + decodeErr := errors.Annotatef(err, "json decode failed: %v", jsonErr) + log.Warn("failed to decode encryption meta response", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL), + zap.String("contentType", contentType), + zap.Int("bodySize", len(body)), + zap.String("body", truncateBytesForLog(body, 256)), + zap.Error(decodeErr)) + return nil, errors.Trace(decodeErr) + } + responseFormat = "protobuf" + } + + meta, err := metaResp.toEncryptionMeta() + if err != nil { + log.Warn("failed to convert encryption meta response", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL), + zap.Error(err)) + return nil, err + } + + if meta.KeyspaceId != 0 && meta.KeyspaceId != keyspaceID { + log.Warn("encryption meta keyspace ID mismatch", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("requestedKeyspaceID", keyspaceID), + zap.Uint32("metaKeyspaceID", meta.KeyspaceId), + zap.String("url", storeURL)) + } + + masterKeyCiphertextLen := 0 + if meta.MasterKey != nil { + masterKeyCiphertextLen = len(meta.MasterKey.Ciphertext) + } + + log.Info("fetched valid encryption meta from TiKV store", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.String("responseFormat", responseFormat), + zap.String("contentType", contentType), + zap.Uint32("requestedKeyspaceID", 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)), + zap.Int("masterKeyCiphertextLen", masterKeyCiphertextLen)) + + return meta, nil +} + +func (c *tikvEncryptionHTTPClient) buildStatusURL(statusAddr string, keyspaceID uint32) string { + if strings.Contains(statusAddr, "://") { + return fmt.Sprintf("%s/encryption/get-meta?keyspace_id=%d", strings.TrimRight(statusAddr, "/"), keyspaceID) + } + return fmt.Sprintf("%s://%s/encryption/get-meta?keyspace_id=%d", c.httpScheme, statusAddr, keyspaceID) +} + +type encryptionMetaResponse struct { + KeyspaceId uint32 `json:"keyspace_id"` + Current encryptionEpochResponse `json:"current"` + MasterKey masterKeyResponse `json:"master_key"` + DataKeys map[uint32]dataKeyResponse `json:"data_keys"` + History []encryptionEpochResponse `json:"history"` +} + +type encryptionEpochResponse struct { + FileId uint64 `json:"file_id"` + DataKeyId uint32 `json:"data_key_id"` + CreatedAt uint64 `json:"created_at"` +} + +type masterKeyResponse struct { + Vendor string `json:"vendor"` + CmekId string `json:"cmek_id"` + Region string `json:"region"` + Endpoint string `json:"endpoint"` + Ciphertext ByteArray `json:"ciphertext"` +} + +type dataKeyResponse struct { + Ciphertext ByteArray `json:"ciphertext"` +} + +type keyspaceEncryptionMetaPB struct { + KeyspaceId uint32 `protobuf:"varint,1,opt,name=keyspace_id,json=keyspaceId,proto3"` + Current *keyspaceEncryptionEpochPB `protobuf:"bytes,2,opt,name=current,proto3"` + MasterKey *keyspaceMasterKeyPB `protobuf:"bytes,3,opt,name=master_key,json=masterKey,proto3"` + DataKeys map[uint32]*keyspaceDataKeyPB `protobuf:"bytes,4,rep,name=data_keys,json=dataKeys,proto3" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + History []*keyspaceEncryptionEpochPB `protobuf:"bytes,5,rep,name=history,proto3"` +} + +func (m *keyspaceEncryptionMetaPB) Reset() { *m = keyspaceEncryptionMetaPB{} } +func (m *keyspaceEncryptionMetaPB) String() string { return "" } +func (*keyspaceEncryptionMetaPB) ProtoMessage() {} + +type keyspaceEncryptionEpochPB struct { + FileId uint64 `protobuf:"varint,1,opt,name=file_id,json=fileId,proto3"` + DataKeyId uint32 `protobuf:"varint,2,opt,name=data_key_id,json=dataKeyId,proto3"` + CreatedAt uint64 `protobuf:"varint,3,opt,name=created_at,json=createdAt,proto3"` +} + +func (m *keyspaceEncryptionEpochPB) Reset() { *m = keyspaceEncryptionEpochPB{} } +func (m *keyspaceEncryptionEpochPB) String() string { return "" } +func (*keyspaceEncryptionEpochPB) ProtoMessage() {} + +type keyspaceMasterKeyPB struct { + Vendor string `protobuf:"bytes,1,opt,name=vendor,proto3"` + CmekId string `protobuf:"bytes,2,opt,name=cmek_id,json=cmekId,proto3"` + Region string `protobuf:"bytes,3,opt,name=region,proto3"` + Endpoint string `protobuf:"bytes,4,opt,name=endpoint,proto3"` + Ciphertext []byte `protobuf:"bytes,5,opt,name=ciphertext,proto3"` +} + +func (m *keyspaceMasterKeyPB) Reset() { *m = keyspaceMasterKeyPB{} } +func (m *keyspaceMasterKeyPB) String() string { return "" } +func (*keyspaceMasterKeyPB) ProtoMessage() {} + +type keyspaceDataKeyPB struct { + Ciphertext []byte `protobuf:"bytes,1,opt,name=ciphertext,proto3"` +} + +func (m *keyspaceDataKeyPB) Reset() { *m = keyspaceDataKeyPB{} } +func (m *keyspaceDataKeyPB) String() string { return "" } +func (*keyspaceDataKeyPB) ProtoMessage() {} + +func decodeEncryptionMetaResponseFromProtobuf(body []byte) (*encryptionMetaResponse, error) { + metaPB := &keyspaceEncryptionMetaPB{} + if err := oldproto.Unmarshal(body, metaPB); err != nil { + 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 metaPB.toEncryptionMetaResponse(), nil +} + +func (m *keyspaceEncryptionMetaPB) toEncryptionMetaResponse() *encryptionMetaResponse { + resp := &encryptionMetaResponse{ + KeyspaceId: m.KeyspaceId, + DataKeys: make(map[uint32]dataKeyResponse, len(m.DataKeys)), + History: make([]encryptionEpochResponse, 0, len(m.History)), + } + + if m.Current != nil { + resp.Current = encryptionEpochResponse{ + FileId: m.Current.FileId, + DataKeyId: m.Current.DataKeyId, + CreatedAt: m.Current.CreatedAt, + } + } + + if m.MasterKey != nil { + resp.MasterKey = masterKeyResponse{ + Vendor: m.MasterKey.Vendor, + CmekId: m.MasterKey.CmekId, + Region: m.MasterKey.Region, + Endpoint: m.MasterKey.Endpoint, + Ciphertext: ByteArray(m.MasterKey.Ciphertext), + } + } + + for id, dataKey := range m.DataKeys { + if dataKey == nil { + continue + } + resp.DataKeys[id] = dataKeyResponse{ + Ciphertext: ByteArray(dataKey.Ciphertext), + } + } + + for _, epoch := range m.History { + if epoch == nil { + continue + } + resp.History = append(resp.History, encryptionEpochResponse{ + FileId: epoch.FileId, + DataKeyId: epoch.DataKeyId, + CreatedAt: epoch.CreatedAt, + }) + } + + return resp +} + +func (r *encryptionMetaResponse) toEncryptionMeta() (*EncryptionMeta, error) { + if r.Current.DataKeyId == 0 { + log.Warn("invalid encryption meta from TiKV: current data key ID is empty", + zap.Uint32("metaKeyspaceID", r.KeyspaceId)) + return nil, cerrors.ErrEncryptionMetaNotFound + } + + version := byte(r.Current.DataKeyId & 0xFF) + if version == VersionUnencrypted { + log.Warn("invalid encryption meta from TiKV: version must be non-zero", + zap.Uint32("metaKeyspaceID", r.KeyspaceId), + zap.Uint32("currentDataKeyID", r.Current.DataKeyId)) + return nil, cerrors.ErrEncryptionFailed.GenWithStackByArgs("version must be non-zero") + } + + dataKeys := make(map[uint32]*DataKey, len(r.DataKeys)) + for id, dk := range r.DataKeys { + dataKeys[id] = &DataKey{Ciphertext: []byte(dk.Ciphertext)} + } + + if _, ok := dataKeys[r.Current.DataKeyId]; !ok { + log.Warn("invalid encryption meta from TiKV: current data key missing", + zap.Uint32("metaKeyspaceID", r.KeyspaceId), + zap.Uint32("currentDataKeyID", r.Current.DataKeyId), + zap.Int("dataKeyCount", len(dataKeys))) + return nil, cerrors.ErrDataKeyNotFound.GenWithStackByArgs("current data key not found") + } + + history := make([]*EncryptionEpoch, 0, len(r.History)) + for _, epoch := range r.History { + history = append(history, &EncryptionEpoch{ + FileId: epoch.FileId, + DataKeyId: epoch.DataKeyId, + CreatedAt: epoch.CreatedAt, + }) + } + + return &EncryptionMeta{ + KeyspaceId: r.KeyspaceId, + Current: &EncryptionEpoch{ + FileId: r.Current.FileId, + DataKeyId: r.Current.DataKeyId, + CreatedAt: r.Current.CreatedAt, + }, + MasterKey: r.MasterKey.toMasterKey(), + DataKeys: dataKeys, + History: history, + }, nil +} + +func (r *masterKeyResponse) toMasterKey() *MasterKey { + return &MasterKey{ + Vendor: r.Vendor, + CmekId: r.CmekId, + Region: r.Region, + Endpoint: r.Endpoint, + Ciphertext: []byte(r.Ciphertext), + } +} + +func truncateBytesForLog(b []byte, max int) string { + if len(b) <= max { + return string(b) + } + return fmt.Sprintf("%s...(truncated, %d bytes total)", string(b[:max]), len(b)) +} diff --git a/pkg/encryption/tikv_http_client_test.go b/pkg/encryption/tikv_http_client_test.go new file mode 100644 index 0000000000..bbd79a46e0 --- /dev/null +++ b/pkg/encryption/tikv_http_client_test.go @@ -0,0 +1,312 @@ +// 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" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + oldproto "github.com/gogo/protobuf/proto" + "github.com/pingcap/kvproto/pkg/metapb" + cerrors "github.com/pingcap/ticdc/pkg/errors" + "github.com/stretchr/testify/require" + pd "github.com/tikv/pd/client" + pdopt "github.com/tikv/pd/client/opt" +) + +type mockTiKVMetaPDClient struct { + pd.Client + stores []*metapb.Store +} + +func (m *mockTiKVMetaPDClient) GetAllStores(ctx context.Context, opts ...pdopt.GetStoreOption) ([]*metapb.Store, error) { + return m.stores, nil +} + +func TestTiKVEncryptionHTTPClientGetKeyspaceEncryptionMeta(t *testing.T) { + t.Parallel() + + const keyspaceID = uint32(1) + const dataKeyID = uint32(0x010203) // 24-bit big-endian -> [0x01 0x02 0x03] + + handler := http.NewServeMux() + handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + qs := r.URL.Query() + if qs.Get("keyspace_id") != "1" { + w.WriteHeader(http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "keyspace_id": 1, + "current": {"file_id": 1, "data_key_id": 66051, "created_at": 0}, + "master_key": {"vendor": "aws-kms", "cmek_id": "cmek-1", "region": "us-west-1", "endpoint": "", "ciphertext": [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31]}, + "data_keys": { + "66051": {"ciphertext": [31,30,29,28,27,26,25,24,23,22,21,20,19,18,17,16,15,14,13,12,11,10,9,8,7,6,5,4,3,2,1,0]} + }, + "history": [] +}`)) + }) + + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + + srvURL, err := url.Parse(srv.URL) + require.NoError(t, err) + statusAddr := srvURL.Host + + pdCli := &mockTiKVMetaPDClient{ + stores: []*metapb.Store{ + {Id: 1, StatusAddress: statusAddr, Address: "unused"}, + }, + } + + client, err := NewTiKVEncryptionHTTPClient(pdCli, nil) + require.NoError(t, err) + + meta, err := client.GetKeyspaceEncryptionMeta(context.Background(), keyspaceID) + require.NoError(t, err) + require.NotNil(t, meta) + require.Equal(t, keyspaceID, meta.KeyspaceId) + require.NotNil(t, meta.Current) + + expectedKeyID := string([]byte{0x01, 0x02, 0x03}) + currentKeyID, err := encodeDataKeyID24BE(meta.Current.DataKeyId) + require.NoError(t, err) + require.Equal(t, expectedKeyID, currentKeyID) + require.Equal(t, dataKeyID, meta.Current.DataKeyId) + + dk, ok := meta.DataKeys[dataKeyID] + require.True(t, ok) + require.Len(t, dk.Ciphertext, 32) +} + +func TestTiKVEncryptionHTTPClientGetKeyspaceEncryptionMetaFromProtobuf(t *testing.T) { + t.Parallel() + + const keyspaceID = uint32(2) + const dataKeyID = uint32(0x010203) + + metaPB := &testKeyspaceEncryptionMetaPB{ + KeyspaceId: keyspaceID, + Current: &testEncryptionEpochPB{ + FileId: 1, + DataKeyId: dataKeyID, + CreatedAt: 123, + }, + MasterKey: &testMasterKeyPB{ + Vendor: "aws", + CmekId: "cmek-2", + Region: "eu-west-2", + Endpoint: "http://0.0.0.0:8080", + Ciphertext: []byte{1, 2, 3, 4}, + }, + DataKeys: map[uint32]*testDataKeyPB{ + dataKeyID: {Ciphertext: []byte{9, 8, 7, 6}}, + }, + History: []*testEncryptionEpochPB{ + { + FileId: 2, + DataKeyId: dataKeyID, + CreatedAt: 124, + }, + }, + } + + payload, err := oldproto.Marshal(metaPB) + require.NoError(t, err) + + handler := http.NewServeMux() + handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/protobuf") + _, _ = w.Write(payload) + }) + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + + srvURL, err := url.Parse(srv.URL) + require.NoError(t, err) + + pdCli := &mockTiKVMetaPDClient{ + stores: []*metapb.Store{ + {Id: 1, StatusAddress: srvURL.Host}, + }, + } + + client, err := NewTiKVEncryptionHTTPClient(pdCli, nil) + require.NoError(t, err) + + meta, err := client.GetKeyspaceEncryptionMeta(context.Background(), keyspaceID) + require.NoError(t, err) + require.Equal(t, keyspaceID, meta.KeyspaceId) + require.Equal(t, dataKeyID, meta.Current.DataKeyId) + require.Equal(t, "aws", meta.MasterKey.Vendor) + require.Equal(t, []byte{9, 8, 7, 6}, meta.DataKeys[dataKeyID].Ciphertext) +} + +func TestTiKVEncryptionHTTPClientNotFoundReturnsErrEncryptionMetaNotFound(t *testing.T) { + t.Parallel() + + handler := http.NewServeMux() + handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte("not found")) + }) + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + + srvURL, err := url.Parse(srv.URL) + require.NoError(t, err) + statusAddr := srvURL.Host + + pdCli := &mockTiKVMetaPDClient{ + stores: []*metapb.Store{ + {Id: 1, StatusAddress: statusAddr}, + }, + } + + client, err := NewTiKVEncryptionHTTPClient(pdCli, nil) + require.NoError(t, err) + + _, err = client.GetKeyspaceEncryptionMeta(context.Background(), 1) + require.True(t, cerrors.ErrEncryptionMetaNotFound.Equal(err), "err=%v", err) +} + +func TestTiKVEncryptionHTTPClientRejectsVersionZeroMeta(t *testing.T) { + t.Parallel() + + handler := http.NewServeMux() + handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "keyspace_id": 1, + "current": {"file_id": 1, "data_key_id": 66048, "created_at": 0}, + "master_key": {"vendor": "aws-kms", "cmek_id": "cmek-1", "region": "us-west-1", "endpoint": "", "ciphertext": [0,1,2]}, + "data_keys": {"66048": {"ciphertext": [1,2,3]}}, + "history": [] +}`)) + }) + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + + srvURL, err := url.Parse(srv.URL) + require.NoError(t, err) + pdCli := &mockTiKVMetaPDClient{ + stores: []*metapb.Store{ + {Id: 1, StatusAddress: srvURL.Host}, + }, + } + + client, err := NewTiKVEncryptionHTTPClient(pdCli, nil) + require.NoError(t, err) + + _, err = client.GetKeyspaceEncryptionMeta(context.Background(), 1) + require.True(t, cerrors.ErrEncryptionFailed.Equal(err), "err=%v", err) +} + +func TestTiKVEncryptionHTTPClientRejectsMetaMissingCurrentDataKey(t *testing.T) { + t.Parallel() + + handler := http.NewServeMux() + handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "keyspace_id": 1, + "current": {"file_id": 1, "data_key_id": 66051, "created_at": 0}, + "master_key": {"vendor": "aws-kms", "cmek_id": "cmek-1", "region": "us-west-1", "endpoint": "", "ciphertext": [0,1,2]}, + "data_keys": {"66052": {"ciphertext": [1,2,3]}}, + "history": [] +}`)) + }) + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + + srvURL, err := url.Parse(srv.URL) + require.NoError(t, err) + pdCli := &mockTiKVMetaPDClient{ + stores: []*metapb.Store{ + {Id: 1, StatusAddress: srvURL.Host}, + }, + } + + client, err := NewTiKVEncryptionHTTPClient(pdCli, nil) + require.NoError(t, err) + + _, err = client.GetKeyspaceEncryptionMeta(context.Background(), 1) + require.True(t, cerrors.ErrDataKeyNotFound.Equal(err), "err=%v", err) +} + +func TestByteArrayUnmarshalSupportsUint8Array(t *testing.T) { + t.Parallel() + + var b ByteArray + err := b.UnmarshalJSON([]byte(`[0, 1, 2, 255]`)) + require.NoError(t, err) + require.Equal(t, []byte{0, 1, 2, 255}, []byte(b)) + + var bad ByteArray + err = bad.UnmarshalJSON([]byte(`[256]`)) + require.Error(t, err) + require.True(t, strings.Contains(err.Error(), "out of range")) +} + +type testKeyspaceEncryptionMetaPB struct { + KeyspaceId uint32 `protobuf:"varint,1,opt,name=keyspace_id,json=keyspaceId,proto3"` + Current *testEncryptionEpochPB `protobuf:"bytes,2,opt,name=current,proto3"` + MasterKey *testMasterKeyPB `protobuf:"bytes,3,opt,name=master_key,json=masterKey,proto3"` + DataKeys map[uint32]*testDataKeyPB `protobuf:"bytes,4,rep,name=data_keys,json=dataKeys,proto3" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + History []*testEncryptionEpochPB `protobuf:"bytes,5,rep,name=history,proto3"` +} + +func (m *testKeyspaceEncryptionMetaPB) Reset() { *m = testKeyspaceEncryptionMetaPB{} } +func (m *testKeyspaceEncryptionMetaPB) String() string { return "" } +func (*testKeyspaceEncryptionMetaPB) ProtoMessage() {} + +type testEncryptionEpochPB struct { + FileId uint64 `protobuf:"varint,1,opt,name=file_id,json=fileId,proto3"` + DataKeyId uint32 `protobuf:"varint,2,opt,name=data_key_id,json=dataKeyId,proto3"` + CreatedAt uint64 `protobuf:"varint,3,opt,name=created_at,json=createdAt,proto3"` +} + +func (m *testEncryptionEpochPB) Reset() { *m = testEncryptionEpochPB{} } +func (m *testEncryptionEpochPB) String() string { return "" } +func (*testEncryptionEpochPB) ProtoMessage() {} + +type testMasterKeyPB struct { + Vendor string `protobuf:"bytes,1,opt,name=vendor,proto3"` + CmekId string `protobuf:"bytes,2,opt,name=cmek_id,json=cmekId,proto3"` + Region string `protobuf:"bytes,3,opt,name=region,proto3"` + Endpoint string `protobuf:"bytes,4,opt,name=endpoint,proto3"` + Ciphertext []byte `protobuf:"bytes,5,opt,name=ciphertext,proto3"` +} + +func (m *testMasterKeyPB) Reset() { *m = testMasterKeyPB{} } +func (m *testMasterKeyPB) String() string { return "" } +func (*testMasterKeyPB) ProtoMessage() {} + +type testDataKeyPB struct { + Ciphertext []byte `protobuf:"bytes,1,opt,name=ciphertext,proto3"` +} + +func (m *testDataKeyPB) Reset() { *m = testDataKeyPB{} } +func (m *testDataKeyPB) String() string { return "" } +func (*testDataKeyPB) ProtoMessage() {} From be909686316ae536628b43dea835e834a3d39d0a Mon Sep 17 00:00:00 2001 From: tenfyzhong Date: Mon, 23 Mar 2026 17:53:00 +0800 Subject: [PATCH 2/6] fix(encryption): make tikv meta client self-contained Signed-off-by: tenfyzhong --- pkg/encryption/tikv_http_client.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go index 96421dbc35..79ba3d5008 100644 --- a/pkg/encryption/tikv_http_client.go +++ b/pkg/encryption/tikv_http_client.go @@ -41,6 +41,12 @@ 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 { @@ -420,3 +426,17 @@ 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 +} From 2f1df327dfcb24cd66b8dae23024643c7002cf1f Mon Sep 17 00:00:00 2001 From: tenfyzhong Date: Mon, 23 Mar 2026 19:07:27 +0800 Subject: [PATCH 3/6] fix(encryption): remove JSON fallback and simplify TikV HTTP client - Remove JSON decoding fallback for encryption meta responses - Use protobuf directly as the only supported format - Simplify error handling by removing redundant JSON decode attempts - Clean up logging by removing response format field - Update protobuf import to use standard package alias Signed-off-by: tenfyzhong --- pkg/encryption/tikv_http_client.go | 34 ++++++++---------------------- 1 file changed, 9 insertions(+), 25 deletions(-) diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go index 79ba3d5008..5b54e86fd6 100644 --- a/pkg/encryption/tikv_http_client.go +++ b/pkg/encryption/tikv_http_client.go @@ -15,14 +15,13 @@ package encryption import ( "context" - "encoding/json" "fmt" "io" "net/http" "strings" "time" - oldproto "github.com/gogo/protobuf/proto" + "github.com/gogo/protobuf/proto" "github.com/pingcap/errors" "github.com/pingcap/log" cerrors "github.com/pingcap/ticdc/pkg/errors" @@ -154,33 +153,19 @@ func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Contex return nil, errors.Errorf("[%d] %s", resp.StatusCode, body) } - metaResp := &encryptionMetaResponse{} - responseFormat := "json" - if jsonErr := json.Unmarshal(body, metaResp); jsonErr != nil { - log.Debug("failed to decode encryption meta response as json, fallback to protobuf", + metaResp, err := decodeEncryptionMetaResponseFromProtobuf(body) + if err != nil { + decodeErr := errors.Annotatef(err, "json decode failed: %v", err) + log.Warn("failed to decode encryption meta response", zap.Uint64("storeID", storeID), zap.String("statusAddr", statusAddr), zap.Uint32("keyspaceID", keyspaceID), zap.String("url", storeURL), zap.String("contentType", contentType), zap.Int("bodySize", len(body)), - zap.Error(jsonErr)) - - metaResp, err = decodeEncryptionMetaResponseFromProtobuf(body) - if err != nil { - decodeErr := errors.Annotatef(err, "json decode failed: %v", jsonErr) - log.Warn("failed to decode encryption meta response", - zap.Uint64("storeID", storeID), - zap.String("statusAddr", statusAddr), - zap.Uint32("keyspaceID", keyspaceID), - zap.String("url", storeURL), - zap.String("contentType", contentType), - zap.Int("bodySize", len(body)), - zap.String("body", truncateBytesForLog(body, 256)), - zap.Error(decodeErr)) - return nil, errors.Trace(decodeErr) - } - responseFormat = "protobuf" + zap.String("body", truncateBytesForLog(body, 256)), + zap.Error(decodeErr)) + return nil, errors.Trace(decodeErr) } meta, err := metaResp.toEncryptionMeta() @@ -211,7 +196,6 @@ func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Contex log.Info("fetched valid encryption meta from TiKV store", zap.Uint64("storeID", storeID), zap.String("statusAddr", statusAddr), - zap.String("responseFormat", responseFormat), zap.String("contentType", contentType), zap.Uint32("requestedKeyspaceID", keyspaceID), zap.Uint32("metaKeyspaceID", meta.KeyspaceId), @@ -303,7 +287,7 @@ func (*keyspaceDataKeyPB) ProtoMessage() {} func decodeEncryptionMetaResponseFromProtobuf(body []byte) (*encryptionMetaResponse, error) { metaPB := &keyspaceEncryptionMetaPB{} - if err := oldproto.Unmarshal(body, metaPB); err != nil { + if err := proto.Unmarshal(body, metaPB); err != nil { return nil, errors.Trace(err) } if metaPB.Current == nil && metaPB.MasterKey == nil && len(metaPB.DataKeys) == 0 && len(metaPB.History) == 0 && metaPB.KeyspaceId == 0 { From 51279d71c2c83884c751bb52c2b1cc0d676caea1 Mon Sep 17 00:00:00 2001 From: tenfyzhong Date: Mon, 23 Mar 2026 19:32:20 +0800 Subject: [PATCH 4/6] fix(encryption): restore json fallback for tikv meta response Signed-off-by: tenfyzhong --- pkg/encryption/tikv_http_client.go | 34 ++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 9 deletions(-) diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go index 5b54e86fd6..79ba3d5008 100644 --- a/pkg/encryption/tikv_http_client.go +++ b/pkg/encryption/tikv_http_client.go @@ -15,13 +15,14 @@ package encryption import ( "context" + "encoding/json" "fmt" "io" "net/http" "strings" "time" - "github.com/gogo/protobuf/proto" + oldproto "github.com/gogo/protobuf/proto" "github.com/pingcap/errors" "github.com/pingcap/log" cerrors "github.com/pingcap/ticdc/pkg/errors" @@ -153,19 +154,33 @@ func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Contex return nil, errors.Errorf("[%d] %s", resp.StatusCode, body) } - metaResp, err := decodeEncryptionMetaResponseFromProtobuf(body) - if err != nil { - decodeErr := errors.Annotatef(err, "json decode failed: %v", err) - log.Warn("failed to decode encryption meta response", + metaResp := &encryptionMetaResponse{} + responseFormat := "json" + if jsonErr := json.Unmarshal(body, metaResp); jsonErr != nil { + log.Debug("failed to decode encryption meta response as json, fallback to protobuf", zap.Uint64("storeID", storeID), zap.String("statusAddr", statusAddr), zap.Uint32("keyspaceID", keyspaceID), zap.String("url", storeURL), zap.String("contentType", contentType), zap.Int("bodySize", len(body)), - zap.String("body", truncateBytesForLog(body, 256)), - zap.Error(decodeErr)) - return nil, errors.Trace(decodeErr) + zap.Error(jsonErr)) + + metaResp, err = decodeEncryptionMetaResponseFromProtobuf(body) + if err != nil { + decodeErr := errors.Annotatef(err, "json decode failed: %v", jsonErr) + log.Warn("failed to decode encryption meta response", + zap.Uint64("storeID", storeID), + zap.String("statusAddr", statusAddr), + zap.Uint32("keyspaceID", keyspaceID), + zap.String("url", storeURL), + zap.String("contentType", contentType), + zap.Int("bodySize", len(body)), + zap.String("body", truncateBytesForLog(body, 256)), + zap.Error(decodeErr)) + return nil, errors.Trace(decodeErr) + } + responseFormat = "protobuf" } meta, err := metaResp.toEncryptionMeta() @@ -196,6 +211,7 @@ func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Contex log.Info("fetched valid encryption meta from TiKV store", zap.Uint64("storeID", storeID), zap.String("statusAddr", statusAddr), + zap.String("responseFormat", responseFormat), zap.String("contentType", contentType), zap.Uint32("requestedKeyspaceID", keyspaceID), zap.Uint32("metaKeyspaceID", meta.KeyspaceId), @@ -287,7 +303,7 @@ func (*keyspaceDataKeyPB) ProtoMessage() {} func decodeEncryptionMetaResponseFromProtobuf(body []byte) (*encryptionMetaResponse, error) { metaPB := &keyspaceEncryptionMetaPB{} - if err := proto.Unmarshal(body, metaPB); err != nil { + if err := oldproto.Unmarshal(body, metaPB); err != nil { return nil, errors.Trace(err) } if metaPB.Current == nil && metaPB.MasterKey == nil && len(metaPB.DataKeys) == 0 && len(metaPB.History) == 0 && metaPB.KeyspaceId == 0 { From 5bc9a70295d940da27066307f7f6e80a024ab093 Mon Sep 17 00:00:00 2001 From: tenfyzhong Date: Tue, 24 Mar 2026 14:22:44 +0800 Subject: [PATCH 5/6] Revert "fix(encryption): restore json fallback for tikv meta response" This reverts commit fc6c88b9ea49e49c0482357125cac84fc80c4a34. --- pkg/encryption/tikv_http_client.go | 34 ++++++++---------------------- 1 file changed, 9 insertions(+), 25 deletions(-) diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go index 79ba3d5008..5b54e86fd6 100644 --- a/pkg/encryption/tikv_http_client.go +++ b/pkg/encryption/tikv_http_client.go @@ -15,14 +15,13 @@ package encryption import ( "context" - "encoding/json" "fmt" "io" "net/http" "strings" "time" - oldproto "github.com/gogo/protobuf/proto" + "github.com/gogo/protobuf/proto" "github.com/pingcap/errors" "github.com/pingcap/log" cerrors "github.com/pingcap/ticdc/pkg/errors" @@ -154,33 +153,19 @@ func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Contex return nil, errors.Errorf("[%d] %s", resp.StatusCode, body) } - metaResp := &encryptionMetaResponse{} - responseFormat := "json" - if jsonErr := json.Unmarshal(body, metaResp); jsonErr != nil { - log.Debug("failed to decode encryption meta response as json, fallback to protobuf", + metaResp, err := decodeEncryptionMetaResponseFromProtobuf(body) + if err != nil { + decodeErr := errors.Annotatef(err, "json decode failed: %v", err) + log.Warn("failed to decode encryption meta response", zap.Uint64("storeID", storeID), zap.String("statusAddr", statusAddr), zap.Uint32("keyspaceID", keyspaceID), zap.String("url", storeURL), zap.String("contentType", contentType), zap.Int("bodySize", len(body)), - zap.Error(jsonErr)) - - metaResp, err = decodeEncryptionMetaResponseFromProtobuf(body) - if err != nil { - decodeErr := errors.Annotatef(err, "json decode failed: %v", jsonErr) - log.Warn("failed to decode encryption meta response", - zap.Uint64("storeID", storeID), - zap.String("statusAddr", statusAddr), - zap.Uint32("keyspaceID", keyspaceID), - zap.String("url", storeURL), - zap.String("contentType", contentType), - zap.Int("bodySize", len(body)), - zap.String("body", truncateBytesForLog(body, 256)), - zap.Error(decodeErr)) - return nil, errors.Trace(decodeErr) - } - responseFormat = "protobuf" + zap.String("body", truncateBytesForLog(body, 256)), + zap.Error(decodeErr)) + return nil, errors.Trace(decodeErr) } meta, err := metaResp.toEncryptionMeta() @@ -211,7 +196,6 @@ func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Contex log.Info("fetched valid encryption meta from TiKV store", zap.Uint64("storeID", storeID), zap.String("statusAddr", statusAddr), - zap.String("responseFormat", responseFormat), zap.String("contentType", contentType), zap.Uint32("requestedKeyspaceID", keyspaceID), zap.Uint32("metaKeyspaceID", meta.KeyspaceId), @@ -303,7 +287,7 @@ func (*keyspaceDataKeyPB) ProtoMessage() {} func decodeEncryptionMetaResponseFromProtobuf(body []byte) (*encryptionMetaResponse, error) { metaPB := &keyspaceEncryptionMetaPB{} - if err := oldproto.Unmarshal(body, metaPB); err != nil { + if err := proto.Unmarshal(body, metaPB); err != nil { return nil, errors.Trace(err) } if metaPB.Current == nil && metaPB.MasterKey == nil && len(metaPB.DataKeys) == 0 && len(metaPB.History) == 0 && metaPB.KeyspaceId == 0 { From 8ec053c11dcac831434fe388fa340f7bbe08c2d8 Mon Sep 17 00:00:00 2001 From: tenfyzhong Date: Tue, 24 Mar 2026 16:20:00 +0800 Subject: [PATCH 6/6] encryption: remove json handling from tikv meta client Signed-off-by: tenfyzhong --- pkg/encryption/json_types.go | 62 ------------ pkg/encryption/tikv_http_client.go | 48 +++++----- pkg/encryption/tikv_http_client_test.go | 122 ++++++++++++++---------- 3 files changed, 98 insertions(+), 134 deletions(-) delete mode 100644 pkg/encryption/json_types.go diff --git a/pkg/encryption/json_types.go b/pkg/encryption/json_types.go deleted file mode 100644 index b468846921..0000000000 --- a/pkg/encryption/json_types.go +++ /dev/null @@ -1,62 +0,0 @@ -// 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 ( - "encoding/base64" - "encoding/json" - "fmt" -) - -// ByteArray supports decoding either: -// - a JSON string (base64-encoded bytes), or -// - a JSON array of uint8 values (TiKV status API style). -type ByteArray []byte - -func (b *ByteArray) UnmarshalJSON(data []byte) error { - if len(data) == 0 || string(data) == "null" { - *b = nil - return nil - } - - switch data[0] { - case '"': - var s string - if err := json.Unmarshal(data, &s); err != nil { - return err - } - decoded, err := base64.StdEncoding.DecodeString(s) - if err != nil { - return err - } - *b = decoded - return nil - case '[': - var ints []int - if err := json.Unmarshal(data, &ints); err != nil { - return err - } - out := make([]byte, len(ints)) - for i, v := range ints { - if v < 0 || v > 255 { - return fmt.Errorf("byte value out of range: %d", v) - } - out[i] = byte(v) - } - *b = out - return nil - default: - return fmt.Errorf("unsupported JSON type for bytes: %s", string(data)) - } -} diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go index 5b54e86fd6..9794b50b80 100644 --- a/pkg/encryption/tikv_http_client.go +++ b/pkg/encryption/tikv_http_client.go @@ -155,7 +155,7 @@ func (c *tikvEncryptionHTTPClient) getEncryptionMetaFromStore(ctx context.Contex metaResp, err := decodeEncryptionMetaResponseFromProtobuf(body) if err != nil { - decodeErr := errors.Annotatef(err, "json decode failed: %v", err) + decodeErr := errors.Annotatef(err, "protobuf decode failed: %v", err) log.Warn("failed to decode encryption meta response", zap.Uint64("storeID", storeID), zap.String("statusAddr", statusAddr), @@ -218,36 +218,36 @@ func (c *tikvEncryptionHTTPClient) buildStatusURL(statusAddr string, keyspaceID } type encryptionMetaResponse struct { - KeyspaceId uint32 `json:"keyspace_id"` - Current encryptionEpochResponse `json:"current"` - MasterKey masterKeyResponse `json:"master_key"` - DataKeys map[uint32]dataKeyResponse `json:"data_keys"` - History []encryptionEpochResponse `json:"history"` + KeyspaceId uint32 + Current encryptionEpochResponse + MasterKey masterKeyResponse + DataKeys map[uint32]dataKeyResponse + History []encryptionEpochResponse } type encryptionEpochResponse struct { - FileId uint64 `json:"file_id"` - DataKeyId uint32 `json:"data_key_id"` - CreatedAt uint64 `json:"created_at"` + FileId uint64 + DataKeyId uint32 + CreatedAt uint64 } type masterKeyResponse struct { - Vendor string `json:"vendor"` - CmekId string `json:"cmek_id"` - Region string `json:"region"` - Endpoint string `json:"endpoint"` - Ciphertext ByteArray `json:"ciphertext"` + Vendor string + CmekId string + Region string + Endpoint string + Ciphertext []byte } type dataKeyResponse struct { - Ciphertext ByteArray `json:"ciphertext"` + Ciphertext []byte } type keyspaceEncryptionMetaPB struct { - KeyspaceId uint32 `protobuf:"varint,1,opt,name=keyspace_id,json=keyspaceId,proto3"` + KeyspaceId uint32 `protobuf:"varint,1,opt,name=keyspace_id,proto3"` Current *keyspaceEncryptionEpochPB `protobuf:"bytes,2,opt,name=current,proto3"` - MasterKey *keyspaceMasterKeyPB `protobuf:"bytes,3,opt,name=master_key,json=masterKey,proto3"` - DataKeys map[uint32]*keyspaceDataKeyPB `protobuf:"bytes,4,rep,name=data_keys,json=dataKeys,proto3" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + MasterKey *keyspaceMasterKeyPB `protobuf:"bytes,3,opt,name=master_key,proto3"` + DataKeys map[uint32]*keyspaceDataKeyPB `protobuf:"bytes,4,rep,name=data_keys,proto3" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` History []*keyspaceEncryptionEpochPB `protobuf:"bytes,5,rep,name=history,proto3"` } @@ -256,9 +256,9 @@ func (m *keyspaceEncryptionMetaPB) String() string { return "" } func (*keyspaceEncryptionMetaPB) ProtoMessage() {} type keyspaceEncryptionEpochPB struct { - FileId uint64 `protobuf:"varint,1,opt,name=file_id,json=fileId,proto3"` - DataKeyId uint32 `protobuf:"varint,2,opt,name=data_key_id,json=dataKeyId,proto3"` - CreatedAt uint64 `protobuf:"varint,3,opt,name=created_at,json=createdAt,proto3"` + FileId uint64 `protobuf:"varint,1,opt,name=file_id,proto3"` + DataKeyId uint32 `protobuf:"varint,2,opt,name=data_key_id,proto3"` + CreatedAt uint64 `protobuf:"varint,3,opt,name=created_at,proto3"` } func (m *keyspaceEncryptionEpochPB) Reset() { *m = keyspaceEncryptionEpochPB{} } @@ -267,7 +267,7 @@ func (*keyspaceEncryptionEpochPB) ProtoMessage() {} type keyspaceMasterKeyPB struct { Vendor string `protobuf:"bytes,1,opt,name=vendor,proto3"` - CmekId string `protobuf:"bytes,2,opt,name=cmek_id,json=cmekId,proto3"` + CmekId string `protobuf:"bytes,2,opt,name=cmek_id,proto3"` Region string `protobuf:"bytes,3,opt,name=region,proto3"` Endpoint string `protobuf:"bytes,4,opt,name=endpoint,proto3"` Ciphertext []byte `protobuf:"bytes,5,opt,name=ciphertext,proto3"` @@ -317,7 +317,7 @@ func (m *keyspaceEncryptionMetaPB) toEncryptionMetaResponse() *encryptionMetaRes CmekId: m.MasterKey.CmekId, Region: m.MasterKey.Region, Endpoint: m.MasterKey.Endpoint, - Ciphertext: ByteArray(m.MasterKey.Ciphertext), + Ciphertext: m.MasterKey.Ciphertext, } } @@ -326,7 +326,7 @@ func (m *keyspaceEncryptionMetaPB) toEncryptionMetaResponse() *encryptionMetaRes continue } resp.DataKeys[id] = dataKeyResponse{ - Ciphertext: ByteArray(dataKey.Ciphertext), + Ciphertext: dataKey.Ciphertext, } } diff --git a/pkg/encryption/tikv_http_client_test.go b/pkg/encryption/tikv_http_client_test.go index bbd79a46e0..128a0dfab9 100644 --- a/pkg/encryption/tikv_http_client_test.go +++ b/pkg/encryption/tikv_http_client_test.go @@ -18,7 +18,6 @@ import ( "net/http" "net/http/httptest" "net/url" - "strings" "testing" oldproto "github.com/gogo/protobuf/proto" @@ -43,6 +42,27 @@ func TestTiKVEncryptionHTTPClientGetKeyspaceEncryptionMeta(t *testing.T) { const keyspaceID = uint32(1) const dataKeyID = uint32(0x010203) // 24-bit big-endian -> [0x01 0x02 0x03] + metaPB := &testKeyspaceEncryptionMetaPB{ + KeyspaceId: keyspaceID, + Current: &testEncryptionEpochPB{ + FileId: 1, + DataKeyId: dataKeyID, + CreatedAt: 0, + }, + MasterKey: &testMasterKeyPB{ + Vendor: "aws-kms", + CmekId: "cmek-1", + Region: "us-west-1", + Endpoint: "", + Ciphertext: []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31}, + }, + DataKeys: map[uint32]*testDataKeyPB{ + dataKeyID: {Ciphertext: []byte{31, 30, 29, 28, 27, 26, 25, 24, 23, 22, 21, 20, 19, 18, 17, 16, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0}}, + }, + History: []*testEncryptionEpochPB{}, + } + payload, err := oldproto.Marshal(metaPB) + require.NoError(t, err) handler := http.NewServeMux() handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { @@ -55,16 +75,8 @@ func TestTiKVEncryptionHTTPClientGetKeyspaceEncryptionMeta(t *testing.T) { w.WriteHeader(http.StatusBadRequest) return } - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{ - "keyspace_id": 1, - "current": {"file_id": 1, "data_key_id": 66051, "created_at": 0}, - "master_key": {"vendor": "aws-kms", "cmek_id": "cmek-1", "region": "us-west-1", "endpoint": "", "ciphertext": [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31]}, - "data_keys": { - "66051": {"ciphertext": [31,30,29,28,27,26,25,24,23,22,21,20,19,18,17,16,15,14,13,12,11,10,9,8,7,6,5,4,3,2,1,0]} - }, - "history": [] -}`)) + w.Header().Set("Content-Type", "application/protobuf") + _, _ = w.Write(payload) }) srv := httptest.NewServer(handler) @@ -193,17 +205,31 @@ func TestTiKVEncryptionHTTPClientNotFoundReturnsErrEncryptionMetaNotFound(t *tes func TestTiKVEncryptionHTTPClientRejectsVersionZeroMeta(t *testing.T) { t.Parallel() + metaPB := &testKeyspaceEncryptionMetaPB{ + KeyspaceId: 1, + Current: &testEncryptionEpochPB{ + FileId: 1, + DataKeyId: 66048, + CreatedAt: 0, + }, + MasterKey: &testMasterKeyPB{ + Vendor: "aws-kms", + CmekId: "cmek-1", + Region: "us-west-1", + Ciphertext: []byte{0, 1, 2}, + }, + DataKeys: map[uint32]*testDataKeyPB{ + 66048: {Ciphertext: []byte{1, 2, 3}}, + }, + History: []*testEncryptionEpochPB{}, + } + payload, err := oldproto.Marshal(metaPB) + require.NoError(t, err) handler := http.NewServeMux() handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{ - "keyspace_id": 1, - "current": {"file_id": 1, "data_key_id": 66048, "created_at": 0}, - "master_key": {"vendor": "aws-kms", "cmek_id": "cmek-1", "region": "us-west-1", "endpoint": "", "ciphertext": [0,1,2]}, - "data_keys": {"66048": {"ciphertext": [1,2,3]}}, - "history": [] -}`)) + w.Header().Set("Content-Type", "application/protobuf") + _, _ = w.Write(payload) }) srv := httptest.NewServer(handler) t.Cleanup(srv.Close) @@ -225,17 +251,31 @@ func TestTiKVEncryptionHTTPClientRejectsVersionZeroMeta(t *testing.T) { func TestTiKVEncryptionHTTPClientRejectsMetaMissingCurrentDataKey(t *testing.T) { t.Parallel() + metaPB := &testKeyspaceEncryptionMetaPB{ + KeyspaceId: 1, + Current: &testEncryptionEpochPB{ + FileId: 1, + DataKeyId: 66051, + CreatedAt: 0, + }, + MasterKey: &testMasterKeyPB{ + Vendor: "aws-kms", + CmekId: "cmek-1", + Region: "us-west-1", + Ciphertext: []byte{0, 1, 2}, + }, + DataKeys: map[uint32]*testDataKeyPB{ + 66052: {Ciphertext: []byte{1, 2, 3}}, + }, + History: []*testEncryptionEpochPB{}, + } + payload, err := oldproto.Marshal(metaPB) + require.NoError(t, err) handler := http.NewServeMux() handler.HandleFunc("/encryption/get-meta", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{ - "keyspace_id": 1, - "current": {"file_id": 1, "data_key_id": 66051, "created_at": 0}, - "master_key": {"vendor": "aws-kms", "cmek_id": "cmek-1", "region": "us-west-1", "endpoint": "", "ciphertext": [0,1,2]}, - "data_keys": {"66052": {"ciphertext": [1,2,3]}}, - "history": [] -}`)) + w.Header().Set("Content-Type", "application/protobuf") + _, _ = w.Write(payload) }) srv := httptest.NewServer(handler) t.Cleanup(srv.Close) @@ -255,25 +295,11 @@ func TestTiKVEncryptionHTTPClientRejectsMetaMissingCurrentDataKey(t *testing.T) require.True(t, cerrors.ErrDataKeyNotFound.Equal(err), "err=%v", err) } -func TestByteArrayUnmarshalSupportsUint8Array(t *testing.T) { - t.Parallel() - - var b ByteArray - err := b.UnmarshalJSON([]byte(`[0, 1, 2, 255]`)) - require.NoError(t, err) - require.Equal(t, []byte{0, 1, 2, 255}, []byte(b)) - - var bad ByteArray - err = bad.UnmarshalJSON([]byte(`[256]`)) - require.Error(t, err) - require.True(t, strings.Contains(err.Error(), "out of range")) -} - type testKeyspaceEncryptionMetaPB struct { - KeyspaceId uint32 `protobuf:"varint,1,opt,name=keyspace_id,json=keyspaceId,proto3"` + KeyspaceId uint32 `protobuf:"varint,1,opt,name=keyspace_id,proto3"` Current *testEncryptionEpochPB `protobuf:"bytes,2,opt,name=current,proto3"` - MasterKey *testMasterKeyPB `protobuf:"bytes,3,opt,name=master_key,json=masterKey,proto3"` - DataKeys map[uint32]*testDataKeyPB `protobuf:"bytes,4,rep,name=data_keys,json=dataKeys,proto3" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + MasterKey *testMasterKeyPB `protobuf:"bytes,3,opt,name=master_key,proto3"` + DataKeys map[uint32]*testDataKeyPB `protobuf:"bytes,4,rep,name=data_keys,proto3" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` History []*testEncryptionEpochPB `protobuf:"bytes,5,rep,name=history,proto3"` } @@ -282,9 +308,9 @@ func (m *testKeyspaceEncryptionMetaPB) String() string { return "" } func (*testKeyspaceEncryptionMetaPB) ProtoMessage() {} type testEncryptionEpochPB struct { - FileId uint64 `protobuf:"varint,1,opt,name=file_id,json=fileId,proto3"` - DataKeyId uint32 `protobuf:"varint,2,opt,name=data_key_id,json=dataKeyId,proto3"` - CreatedAt uint64 `protobuf:"varint,3,opt,name=created_at,json=createdAt,proto3"` + FileId uint64 `protobuf:"varint,1,opt,name=file_id,proto3"` + DataKeyId uint32 `protobuf:"varint,2,opt,name=data_key_id,proto3"` + CreatedAt uint64 `protobuf:"varint,3,opt,name=created_at,proto3"` } func (m *testEncryptionEpochPB) Reset() { *m = testEncryptionEpochPB{} } @@ -293,7 +319,7 @@ func (*testEncryptionEpochPB) ProtoMessage() {} type testMasterKeyPB struct { Vendor string `protobuf:"bytes,1,opt,name=vendor,proto3"` - CmekId string `protobuf:"bytes,2,opt,name=cmek_id,json=cmekId,proto3"` + CmekId string `protobuf:"bytes,2,opt,name=cmek_id,proto3"` Region string `protobuf:"bytes,3,opt,name=region,proto3"` Endpoint string `protobuf:"bytes,4,opt,name=endpoint,proto3"` Ciphertext []byte `protobuf:"bytes,5,opt,name=ciphertext,proto3"`