feat(bigtable): Customer Managed Encryption (CMEK)

bigtable/admin.go

@@ -39,6 +39,7 @@ import (
 	"google.golang.org/api/option"
 	gtransport "google.golang.org/api/transport/grpc"
 	btapb "google.golang.org/genproto/googleapis/bigtable/admin/v2"
+	"google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/genproto/protobuf/field_mask"
 	"google.golang.org/grpc/metadata"
 )
@@ -119,6 +120,76 @@ func (ac *AdminClient) backupPath(cluster, backup string) string {
 	return fmt.Sprintf("projects/%s/instances/%s/clusters/%s/backups/%s", ac.project, ac.instance, cluster, backup)
 }
 
+// EncryptionInfo represents the encryption info of a table.
+type EncryptionInfo struct {
+	// TODO: compare to https://github.com/googleapis/java-bigtable/pull/656/files#diff-10655a12f2a8430533dca76435f6829a1a0bc978ada1295529b59a46afbcd95fR28
+	EncryptionStatus *status.Status // TODO: should this be wrapped as EncryptionType has been?
+	EncryptionType   EncryptionType
+	KMSKeyVersion    string
+}
+
+func newEncryptionInfo(pbInfo *btapb.EncryptionInfo) EncryptionInfo {
+	info := EncryptionInfo{}
+	info.EncryptionStatus = pbInfo.EncryptionStatus
+	// TODO: could also just return this as a string, but wrapped
+	info.EncryptionType = EncryptionType(pbInfo.EncryptionType.Number())
+	info.KMSKeyVersion = pbInfo.KmsKeyVersion
+
+	return info
+}
+
+type EncryptionType int32
+
+const (
+	// Encryption type was not specified, though data at rest remains encrypted.
+	ENCRYPTION_TYPE_UNSPECIFIED EncryptionType = 0
+	// The data backing this resource is encrypted at rest with a key that is
+	// fully managed by Google. No key version or status will be populated.
+	// This is the default state.
+	GOOGLE_DEFAULT_ENCRYPTION EncryptionType = 1
+	// The data backing this resource is encrypted at rest with a key that is
+	// managed by the customer.
+	// The in-use version of the key and its status are populated for
+	// CMEK-protected tables.
+	// CMEK-protected backups are pinned to the key version that was in use at
+	// the time the backup was taken. This key version is populated but its
+	// status is not tracked and is reported as `UNKNOWN`.
+	CUSTOMER_MANAGED_ENCRYPTION EncryptionType = 2
+)
+
+// Gets the current encryption info for the table across all of the clusters.
+// The returned map will be keyed by cluster id and contain a status for all of the keys in use.
+func (ac *AdminClient) EncryptionInfo(ctx context.Context, table string) (map[string][]EncryptionInfo, error) {
+	ctx = mergeOutgoingMetadata(ctx, ac.md)
+
+	res, err := ac.getTable(ctx, table)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: backups also needs this.
+	// TODO: why not just return the clusterstates, as it has a map of encryption info.
+	encryptionInfo := map[string][]EncryptionInfo{}
+	for key, cs := range res.ClusterStates {
+		// TODO: if we didn't wrap EncryptionInfo, this could reduce to
+		// returning map[string][]*btapb.EncryptionInfo directly
+		// encryptionInfo[key] = cs.EncryptionInfo
+		encInfo := cs.GetEncryptionInfo()
+		if encInfo == nil {
+			return nil, nil
+		}
+		// TODO: can make a custom class?
+		for _, pbInfo := range cs.EncryptionInfo {
+			info := EncryptionInfo{}
+			info.EncryptionStatus = pbInfo.EncryptionStatus
+			info.EncryptionType = EncryptionType(pbInfo.EncryptionType.Number())
+			info.KMSKeyVersion = pbInfo.KmsKeyVersion
+			encryptionInfo[key] = append(encryptionInfo[key], info)
+		}
+	}
+
+	return encryptionInfo, nil
+}
+
 // Tables returns a list of the tables in the instance.
 func (ac *AdminClient) Tables(ctx context.Context) ([]string, error) {
 	ctx = mergeOutgoingMetadata(ctx, ac.md)
@@ -247,12 +318,13 @@ type FamilyInfo struct {
 	GCPolicy string
 }
 
-// TableInfo retrieves information about a table.
-func (ac *AdminClient) TableInfo(ctx context.Context, table string) (*TableInfo, error) {
+func (ac *AdminClient) getTable(ctx context.Context, table string) (*btapb.Table, error) {
 	ctx = mergeOutgoingMetadata(ctx, ac.md)
 	prefix := ac.instancePrefix()
 	req := &btapb.GetTableRequest{
 		Name: prefix + "/tables/" + table,
+		View: btapb.Table_FULL,
+		// View: btapb.Table_ENCRYPTION_VIEW,
 	}
 
 	var res *btapb.Table
@@ -265,6 +337,17 @@ func (ac *AdminClient) TableInfo(ctx context.Context, table string) (*TableInfo,
 	if err != nil {
 		return nil, err
 	}
+	return res, nil
+}
+
+// TableInfo retrieves information about a table.
+func (ac *AdminClient) TableInfo(ctx context.Context, table string) (*TableInfo, error) {
+	ctx = mergeOutgoingMetadata(ctx, ac.md)
+
+	res, err := ac.getTable(ctx, table)
+	if err != nil {
+		return nil, err
+	}
 
 	ti := &TableInfo{}
 	for name, fam := range res.ColumnFamilies {
@@ -960,13 +1043,26 @@ type ClusterConfig struct {
 	InstanceID, ClusterID, Zone string
 	NumNodes                    int32
 	StorageType                 StorageType
+
+	// Describes the Cloud KMS encryption key that will be used to protect the
+	// destination Bigtable cluster. The requirements for this key are:
+	//  1) The Cloud Bigtable service account associated with the project that
+	//  contains this cluster must be granted the
+	//  `cloudkms.cryptoKeyEncrypterDecrypter` role on the CMEK key.
+	//  2) Only regional keys can be used and the region of the CMEK key must
+	//  match the region of the cluster.
+	// 3) All clusters within an instance must use the same CMEK key.
+	KMSKeyName string
 }
 
 func (cc *ClusterConfig) proto(project string) *btapb.Cluster {
+	ec := btapb.Cluster_EncryptionConfig{}
+	ec.KmsKeyName = cc.KMSKeyName
 	return &btapb.Cluster{
 		ServeNodes:         cc.NumNodes,
 		DefaultStorageType: cc.StorageType.proto(),
 		Location:           "projects/" + project + "/locations/" + cc.Zone,
+		EncryptionConfig:   &ec,
 	}
 }
 
@@ -977,6 +1073,7 @@ type ClusterInfo struct {
 	ServeNodes  int         // number of allocated serve nodes
 	State       string      // state of the cluster
 	StorageType StorageType // the storage type of the cluster
+	KMSKeyName  string      // the customer managed encryption key for the cluster
 }
 
 // CreateCluster creates a new cluster in an instance.
@@ -1045,6 +1142,7 @@ func (iac *InstanceAdminClient) Clusters(ctx context.Context, instanceID string)
 			ServeNodes:  int(c.ServeNodes),
 			State:       c.State.String(),
 			StorageType: storageTypeFromProto(c.DefaultStorageType),
+			KMSKeyName:  c.EncryptionConfig.KmsKeyName,
 		})
 	}
 	if len(res.FailedLocations) > 0 {
@@ -1058,7 +1156,10 @@ func (iac *InstanceAdminClient) Clusters(ctx context.Context, instanceID string)
 // GetCluster fetches a cluster in an instance
 func (iac *InstanceAdminClient) GetCluster(ctx context.Context, instanceID, clusterID string) (*ClusterInfo, error) {
 	ctx = mergeOutgoingMetadata(ctx, iac.md)
-	req := &btapb.GetClusterRequest{Name: "projects/" + iac.project + "/instances/" + instanceID + "/clusters/" + clusterID}
+	req := &btapb.GetClusterRequest{
+		Name: "projects/" + iac.project + "/instances/" + instanceID + "/clusters/" + clusterID,
+		// View: btapb.Table_ENCRYPTION_VIEW
+	}
 	var c *btapb.Cluster
 	err := gax.Invoke(ctx, func(ctx context.Context, _ gax.CallSettings) error {
 		var err error
@@ -1077,6 +1178,7 @@ func (iac *InstanceAdminClient) GetCluster(ctx context.Context, instanceID, clus
 		ServeNodes:  int(c.ServeNodes),
 		State:       c.State.String(),
 		StorageType: storageTypeFromProto(c.DefaultStorageType),
+		KMSKeyName:  c.EncryptionConfig.KmsKeyName,
 	}
 	return cis, nil
 }
@@ -1570,15 +1672,21 @@ func newBackupInfo(backup *btapb.Backup) (*BackupInfo, error) {
 		return nil, fmt.Errorf("invalid expireTime: %v", err)
 	}
 
-	return &BackupInfo{
+	bi := BackupInfo{
 		Name:        name,
 		SourceTable: tableID,
 		SizeBytes:   backup.SizeBytes,
 		StartTime:   startTime,
 		EndTime:     endTime,
 		ExpireTime:  expireTime,
 		State:       backup.State.String(),
-	}, nil
+		// EncryptionInfo: // populate this after verifying 'nil-nes'
+	}
+	// Do not attempt to populate encryption info if not available.
+	if backup.EncryptionInfo != nil {
+		bi.EncryptionInfo = newEncryptionInfo(backup.EncryptionInfo)
+	}
+	return &bi, nil
 }
 
 // BackupIterator is an EntryIterator that iterates over log entries.
@@ -1607,13 +1715,14 @@ func (it *BackupIterator) Next() (*BackupInfo, error) {
 
 // BackupInfo contains backup metadata. This struct is read-only.
 type BackupInfo struct {
-	Name        string
-	SourceTable string
-	SizeBytes   int64
-	StartTime   time.Time
-	EndTime     time.Time
-	ExpireTime  time.Time
-	State       string
+	Name           string
+	SourceTable    string
+	SizeBytes      int64
+	StartTime      time.Time
+	EndTime        time.Time
+	ExpireTime     time.Time
+	State          string
+	EncryptionInfo EncryptionInfo
 }
 
 // BackupInfo gets backup metadata.
@@ -1623,6 +1732,7 @@ func (ac *AdminClient) BackupInfo(ctx context.Context, cluster, backup string) (
 
 	req := &btapb.GetBackupRequest{
 		Name: backupPath,
+		//View: btapb.Table_ENCRYPTION_VIEW,
 	}
 
 	var resp *btapb.Backup