diff --git a/pkg/encryption/tikv_http_client.go b/pkg/encryption/tikv_http_client.go new file mode 100644 index 0000000000..9794b50b80 --- /dev/null +++ b/pkg/encryption/tikv_http_client.go @@ -0,0 +1,426 @@ +// 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" + "fmt" + "io" + "net/http" + "strings" + "time" + + "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 +} + +// 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 { + 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, err := decodeEncryptionMetaResponseFromProtobuf(body) + if err != nil { + 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), + 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) + } + + 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("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 + Current encryptionEpochResponse + MasterKey masterKeyResponse + DataKeys map[uint32]dataKeyResponse + History []encryptionEpochResponse +} + +type encryptionEpochResponse struct { + FileId uint64 + DataKeyId uint32 + CreatedAt uint64 +} + +type masterKeyResponse struct { + Vendor string + CmekId string + Region string + Endpoint string + Ciphertext []byte +} + +type dataKeyResponse struct { + Ciphertext []byte +} + +type keyspaceEncryptionMetaPB struct { + 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,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"` +} + +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,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{} } +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,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 := 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 { + 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: m.MasterKey.Ciphertext, + } + } + + for id, dataKey := range m.DataKeys { + if dataKey == nil { + continue + } + resp.DataKeys[id] = dataKeyResponse{ + Ciphertext: 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)) +} + +func safeKMSVendor(masterKey *MasterKey) string { + if masterKey == nil { + return "" + } + return masterKey.Vendor +} + +func safeCMEKID(masterKey *MasterKey) string { + if masterKey == nil { + return "" + } + return masterKey.CmekId +} diff --git a/pkg/encryption/tikv_http_client_test.go b/pkg/encryption/tikv_http_client_test.go new file mode 100644 index 0000000000..128a0dfab9 --- /dev/null +++ b/pkg/encryption/tikv_http_client_test.go @@ -0,0 +1,338 @@ +// 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" + "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] + 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) { + 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/protobuf") + _, _ = w.Write(payload) + }) + + 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() + 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/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) + + _, err = client.GetKeyspaceEncryptionMeta(context.Background(), 1) + require.True(t, cerrors.ErrEncryptionFailed.Equal(err), "err=%v", err) +} + +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/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) + + _, err = client.GetKeyspaceEncryptionMeta(context.Background(), 1) + require.True(t, cerrors.ErrDataKeyNotFound.Equal(err), "err=%v", err) +} + +type testKeyspaceEncryptionMetaPB struct { + 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,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"` +} + +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,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{} } +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,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() {}