Secrets: Refactor data_key_id out of the encoded secure value payload (#111852)

* everything compiles

* tests pass

* remove file included by accident

* add entry to gitignore

* some scaffolding for the migration executor

* remove file

* implement and test the migration

* use xkube.Namespace in our interfaces

* add todo

* update wire deps

* add some logs

* fix wire dependency ordering

* create tests to validate error conditions during migrations

.gitignore

@@ -250,6 +250,8 @@ public/mockServiceWorker.js
 /e2e-playwright/test-plugins/*/dist
 /apps/provisioning/cmd/job-controller/bin/
 
-
 # Ignore unified storage kv store files
 /grafana-kv-data
+
+# Ignore debug output from test library
+/pkg/storage/secret/metadata/testdata/rapid/TestStateMachine/

pkg/registry/apis/secret/contracts/encryption.go

@@ -1,6 +1,10 @@
 package contracts
 
-import "context"
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
+)
 
 // EncryptionManager is an envelope encryption service in charge of encrypting/decrypting secrets.
 type EncryptionManager interface {
@@ -8,17 +12,23 @@ type EncryptionManager interface {
 	// For those specific use cases where the encryption operation cannot be moved outside
 	// the database transaction, look at database-specific methods present at the specific
 	// implementation present at manager.EncryptionService.
-	Encrypt(ctx context.Context, namespace string, payload []byte) ([]byte, error)
-	Decrypt(ctx context.Context, namespace string, payload []byte) ([]byte, error)
+	Encrypt(ctx context.Context, namespace xkube.Namespace, payload []byte) (EncryptedPayload, error)
+	Decrypt(ctx context.Context, namespace xkube.Namespace, payload EncryptedPayload) ([]byte, error)
+}
+
+type EncryptedPayload struct {
+	DataKeyID     string
+	EncryptedData []byte
 }
 
 type EncryptedValue struct {
-	Namespace     string
-	Name          string
-	Version       int64
-	EncryptedData []byte
-	Created       int64
-	Updated       int64
+	EncryptedPayload
+
+	Namespace string
+	Name      string
+	Version   int64
+	Created   int64
+	Updated   int64
 }
 
 // ListOpts defines pagination options for listing encrypted values.
@@ -28,10 +38,10 @@ type ListOpts struct {
 }
 
 type EncryptedValueStorage interface {
-	Create(ctx context.Context, namespace, name string, version int64, encryptedData []byte) (*EncryptedValue, error)
-	Update(ctx context.Context, namespace, name string, version int64, encryptedData []byte) error
-	Get(ctx context.Context, namespace, name string, version int64) (*EncryptedValue, error)
-	Delete(ctx context.Context, namespace, name string, version int64) error
+	Create(ctx context.Context, namespace xkube.Namespace, name string, version int64, encryptedData EncryptedPayload) (*EncryptedValue, error)
+	Update(ctx context.Context, namespace xkube.Namespace, name string, version int64, encryptedData EncryptedPayload) error
+	Get(ctx context.Context, namespace xkube.Namespace, name string, version int64) (*EncryptedValue, error)
+	Delete(ctx context.Context, namespace xkube.Namespace, name string, version int64) error
 }
 
 type GlobalEncryptedValueStorage interface {
@@ -39,6 +49,10 @@ type GlobalEncryptedValueStorage interface {
 	CountAll(ctx context.Context, untilTime *int64) (int64, error)
 }
 
+type EncryptedValueMigrationExecutor interface {
+	Execute(ctx context.Context) (int, error)
+}
+
 type ConsolidationService interface {
 	Consolidate(ctx context.Context) error
 }

pkg/registry/apis/secret/contracts/keeper.go

@@ -96,10 +96,10 @@ func (s ExternalID) String() string {
 
 // Keeper is the interface for secret keepers.
 type Keeper interface {
-	Store(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64, exposedValueOrRef string) (ExternalID, error)
-	Update(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64, exposedValueOrRef string) error
-	Expose(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64) (secretv1beta1.ExposedSecureValue, error)
-	Delete(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64) error
+	Store(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64, exposedValueOrRef string) (ExternalID, error)
+	Update(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64, exposedValueOrRef string) error
+	Expose(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64) (secretv1beta1.ExposedSecureValue, error)
+	Delete(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64) error
 }
 
 // Service is the interface for secret keeper services.

pkg/registry/apis/secret/encryption/manager/manager.go

@@ -1,10 +1,8 @@
 package manager
 
 import (
-	"bytes"
 	"context"
 	"crypto/rand"
-	"encoding/base64"
 	"errors"
 	"fmt"
 	"strconv"
@@ -20,13 +18,10 @@ import (
 	"github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/encryption"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/encryption/cipher"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	"github.com/grafana/grafana/pkg/util"
 )
 
-const (
-	keyIdDelimiter = '#'
-)
-
 type EncryptionManager struct {
 	tracer     trace.Tracer
 	store      contracts.DataKeyStorage
@@ -99,12 +94,9 @@ func (s *EncryptionManager) registerUsageMetrics() {
 	})
 }
 
-// TODO: Why do we need to use a global variable for this?
-var b64 = base64.RawStdEncoding
-
-func (s *EncryptionManager) Encrypt(ctx context.Context, namespace string, payload []byte) ([]byte, error) {
+func (s *EncryptionManager) Encrypt(ctx context.Context, namespace xkube.Namespace, payload []byte) (contracts.EncryptedPayload, error) {
 	ctx, span := s.tracer.Start(ctx, "EnvelopeEncryptionManager.Encrypt", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 	))
 	defer span.End()
 
@@ -128,34 +120,30 @@ func (s *EncryptionManager) Encrypt(ctx context.Context, namespace string, paylo
 	id, dataKey, err = s.currentDataKey(ctx, namespace, label)
 	if err != nil {
 		s.log.Error("Failed to get current data key", "error", err, "label", label)
-		return nil, err
+		return contracts.EncryptedPayload{}, err
 	}
 
 	var encrypted []byte
 	encrypted, err = s.cipher.Encrypt(ctx, payload, string(dataKey))
 	if err != nil {
 		s.log.Error("Failed to encrypt secret", "error", err)
-		return nil, err
+		return contracts.EncryptedPayload{}, err
 	}
 
-	prefix := make([]byte, b64.EncodedLen(len(id))+2)
-	b64.Encode(prefix[1:], []byte(id))
-	prefix[0] = keyIdDelimiter
-	prefix[len(prefix)-1] = keyIdDelimiter
+	encryptedPayload := contracts.EncryptedPayload{
+		DataKeyID:     id,
+		EncryptedData: encrypted,
+	}
 
-	blob := make([]byte, len(prefix)+len(encrypted))
-	copy(blob, prefix)
-	copy(blob[len(prefix):], encrypted)
-
-	return blob, nil
+	return encryptedPayload, nil
 }
 
 // currentDataKey looks up for current data key in cache or database by name, and decrypts it.
 // If there's no current data key in cache nor in database it generates a new random data key,
 // and stores it into both the in-memory cache and database (encrypted by the encryption provider).
-func (s *EncryptionManager) currentDataKey(ctx context.Context, namespace string, label string) (string, []byte, error) {
+func (s *EncryptionManager) currentDataKey(ctx context.Context, namespace xkube.Namespace, label string) (string, []byte, error) {
 	ctx, span := s.tracer.Start(ctx, "EnvelopeEncryptionManager.CurrentDataKey", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 		attribute.String("label", label),
 	))
 	defer span.End()
@@ -166,14 +154,14 @@ func (s *EncryptionManager) currentDataKey(ctx context.Context, namespace string
 	defer s.mtx.Unlock()
 
 	// We try to fetch the data key, either from cache or database
-	id, dataKey, err := s.dataKeyByLabel(ctx, namespace, label)
+	id, dataKey, err := s.dataKeyByLabel(ctx, namespace.String(), label)
 	if err != nil {
 		return "", nil, err
 	}
 
 	// If no existing data key was found, create a new one
 	if dataKey == nil {
-		id, dataKey, err = s.newDataKey(ctx, namespace, label)
+		id, dataKey, err = s.newDataKey(ctx, namespace.String(), label)
 		if err != nil {
 			return "", nil, err
 		}
@@ -264,9 +252,9 @@ func newRandomDataKey() ([]byte, error) {
 	return rawDataKey, nil
 }
 
-func (s *EncryptionManager) Decrypt(ctx context.Context, namespace string, payload []byte) ([]byte, error) {
+func (s *EncryptionManager) Decrypt(ctx context.Context, namespace xkube.Namespace, payload contracts.EncryptedPayload) ([]byte, error) {
 	ctx, span := s.tracer.Start(ctx, "EnvelopeEncryptionManager.Decrypt", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 	))
 	defer span.End()
 
@@ -285,50 +273,28 @@ func (s *EncryptionManager) Decrypt(ctx context.Context, namespace string, paylo
 		}
 	}()
 
-	if len(payload) == 0 {
+	if len(payload.EncryptedData) == 0 {
 		err = fmt.Errorf("unable to decrypt empty payload")
 		return nil, err
 	}
 
-	payload = payload[1:]
-	endOfKey := bytes.Index(payload, []byte{keyIdDelimiter})
-	if endOfKey == -1 {
-		err = fmt.Errorf("could not find valid key id in encrypted payload")
-		return nil, err
-	}
-	b64Key := payload[:endOfKey]
-	payload = payload[endOfKey+1:]
-	keyId := make([]byte, b64.DecodedLen(len(b64Key)))
-	_, err = b64.Decode(keyId, b64Key)
-	if err != nil {
+	if payload.DataKeyID == "" {
+		err = fmt.Errorf("unable to decrypt empty data key id")
 		return nil, err
 	}
 
-	dataKey, err := s.dataKeyById(ctx, namespace, string(keyId))
+	dataKey, err := s.dataKeyById(ctx, namespace.String(), payload.DataKeyID)
 	if err != nil {
-		s.log.FromContext(ctx).Error("Failed to lookup data key by id", "id", string(keyId), "error", err)
+		s.log.FromContext(ctx).Error("Failed to lookup data key by id", "id", payload.DataKeyID, "error", err)
 		return nil, err
 	}
 
 	var decrypted []byte
-	decrypted, err = s.cipher.Decrypt(ctx, payload, string(dataKey))
+	decrypted, err = s.cipher.Decrypt(ctx, payload.EncryptedData, string(dataKey))
 
 	return decrypted, err
 }
 
-func (s *EncryptionManager) GetDecryptedValue(ctx context.Context, namespace string, sjd map[string][]byte, key, fallback string) string {
-	if value, ok := sjd[key]; ok {
-		decryptedData, err := s.Decrypt(ctx, namespace, value)
-		if err != nil {
-			return fallback
-		}
-
-		return string(decryptedData)
-	}
-
-	return fallback
-}
-
 // dataKeyById looks up for data key in the database and returns it decrypted.
 func (s *EncryptionManager) dataKeyById(ctx context.Context, namespace, id string) ([]byte, error) {
 	ctx, span := s.tracer.Start(ctx, "EnvelopeEncryptionManager.GetDataKey", trace.WithAttributes(

pkg/registry/apis/secret/encryption/manager/manager_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/grafana/grafana/pkg/registry/apis/secret/encryption"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/encryption/cipher/service"
 	osskmsproviders "github.com/grafana/grafana/pkg/registry/apis/secret/encryption/kmsproviders"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	"github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/storage/secret/database"
@@ -34,7 +35,7 @@ func TestMain(m *testing.M) {
 func TestEncryptionService_EnvelopeEncryption(t *testing.T) {
 	svc := setupTestService(t)
 	ctx := context.Background()
-	namespace := "test-namespace"
+	namespace := xkube.Namespace("test-namespace")
 
 	t.Run("encrypting should create DEK", func(t *testing.T) {
 		plaintext := []byte("very secret string")
@@ -46,7 +47,7 @@ func TestEncryptionService_EnvelopeEncryption(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, plaintext, decrypted)
 
-		keys, err := svc.store.ListDataKeys(ctx, namespace)
+		keys, err := svc.store.ListDataKeys(ctx, namespace.String())
 		require.NoError(t, err)
 		assert.Equal(t, len(keys), 1)
 	})
@@ -61,7 +62,7 @@ func TestEncryptionService_EnvelopeEncryption(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, plaintext, decrypted)
 
-		keys, err := svc.store.ListDataKeys(ctx, namespace)
+		keys, err := svc.store.ListDataKeys(ctx, namespace.String())
 		require.NoError(t, err)
 		assert.Equal(t, len(keys), 1)
 	})
@@ -212,7 +213,7 @@ func TestEncryptionService_UseCurrentProvider(t *testing.T) {
 		}
 		encryptionManager.providerConfig.CurrentProvider = encryption.ProviderID("fakeProvider.v1")
 
-		namespace := "test-namespace"
+		namespace := xkube.Namespace("test-namespace")
 		encrypted, _ := encryptionManager.Encrypt(context.Background(), namespace, []byte{})
 		assert.True(t, fake.encryptCalled)
 		assert.False(t, fake.decryptCalled)
@@ -241,7 +242,7 @@ func TestEncryptionService_UseCurrentProvider(t *testing.T) {
 
 func TestEncryptionService_SecretKeyVersionUpgrade(t *testing.T) {
 	ctx := context.Background()
-	namespace := "test-namespace"
+	namespace := xkube.Namespace("test-namespace")
 
 	// Generate random keys for testing
 	oldKey := util.GenerateShortUID() + util.GenerateShortUID() // 32 chars
@@ -416,16 +417,30 @@ func (p *fakeProvider) Decrypt(_ context.Context, _ []byte) ([]byte, error) {
 
 func TestEncryptionService_Decrypt(t *testing.T) {
 	ctx := context.Background()
-	namespace := "test-namespace"
+	namespace := xkube.Namespace("test-namespace")
 
 	t.Run("empty payload should fail", func(t *testing.T) {
 		svc := setupTestService(t)
-		_, err := svc.Decrypt(context.Background(), namespace, []byte(""))
+		_, err := svc.Decrypt(context.Background(), namespace, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte(""),
+		})
 		require.Error(t, err)
 
 		assert.Equal(t, "unable to decrypt empty payload", err.Error())
 	})
 
+	t.Run("empty data key id should fail", func(t *testing.T) {
+		svc := setupTestService(t)
+		_, err := svc.Decrypt(context.Background(), namespace, contracts.EncryptedPayload{
+			DataKeyID:     "",
+			EncryptedData: []byte("some payload"),
+		})
+		require.Error(t, err)
+
+		assert.Equal(t, "unable to decrypt empty data key id", err.Error())
+	})
+
 	t.Run("ee encrypted payload with ee enabled should work", func(t *testing.T) {
 		svc := setupTestService(t)
 		ciphertext, err := svc.Encrypt(ctx, namespace, []byte("grafana"))
@@ -442,7 +457,7 @@ func TestIntegration_SecretsService(t *testing.T) {
 
 	ctx := context.Background()
 	someData := []byte(`some-data`)
-	namespace := "test-namespace"
+	namespace := xkube.Namespace("test-namespace")
 
 	tcs := map[string]func(*testing.T, db.DB, contracts.EncryptionManager){
 		"regular": func(t *testing.T, _ db.DB, svc contracts.EncryptionManager) {
@@ -562,7 +577,7 @@ func TestIntegration_SecretsService(t *testing.T) {
 			require.NoError(t, err)
 
 			ctx := context.Background()
-			namespace := "test-namespace"
+			namespace := xkube.Namespace("test-namespace")
 
 			// Here's what actually matters and varies on each test: look at the test case name.
 			//

pkg/registry/apis/secret/garbagecollectionworker/worker.go

@@ -104,7 +104,7 @@ func (w *Worker) Cleanup(ctx context.Context, sv *secretv1beta1.SecureValue) err
 	}
 
 	// Keeper deletion is idempotent
-	if err := keeper.Delete(ctx, keeperCfg, sv.Namespace, sv.Name, sv.Status.Version); err != nil {
+	if err := keeper.Delete(ctx, keeperCfg, xkube.Namespace(sv.Namespace), sv.Name, sv.Status.Version); err != nil {
 		return fmt.Errorf("deleting secure value from keeper: %w", err)
 	}
 

pkg/registry/apis/secret/garbagecollectionworker/worker_test.go

@@ -9,6 +9,7 @@ import (
 	secretv1beta1 "github.com/grafana/grafana/apps/secret/pkg/apis/secret/v1beta1"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/testutils"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	"github.com/grafana/grafana/pkg/storage/secret/encryption"
 	"github.com/mitchellh/copystructure"
 	"github.com/stretchr/testify/require"
@@ -58,7 +59,7 @@ func TestBasic(t *testing.T) {
 		require.NoError(t, err)
 
 		// Get the secret value once to make sure it's reachable
-		exposedValue, err := keeper.Expose(t.Context(), keeperCfg, sv.Namespace, sv.Name, sv.Status.Version)
+		exposedValue, err := keeper.Expose(t.Context(), keeperCfg, xkube.Namespace(sv.Namespace), sv.Name, sv.Status.Version)
 		require.NoError(t, err)
 		require.NotEmpty(t, exposedValue.DangerouslyExposeAndConsumeValue())
 
@@ -78,7 +79,7 @@ func TestBasic(t *testing.T) {
 		require.Empty(t, svs)
 
 		// Try to get the secreet value again to make sure it's been deleted from the keeper
-		exposedValue, err = keeper.Expose(t.Context(), keeperCfg, sv.Namespace, sv.Name, sv.Status.Version)
+		exposedValue, err = keeper.Expose(t.Context(), keeperCfg, xkube.Namespace(sv.Namespace), sv.Name, sv.Status.Version)
 		require.ErrorIs(t, err, encryption.ErrEncryptedValueNotFound)
 		require.Empty(t, exposedValue)
 	})

pkg/registry/apis/secret/secretkeeper/secretkeeper.go

@@ -1,9 +1,12 @@
 package secretkeeper
 
 import (
+	"fmt"
+
 	"go.opentelemetry.io/otel/trace"
 
 	secretv1beta1 "github.com/grafana/grafana/apps/secret/pkg/apis/secret/v1beta1"
+	"github.com/grafana/grafana/pkg/registry/apis/secret"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/secretkeeper/sqlkeeper"
 	"github.com/prometheus/client_golang/prometheus"
@@ -20,11 +23,17 @@ func ProvideService(
 	tracer trace.Tracer,
 	store contracts.EncryptedValueStorage,
 	encryptionManager contracts.EncryptionManager,
+	migrationExecutor contracts.EncryptedValueMigrationExecutor,
 	reg prometheus.Registerer,
+	_ *secret.DependencyRegisterer, // noop import so wire runs DB migrations before instantiating this service -- can be nil when manually instantiating
 ) (*OSSKeeperService, error) {
+	systemKeeper, err := sqlkeeper.NewSQLKeeper(tracer, encryptionManager, store, migrationExecutor, reg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create system keeper: %w", err)
+	}
+
 	return &OSSKeeperService{
-		// TODO: rename to system keeper or something like that
-		systemKeeper: sqlkeeper.NewSQLKeeper(tracer, encryptionManager, store, reg),
+		systemKeeper: systemKeeper,
 	}, nil
 }
 

pkg/registry/apis/secret/secretkeeper/secretkeeper_test.go

@@ -12,6 +12,7 @@ import (
 	osskmsproviders "github.com/grafana/grafana/pkg/registry/apis/secret/encryption/kmsproviders"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/encryption/manager"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/secretkeeper/sqlkeeper"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/testutils"
 	"github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/storage/secret/database"
@@ -65,7 +66,8 @@ func setupTestService(t *testing.T, cfg *setting.Cfg) (*OSSKeeperService, error)
 	require.NoError(t, err)
 
 	// Initialize the keeper service
-	keeperService, err := ProvideService(tracer, encValueStore, encryptionManager, nil)
+	keeperService, err := ProvideService(tracer, encValueStore, encryptionManager, &testutils.NoopMigrationExecutor{}, nil, nil)
+	require.NoError(t, err)
 
 	return keeperService, err
 }

pkg/registry/apis/secret/secretkeeper/sqlkeeper/keeper.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/grafana/grafana-app-sdk/logging"
 	secretv1beta1 "github.com/grafana/grafana/apps/secret/pkg/apis/secret/v1beta1"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/secretkeeper/metrics"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -26,20 +28,32 @@ func NewSQLKeeper(
 	tracer trace.Tracer,
 	encryptionManager contracts.EncryptionManager,
 	store contracts.EncryptedValueStorage,
+	migrationExecutor contracts.EncryptedValueMigrationExecutor,
 	reg prometheus.Registerer,
-) *SQLKeeper {
+) (*SQLKeeper, error) {
+	// Run the encrypted value store migration before anything else, otherwise operations may fail
+	// TODO: This does not need to be here forever, but we may currently have on-prem deployments using GSM, so it needs to be here for now.
+	// Periodically assess whether it is safe to remove - most likely for G13 should be fine.
+	log := logging.FromContext(context.Background())
+	log.Debug("sqlkeeper: executing encrypted value store migration")
+	rowsAffected, err := migrationExecutor.Execute(context.Background())
+	log.Debug("sqlkeeper: encrypted value store migration completed", "rows_affected", rowsAffected)
+	if err != nil {
+		return nil, fmt.Errorf("error encountered during encrypted value store migration: %w", err)
+	}
+
 	return &SQLKeeper{
 		tracer:            tracer,
 		encryptionManager: encryptionManager,
 		store:             store,
 		metrics:           metrics.NewKeeperMetrics(reg),
-	}
+	}, nil
 }
 
-func (s *SQLKeeper) Store(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64, exposedValueOrRef string) (contracts.ExternalID, error) {
+func (s *SQLKeeper) Store(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64, exposedValueOrRef string) (contracts.ExternalID, error) {
 	ctx, span := s.tracer.Start(ctx, "SQLKeeper.Store",
 		trace.WithAttributes(
-			attribute.String("namespace", namespace),
+			attribute.String("namespace", namespace.String()),
 			attribute.String("name", name),
 			attribute.Int64("version", version)),
 	)
@@ -63,9 +77,9 @@ func (s *SQLKeeper) Store(ctx context.Context, cfg secretv1beta1.KeeperConfig, n
 	return contracts.ExternalID(""), nil
 }
 
-func (s *SQLKeeper) Expose(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64) (secretv1beta1.ExposedSecureValue, error) {
+func (s *SQLKeeper) Expose(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64) (secretv1beta1.ExposedSecureValue, error) {
 	ctx, span := s.tracer.Start(ctx, "SQLKeeper.Expose", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 		attribute.String("name", name),
 		attribute.Int64("version", version),
 	))
@@ -77,7 +91,7 @@ func (s *SQLKeeper) Expose(ctx context.Context, cfg secretv1beta1.KeeperConfig,
 		return "", fmt.Errorf("unable to get encrypted value: %w", err)
 	}
 
-	exposedBytes, err := s.encryptionManager.Decrypt(ctx, namespace, encryptedValue.EncryptedData)
+	exposedBytes, err := s.encryptionManager.Decrypt(ctx, namespace, encryptedValue.EncryptedPayload)
 	if err != nil {
 		return "", fmt.Errorf("unable to decrypt value: %w", err)
 	}
@@ -88,9 +102,9 @@ func (s *SQLKeeper) Expose(ctx context.Context, cfg secretv1beta1.KeeperConfig,
 	return exposedValue, nil
 }
 
-func (s *SQLKeeper) Delete(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64) error {
+func (s *SQLKeeper) Delete(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64) error {
 	ctx, span := s.tracer.Start(ctx, "SQLKeeper.Delete", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 		attribute.String("name", name),
 		attribute.Int64("version", version),
 	))
@@ -107,9 +121,9 @@ func (s *SQLKeeper) Delete(ctx context.Context, cfg secretv1beta1.KeeperConfig,
 	return nil
 }
 
-func (s *SQLKeeper) Update(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace, name string, version int64, exposedValueOrRef string) error {
+func (s *SQLKeeper) Update(ctx context.Context, cfg secretv1beta1.KeeperConfig, namespace xkube.Namespace, name string, version int64, exposedValueOrRef string) error {
 	ctx, span := s.tracer.Start(ctx, "SQLKeeper.Update", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 		attribute.String("name", name),
 		attribute.Int64("version", version),
 	))

pkg/registry/apis/secret/secretkeeper/sqlkeeper/keeper_test.go

@@ -8,6 +8,7 @@ import (
 
 	secretv1beta1 "github.com/grafana/grafana/apps/secret/pkg/apis/secret/v1beta1"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/testutils"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	"github.com/grafana/grafana/pkg/tests/testsuite"
 )
 
@@ -16,10 +17,10 @@ func TestMain(m *testing.M) {
 }
 
 func Test_SQLKeeperSetup(t *testing.T) {
-	namespace1 := "namespace1"
+	namespace1 := xkube.Namespace("namespace1")
 	name1 := "name1"
 	version1 := int64(1)
-	namespace2 := "namespace2"
+	namespace2 := xkube.Namespace("namespace2")
 	name2 := "name2"
 	plaintext1 := "very secret string in namespace 1"
 	plaintext2 := "very secret string in namespace 2"

pkg/registry/apis/secret/service/consolidation.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/grafana/grafana-app-sdk/logging"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	otelcodes "go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -60,21 +61,21 @@ func (s *ConsolidationService) Consolidate(ctx context.Context) (err error) {
 
 	for _, ev := range encryptedValues {
 		// Decrypt the value using its old data key.
-		decryptedValue, err := s.encryptionManager.Decrypt(ctx, ev.Namespace, ev.EncryptedData)
+		decryptedValue, err := s.encryptionManager.Decrypt(ctx, xkube.Namespace(ev.Namespace), ev.EncryptedPayload)
 		if err != nil {
 			logging.FromContext(ctx).Error("Failed to decrypt value", "namespace", ev.Namespace, "name", ev.Name, "error", err)
 			continue
 		}
 
 		// Re-encrypt the value using a new data key.
-		reEncryptedValue, err := s.encryptionManager.Encrypt(ctx, ev.Namespace, decryptedValue)
+		reEncryptedValue, err := s.encryptionManager.Encrypt(ctx, xkube.Namespace(ev.Namespace), decryptedValue)
 		if err != nil {
 			logging.FromContext(ctx).Error("Failed to re-encrypt value", "namespace", ev.Namespace, "name", ev.Name, "error", err)
 			continue
 		}
 
 		// Update the encrypted value in the store.
-		err = s.encryptedValueStore.Update(ctx, ev.Namespace, ev.Name, ev.Version, reEncryptedValue)
+		err = s.encryptedValueStore.Update(ctx, xkube.Namespace(ev.Namespace), ev.Name, ev.Version, reEncryptedValue)
 		if err != nil {
 			logging.FromContext(ctx).Error("Failed to update encrypted value", "namespace", ev.Namespace, "name", ev.Name, "error", err)
 			continue

pkg/registry/apis/secret/service/consolidation_test.go

@@ -97,7 +97,7 @@ func TestConsolidation(t *testing.T) {
 			require.NoError(t, err)
 			originalDecryptedValues = append(originalDecryptedValues, decryptedValue.DangerouslyExposeAndConsumeValue())
 
-			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, tc.namespace, tc.name, 1)
+			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, xkube.Namespace(tc.namespace), tc.name, 1)
 			require.NoError(t, err)
 			require.NotNil(t, encryptedValue)
 			originalEncryptedData = append(originalEncryptedData, encryptedValue.EncryptedData)
@@ -115,7 +115,7 @@ func TestConsolidation(t *testing.T) {
 			require.Equal(t, originalDecryptedValues[i], decryptedValue.DangerouslyExposeAndConsumeValue())
 
 			// Verify that the encrypted data has changed (indicating re-encryption)
-			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, tc.namespace, tc.name, 1)
+			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, xkube.Namespace(tc.namespace), tc.name, 1)
 			require.NoError(t, err)
 			require.NotEqual(t, originalEncryptedData[i], encryptedValue.EncryptedData)
 		}
@@ -174,7 +174,7 @@ func TestConsolidation(t *testing.T) {
 			require.NoError(t, err)
 			initialDecryptedValues = append(initialDecryptedValues, decryptedValue.DangerouslyExposeAndConsumeValue())
 
-			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, tc.namespace, tc.name, 1)
+			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, xkube.Namespace(tc.namespace), tc.name, 1)
 			require.NoError(t, err)
 			initialEncryptedData = append(initialEncryptedData, encryptedValue.EncryptedData)
 		}
@@ -223,7 +223,7 @@ func TestConsolidation(t *testing.T) {
 					require.NoError(t, err)
 					newSecretDecryptedValues = append(newSecretDecryptedValues, decryptedValue.DangerouslyExposeAndConsumeValue())
 
-					encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, tc.namespace, tc.name, 1)
+					encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, xkube.Namespace(tc.namespace), tc.name, 1)
 					require.NoError(t, err)
 					newSecretEncryptedData = append(newSecretEncryptedData, encryptedValue.EncryptedData)
 				}
@@ -252,7 +252,7 @@ func TestConsolidation(t *testing.T) {
 			require.Equal(t, initialDecryptedValues[i], decryptedValue.DangerouslyExposeAndConsumeValue())
 
 			// Verify that the encrypted data has changed (indicating re-encryption)
-			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, tc.namespace, tc.name, 1)
+			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, xkube.Namespace(tc.namespace), tc.name, 1)
 			require.NoError(t, err)
 			require.NotEqual(t, initialEncryptedData[i], encryptedValue.EncryptedData)
 		}
@@ -275,7 +275,7 @@ func TestConsolidation(t *testing.T) {
 
 			// Verify that the encrypted data has changed from what it was when first created
 			// (indicating it was re-encrypted during consolidation)
-			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, tc.namespace, tc.name, 1)
+			encryptedValue, err := sut.EncryptedValueStorage.Get(ctx, xkube.Namespace(tc.namespace), tc.name, 1)
 			require.NoError(t, err)
 			require.NotEqual(t, newSecretEncryptedData[i], encryptedValue.EncryptedData)
 		}

pkg/registry/apis/secret/service/secure_value.go

@@ -146,7 +146,7 @@ func (s *SecureValueService) Update(ctx context.Context, newSecureValue *secretv
 		}
 		logging.FromContext(ctx).Debug("retrieved keeper", "namespace", newSecureValue.Namespace, "keeperName", newSecureValue.Spec.Keeper, "type", keeperCfg.Type())
 
-		secret, err := keeper.Expose(ctx, keeperCfg, newSecureValue.Namespace, newSecureValue.Name, currentVersion.Status.Version)
+		secret, err := keeper.Expose(ctx, keeperCfg, xkube.Namespace(newSecureValue.Namespace), newSecureValue.Name, currentVersion.Status.Version)
 		if err != nil {
 			return nil, false, fmt.Errorf("reading secret value from keeper: %w", err)
 		}
@@ -191,7 +191,7 @@ func (s *SecureValueService) createNewVersion(ctx context.Context, sv *secretv1b
 	// TODO: can we stop using external id?
 	// TODO: store uses only the namespace and returns and id. It could be a kv instead.
 	// TODO: check that the encrypted store works with multiple versions
-	externalID, err := keeper.Store(ctx, keeperCfg, createdSv.Namespace, createdSv.Name, createdSv.Status.Version, sv.Spec.Value.DangerouslyExposeAndConsumeValue())
+	externalID, err := keeper.Store(ctx, keeperCfg, xkube.Namespace(createdSv.Namespace), createdSv.Name, createdSv.Status.Version, sv.Spec.Value.DangerouslyExposeAndConsumeValue())
 	if err != nil {
 		return nil, fmt.Errorf("storing secure value in keeper: %w", err)
 	}

pkg/registry/apis/secret/testutils/testutils.go

@@ -126,7 +126,14 @@ func Setup(t *testing.T, opts ...func(*SetupConfig)) Sut {
 	globalEncryptedValueStorage, err := encryptionstorage.ProvideGlobalEncryptedValueStorage(database, tracer)
 	require.NoError(t, err)
 
-	sqlKeeper := sqlkeeper.NewSQLKeeper(tracer, encryptionManager, encryptedValueStorage, nil)
+	// Initialize a noop migration executor for the sql keeper so it doesn't interfere with initialization
+	noopMigrationExecutor := &NoopMigrationExecutor{}
+	sqlKeeper, err := sqlkeeper.NewSQLKeeper(tracer, encryptionManager, encryptedValueStorage, noopMigrationExecutor, nil)
+	require.NoError(t, err)
+
+	// Initialize a real migration executor for test
+	realMigrationExecutor, err := encryptionstorage.ProvideEncryptedValueMigrationExecutor(database, tracer, encryptedValueStorage, globalEncryptedValueStorage)
+	require.NoError(t, err)
 
 	var keeperService contracts.KeeperService = newKeeperServiceWrapper(sqlKeeper)
 
@@ -158,39 +165,41 @@ func Setup(t *testing.T, opts ...func(*SetupConfig)) Sut {
 		keeperService)
 
 	return Sut{
-		SecureValueService:          secureValueService,
-		SecureValueMetadataStorage:  secureValueMetadataStorage,
-		DecryptStorage:              decryptStorage,
-		DecryptService:              decryptService,
-		EncryptedValueStorage:       encryptedValueStorage,
-		GlobalEncryptedValueStorage: globalEncryptedValueStorage,
-		SQLKeeper:                   sqlKeeper,
-		Database:                    database,
-		AccessClient:                accessClient,
-		ConsolidationService:        consolidationService,
-		EncryptionManager:           encryptionManager,
-		GlobalDataKeyStore:          globalDataKeyStore,
-		GarbageCollectionWorker:     garbageCollectionWorker,
-		Clock:                       clock,
-		KeeperService:               keeperService,
-		KeeperMetadataStorage:       keeperMetadataStorage,
+		SecureValueService:              secureValueService,
+		SecureValueMetadataStorage:      secureValueMetadataStorage,
+		DecryptStorage:                  decryptStorage,
+		DecryptService:                  decryptService,
+		EncryptedValueStorage:           encryptedValueStorage,
+		GlobalEncryptedValueStorage:     globalEncryptedValueStorage,
+		EncryptedValueMigrationExecutor: realMigrationExecutor,
+		SQLKeeper:                       sqlKeeper,
+		Database:                        database,
+		AccessClient:                    accessClient,
+		ConsolidationService:            consolidationService,
+		EncryptionManager:               encryptionManager,
+		GlobalDataKeyStore:              globalDataKeyStore,
+		GarbageCollectionWorker:         garbageCollectionWorker,
+		Clock:                           clock,
+		KeeperService:                   keeperService,
+		KeeperMetadataStorage:           keeperMetadataStorage,
 	}
 }
 
 type Sut struct {
-	SecureValueService          contracts.SecureValueService
-	SecureValueMetadataStorage  contracts.SecureValueMetadataStorage
-	DecryptStorage              contracts.DecryptStorage
-	DecryptService              decryptcontracts.DecryptService
-	EncryptedValueStorage       contracts.EncryptedValueStorage
-	GlobalEncryptedValueStorage contracts.GlobalEncryptedValueStorage
-	SQLKeeper                   *sqlkeeper.SQLKeeper
-	Database                    *database.Database
-	AccessClient                types.AccessClient
-	ConsolidationService        contracts.ConsolidationService
-	EncryptionManager           contracts.EncryptionManager
-	GlobalDataKeyStore          contracts.GlobalDataKeyStorage
-	GarbageCollectionWorker     *garbagecollectionworker.Worker
+	SecureValueService              contracts.SecureValueService
+	SecureValueMetadataStorage      contracts.SecureValueMetadataStorage
+	DecryptStorage                  contracts.DecryptStorage
+	DecryptService                  decryptcontracts.DecryptService
+	EncryptedValueStorage           contracts.EncryptedValueStorage
+	GlobalEncryptedValueStorage     contracts.GlobalEncryptedValueStorage
+	EncryptedValueMigrationExecutor contracts.EncryptedValueMigrationExecutor
+	SQLKeeper                       *sqlkeeper.SQLKeeper
+	Database                        *database.Database
+	AccessClient                    types.AccessClient
+	ConsolidationService            contracts.ConsolidationService
+	EncryptionManager               contracts.EncryptionManager
+	GlobalDataKeyStore              contracts.GlobalDataKeyStorage
+	GarbageCollectionWorker         *garbagecollectionworker.Worker
 	// The fake clock passed to implementations to make testing easier
 	Clock                 *FakeClock
 	KeeperService         contracts.KeeperService
@@ -366,3 +375,10 @@ func (c *FakeClock) Now() time.Time {
 func (c *FakeClock) AdvanceBy(duration time.Duration) {
 	c.Current = c.Current.Add(duration)
 }
+
+type NoopMigrationExecutor struct {
+}
+
+func (e *NoopMigrationExecutor) Execute(ctx context.Context) (int, error) {
+	return 0, nil
+}

pkg/server/wire.go

@@ -444,6 +444,7 @@ var wireBasicSet = wire.NewSet(
 	secretencryption.ProvideGlobalDataKeyStorage,
 	secretencryption.ProvideEncryptedValueStorage,
 	secretencryption.ProvideGlobalEncryptedValueStorage,
+	secretencryption.ProvideEncryptedValueMigrationExecutor,
 	secretsecurevalueservice.ProvideSecureValueService,
 	secretvalidator.ProvideKeeperValidator,
 	secretvalidator.ProvideSecureValueValidator,

pkg/storage/secret/encryption/data/encrypted_value_create.sql

@@ -3,6 +3,7 @@ INSERT INTO {{ .Ident "secret_encrypted_value" }} (
   {{ .Ident "name" }},
   {{ .Ident "version" }},
   {{ .Ident "encrypted_data" }},
+  {{ .Ident "data_key_id" }},
   {{ .Ident "created" }},
   {{ .Ident "updated" }}
 ) VALUES (
@@ -10,6 +11,7 @@ INSERT INTO {{ .Ident "secret_encrypted_value" }} (
   {{ .Arg .Row.Name }},
   {{ .Arg .Row.Version }},
   {{ .Arg .Row.EncryptedData }},
+  {{ .Arg .Row.DataKeyID }},
   {{ .Arg .Row.Created }},
   {{ .Arg .Row.Updated }}
 );

pkg/storage/secret/encryption/data/encrypted_value_list_all.sql

@@ -3,6 +3,7 @@ SELECT
   {{ .Ident "name" }},
   {{ .Ident "version" }},
   {{ .Ident "encrypted_data" }},
+  {{ .Ident "data_key_id" }},
   {{ .Ident "created" }},
   {{ .Ident "updated" }}
 FROM

pkg/storage/secret/encryption/data/encrypted_value_read.sql

@@ -3,6 +3,7 @@ SELECT
   {{ .Ident "name" }},
   {{ .Ident "version" }},
   {{ .Ident "encrypted_data" }},
+  {{ .Ident "data_key_id" }},
   {{ .Ident "created" }},
   {{ .Ident "updated" }}
 FROM

pkg/storage/secret/encryption/data/encrypted_value_update.sql

@@ -2,6 +2,7 @@ UPDATE
   {{ .Ident "secret_encrypted_value" }}
 SET
   {{ .Ident "encrypted_data" }} = {{ .Arg .EncryptedData }},
+  {{ .Ident "data_key_id" }} = {{ .Arg .DataKeyID }},
   {{ .Ident "updated" }} = {{ .Arg .Updated }}
 WHERE 
   {{ .Ident "namespace" }} = {{ .Arg .Namespace }} AND

pkg/storage/secret/encryption/encrypted_value_model.go

@@ -6,6 +6,7 @@ type EncryptedValue struct {
 	Namespace     string
 	Name          string
 	Version       int64
+	DataKeyID     string
 	EncryptedData []byte
 	Created       int64
 	Updated       int64

pkg/storage/secret/encryption/encrypted_value_store.go

@@ -1,7 +1,9 @@
 package encryption
 
 import (
+	"bytes"
 	"context"
+	"encoding/base64"
 	"errors"
 	"fmt"
 	"time"
@@ -10,6 +12,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	"github.com/grafana/grafana/pkg/storage/unified/sql"
 	"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
 )
@@ -37,9 +40,9 @@ type encryptedValStorage struct {
 	tracer  trace.Tracer
 }
 
-func (s *encryptedValStorage) Create(ctx context.Context, namespace, name string, version int64, encryptedData []byte) (ev *contracts.EncryptedValue, err error) {
+func (s *encryptedValStorage) Create(ctx context.Context, namespace xkube.Namespace, name string, version int64, encryptedData contracts.EncryptedPayload) (ev *contracts.EncryptedValue, err error) {
 	ctx, span := s.tracer.Start(ctx, "EncryptedValueStorage.Create", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 	))
 	defer span.End()
 
@@ -56,10 +59,11 @@ func (s *encryptedValStorage) Create(ctx context.Context, namespace, name string
 	createdTime := time.Now().Unix()
 
 	encryptedValue := &EncryptedValue{
-		Namespace:     namespace,
+		Namespace:     namespace.String(),
 		Name:          name,
 		Version:       version,
-		EncryptedData: encryptedData,
+		EncryptedData: encryptedData.EncryptedData,
+		DataKeyID:     encryptedData.DataKeyID,
 		Created:       createdTime,
 		Updated:       createdTime,
 	}
@@ -88,18 +92,21 @@ func (s *encryptedValStorage) Create(ctx context.Context, namespace, name string
 	}
 
 	return &contracts.EncryptedValue{
-		Namespace:     encryptedValue.Namespace,
-		Name:          encryptedValue.Name,
-		Version:       encryptedValue.Version,
-		EncryptedData: encryptedValue.EncryptedData,
-		Created:       encryptedValue.Created,
-		Updated:       encryptedValue.Updated,
+		Namespace: encryptedValue.Namespace,
+		Name:      encryptedValue.Name,
+		Version:   encryptedValue.Version,
+		EncryptedPayload: contracts.EncryptedPayload{
+			DataKeyID:     encryptedValue.DataKeyID,
+			EncryptedData: encryptedValue.EncryptedData,
+		},
+		Created: encryptedValue.Created,
+		Updated: encryptedValue.Updated,
 	}, nil
 }
 
-func (s *encryptedValStorage) Update(ctx context.Context, namespace, name string, version int64, encryptedData []byte) error {
+func (s *encryptedValStorage) Update(ctx context.Context, namespace xkube.Namespace, name string, version int64, encryptedData contracts.EncryptedPayload) error {
 	ctx, span := s.tracer.Start(ctx, "EncryptedValueStorage.Update", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 		attribute.String("name", name),
 		attribute.Int64("version", version),
 	))
@@ -107,10 +114,11 @@ func (s *encryptedValStorage) Update(ctx context.Context, namespace, name string
 
 	req := updateEncryptedValue{
 		SQLTemplate:   sqltemplate.New(s.dialect),
-		Namespace:     namespace,
+		Namespace:     namespace.String(),
 		Name:          name,
 		Version:       version,
-		EncryptedData: encryptedData,
+		EncryptedData: encryptedData.EncryptedData,
+		DataKeyID:     encryptedData.DataKeyID,
 		Updated:       time.Now().Unix(),
 	}
 
@@ -133,9 +141,9 @@ func (s *encryptedValStorage) Update(ctx context.Context, namespace, name string
 	return nil
 }
 
-func (s *encryptedValStorage) Get(ctx context.Context, namespace, name string, version int64) (*contracts.EncryptedValue, error) {
+func (s *encryptedValStorage) Get(ctx context.Context, namespace xkube.Namespace, name string, version int64) (*contracts.EncryptedValue, error) {
 	ctx, span := s.tracer.Start(ctx, "EncryptedValueStorage.Get", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 		attribute.String("name", name),
 		attribute.Int64("version", version),
 	))
@@ -143,7 +151,7 @@ func (s *encryptedValStorage) Get(ctx context.Context, namespace, name string, v
 
 	req := &readEncryptedValue{
 		SQLTemplate: sqltemplate.New(s.dialect),
-		Namespace:   namespace,
+		Namespace:   namespace.String(),
 		Name:        name,
 		Version:     version,
 	}
@@ -163,7 +171,7 @@ func (s *encryptedValStorage) Get(ctx context.Context, namespace, name string, v
 	}
 
 	var encryptedValue EncryptedValue
-	err = rows.Scan(&encryptedValue.Namespace, &encryptedValue.Name, &encryptedValue.Version, &encryptedValue.EncryptedData, &encryptedValue.Created, &encryptedValue.Updated)
+	err = rows.Scan(&encryptedValue.Namespace, &encryptedValue.Name, &encryptedValue.Version, &encryptedValue.EncryptedData, &encryptedValue.DataKeyID, &encryptedValue.Created, &encryptedValue.Updated)
 	if err != nil {
 		return nil, fmt.Errorf("failed to scan encrypted value row: %w", err)
 	}
@@ -172,18 +180,21 @@ func (s *encryptedValStorage) Get(ctx context.Context, namespace, name string, v
 	}
 
 	return &contracts.EncryptedValue{
-		Namespace:     encryptedValue.Namespace,
-		Name:          encryptedValue.Name,
-		Version:       encryptedValue.Version,
-		EncryptedData: encryptedValue.EncryptedData,
-		Created:       encryptedValue.Created,
-		Updated:       encryptedValue.Updated,
+		Namespace: encryptedValue.Namespace,
+		Name:      encryptedValue.Name,
+		Version:   encryptedValue.Version,
+		EncryptedPayload: contracts.EncryptedPayload{
+			DataKeyID:     encryptedValue.DataKeyID,
+			EncryptedData: encryptedValue.EncryptedData,
+		},
+		Created: encryptedValue.Created,
+		Updated: encryptedValue.Updated,
 	}, nil
 }
 
-func (s *encryptedValStorage) Delete(ctx context.Context, namespace, name string, version int64) error {
+func (s *encryptedValStorage) Delete(ctx context.Context, namespace xkube.Namespace, name string, version int64) error {
 	ctx, span := s.tracer.Start(ctx, "EncryptedValueStorage.Delete", trace.WithAttributes(
-		attribute.String("namespace", namespace),
+		attribute.String("namespace", namespace.String()),
 		attribute.String("name", name),
 		attribute.Int64("version", version),
 	))
@@ -191,7 +202,7 @@ func (s *encryptedValStorage) Delete(ctx context.Context, namespace, name string
 
 	req := deleteEncryptedValue{
 		SQLTemplate: sqltemplate.New(s.dialect),
-		Namespace:   namespace,
+		Namespace:   namespace.String(),
 		Name:        name,
 		Version:     version,
 	}
@@ -264,6 +275,7 @@ func (s *globalEncryptedValStorage) ListAll(ctx context.Context, opts contracts.
 			&row.Name,
 			&row.Version,
 			&row.EncryptedData,
+			&row.DataKeyID,
 			&row.Created,
 			&row.Updated,
 		)
@@ -272,12 +284,15 @@ func (s *globalEncryptedValStorage) ListAll(ctx context.Context, opts contracts.
 		}
 
 		encryptedValues = append(encryptedValues, &contracts.EncryptedValue{
-			Namespace:     row.Namespace,
-			Name:          row.Name,
-			Version:       row.Version,
-			EncryptedData: row.EncryptedData,
-			Created:       row.Created,
-			Updated:       row.Updated,
+			Namespace: row.Namespace,
+			Name:      row.Name,
+			Version:   row.Version,
+			EncryptedPayload: contracts.EncryptedPayload{
+				DataKeyID:     row.DataKeyID,
+				EncryptedData: row.EncryptedData,
+			},
+			Created: row.Created,
+			Updated: row.Updated,
 		})
 	}
 	if err := rows.Err(); err != nil {
@@ -329,3 +344,77 @@ func (s *globalEncryptedValStorage) CountAll(ctx context.Context, untilTime *int
 
 	return count, nil
 }
+
+type encryptedValMigrationExecutor struct {
+	db                  contracts.Database
+	dialect             sqltemplate.Dialect
+	tracer              trace.Tracer
+	encryptedValueStore contracts.EncryptedValueStorage
+	globalStore         contracts.GlobalEncryptedValueStorage
+}
+
+func ProvideEncryptedValueMigrationExecutor(
+	db contracts.Database,
+	tracer trace.Tracer,
+	encryptedValueStore contracts.EncryptedValueStorage,
+	globalStore contracts.GlobalEncryptedValueStorage,
+) (contracts.EncryptedValueMigrationExecutor, error) {
+	return &encryptedValMigrationExecutor{
+		db:                  db,
+		dialect:             sqltemplate.DialectForDriver(db.DriverName()),
+		tracer:              tracer,
+		encryptedValueStore: encryptedValueStore,
+		globalStore:         globalStore,
+	}, nil
+}
+
+func (s *encryptedValMigrationExecutor) Execute(ctx context.Context) (int, error) {
+	ctx, span := s.tracer.Start(ctx, "EncryptedValueMigrationExecutor.Execute")
+	defer span.End()
+
+	// 1. Retrieve all encrypted values
+	encryptedValues, err := s.globalStore.ListAll(ctx, contracts.ListOpts{}, nil)
+	if err != nil {
+		return 0, fmt.Errorf("listing all encrypted values: %w", err)
+	}
+
+	// This doesn't need to be done in a single transaction because there's no risk to successful rows if other rows fail
+	rowsAffected := 0
+	for _, encryptedValue := range encryptedValues {
+		// 2. If the value already has the data key id broken out, skip it
+		if encryptedValue.DataKeyID != "" {
+			continue
+		}
+
+		// 3. Split the data key id and the encrypted data out from the encoded payload
+		payload := encryptedValue.EncryptedData
+		const keyIdDelimiter = '#'
+		payload = payload[1:]
+		endOfKey := bytes.Index(payload, []byte{keyIdDelimiter})
+		if endOfKey == -1 {
+			return rowsAffected, fmt.Errorf("could not find valid key id in encrypted payload with namespace %s and name %s and version %d", encryptedValue.Namespace, encryptedValue.Name, encryptedValue.Version)
+		}
+		b64Key := payload[:endOfKey]
+		encryptedData := payload[endOfKey+1:]
+		if len(encryptedData) == 0 {
+			return rowsAffected, fmt.Errorf("encrypted data is empty with namespace %s and name %s and version %d", encryptedValue.Namespace, encryptedValue.Name, encryptedValue.Version)
+		}
+		keyId := make([]byte, base64.RawStdEncoding.DecodedLen(len(b64Key)))
+		_, err := base64.RawStdEncoding.Decode(keyId, b64Key)
+		if err != nil {
+			return rowsAffected, fmt.Errorf("decoding key id with namespace %s and name %s and version %d: %w", encryptedValue.Namespace, encryptedValue.Name, encryptedValue.Version, err)
+		}
+
+		// 4. Update the encrypted value with the data key id and the encrypted data
+		err = s.encryptedValueStore.Update(ctx, xkube.Namespace(encryptedValue.Namespace), encryptedValue.Name, encryptedValue.Version, contracts.EncryptedPayload{
+			DataKeyID:     string(keyId),
+			EncryptedData: encryptedData,
+		})
+		if err != nil {
+			return rowsAffected, fmt.Errorf("updating encrypted value with namespace %s and name %s and version %d: %w", encryptedValue.Namespace, encryptedValue.Name, encryptedValue.Version, err)
+		}
+		rowsAffected++
+	}
+
+	return rowsAffected, nil
+}

pkg/storage/secret/encryption/encrypted_value_store_test.go

@@ -2,15 +2,25 @@ package encryption_test
 
 import (
 	"bytes"
+	"context"
+	"encoding/base64"
 	"errors"
+	"fmt"
 	"slices"
 	"testing"
+	"text/template"
 	"time"
 
+	"github.com/grafana/grafana/pkg/infra/usagestats"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/encryption/cipher"
+	cipherService "github.com/grafana/grafana/pkg/registry/apis/secret/encryption/cipher/service"
 	"github.com/grafana/grafana/pkg/registry/apis/secret/testutils"
+	"github.com/grafana/grafana/pkg/registry/apis/secret/xkube"
 	"github.com/grafana/grafana/pkg/storage/secret/encryption"
+	"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace/noop"
 	"pgregory.net/rapid"
 )
 
@@ -21,7 +31,10 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, []byte("test-data"))
+		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 		require.NotEmpty(t, createdEV.Namespace)
 		require.NotEmpty(t, createdEV.Name)
@@ -36,10 +49,13 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, []byte("test-data"))
+		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 
-		obtainedEV, err := sut.EncryptedValueStorage.Get(t.Context(), createdEV.Namespace, createdEV.Name, createdEV.Version)
+		obtainedEV, err := sut.EncryptedValueStorage.Get(t.Context(), xkube.Namespace(createdEV.Namespace), createdEV.Name, createdEV.Version)
 		require.NoError(t, err)
 
 		require.Equal(t, createdEV.Namespace, obtainedEV.Namespace)
@@ -47,6 +63,7 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		require.Equal(t, createdEV.Created, obtainedEV.Created)
 		require.Equal(t, createdEV.Updated, obtainedEV.Updated)
 		require.Equal(t, createdEV.EncryptedData, obtainedEV.EncryptedData)
+		require.Equal(t, createdEV.DataKeyID, obtainedEV.DataKeyID)
 		require.Equal(t, createdEV.Namespace, obtainedEV.Namespace)
 	})
 
@@ -54,7 +71,10 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "ns1", "test-name", 1, []byte("test-data"))
+		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "ns1", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 
 		obtainedEV, err := sut.EncryptedValueStorage.Get(t.Context(), "ns2", createdEV.Name, createdEV.Version)
@@ -78,16 +98,23 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, []byte("test-data"))
+		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 
-		err = sut.EncryptedValueStorage.Update(t.Context(), createdEV.Namespace, createdEV.Name, createdEV.Version, []byte("test-data-updated"))
+		err = sut.EncryptedValueStorage.Update(t.Context(), xkube.Namespace(createdEV.Namespace), createdEV.Name, createdEV.Version, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id-updated",
+			EncryptedData: []byte("test-data-updated"),
+		})
 		require.NoError(t, err)
 
-		updatedEV, err := sut.EncryptedValueStorage.Get(t.Context(), createdEV.Namespace, createdEV.Name, createdEV.Version)
+		updatedEV, err := sut.EncryptedValueStorage.Get(t.Context(), xkube.Namespace(createdEV.Namespace), createdEV.Name, createdEV.Version)
 		require.NoError(t, err)
 
 		require.Equal(t, []byte("test-data-updated"), updatedEV.EncryptedData)
+		require.Equal(t, "test-data-key-id-updated", updatedEV.DataKeyID)
 		require.Equal(t, createdEV.Created, updatedEV.Created)
 		require.Equal(t, createdEV.Namespace, updatedEV.Namespace)
 	})
@@ -96,7 +123,10 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		err := sut.EncryptedValueStorage.Update(t.Context(), "test-namespace", "test-uid", 1, []byte("test-data"))
+		err := sut.EncryptedValueStorage.Update(t.Context(), "test-namespace", "test-uid", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.Error(t, err)
 	})
 
@@ -104,16 +134,19 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, []byte("ttttest-data"))
+		createdEV, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("ttttest-data"),
+		})
 		require.NoError(t, err)
 
-		_, err = sut.EncryptedValueStorage.Get(t.Context(), createdEV.Namespace, createdEV.Name, createdEV.Version)
+		_, err = sut.EncryptedValueStorage.Get(t.Context(), xkube.Namespace(createdEV.Namespace), createdEV.Name, createdEV.Version)
 		require.NoError(t, err)
 
-		err = sut.EncryptedValueStorage.Delete(t.Context(), createdEV.Namespace, createdEV.Name, createdEV.Version)
+		err = sut.EncryptedValueStorage.Delete(t.Context(), xkube.Namespace(createdEV.Namespace), createdEV.Name, createdEV.Version)
 		require.NoError(t, err)
 
-		obtainedEV, err := sut.EncryptedValueStorage.Get(t.Context(), createdEV.Namespace, createdEV.Name, createdEV.Version)
+		obtainedEV, err := sut.EncryptedValueStorage.Get(t.Context(), xkube.Namespace(createdEV.Namespace), createdEV.Name, createdEV.Version)
 		require.Error(t, err)
 		require.Nil(t, obtainedEV)
 	})
@@ -130,10 +163,16 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		createdEvA, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-a", "test-name", 1, []byte("test-data"))
+		createdEvA, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-a", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 
-		createdEvB, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-b", "test-name", 1, []byte("test-data"))
+		createdEvB, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-b", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 
 		// List all encrypted values, without pagination
@@ -180,10 +219,16 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 		t.Parallel()
 
 		sut := testutils.Setup(t)
-		_, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-a", "test-name", 1, []byte("test-data"))
+		_, err := sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-a", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 
-		_, err = sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-b", "test-name", 1, []byte("test-data"))
+		_, err = sut.EncryptedValueStorage.Create(t.Context(), "test-namespace-b", "test-name", 1, contracts.EncryptedPayload{
+			DataKeyID:     "test-data-key-id",
+			EncryptedData: []byte("test-data"),
+		})
 		require.NoError(t, err)
 
 		count, err := sut.GlobalEncryptedValueStorage.CountAll(t.Context(), nil)
@@ -198,6 +243,281 @@ func TestEncryptedValueStoreImpl(t *testing.T) {
 	})
 }
 
+func TestEncryptedValueMigration(t *testing.T) {
+	t.Parallel()
+
+	t.Run("golden path - successful migration of legacy format", func(t *testing.T) {
+		t.Parallel()
+
+		sut := testutils.Setup(t)
+		tracer := noop.NewTracerProvider().Tracer("test")
+		usageStats := &usagestats.UsageStatsMock{T: t}
+		enc, err := cipherService.ProvideAESGCMCipherService(tracer, usageStats)
+		require.NoError(t, err)
+
+		testCases := []struct {
+			namespace string
+			name      string
+			version   int64
+			plaintext string
+			dataKeyId string
+		}{
+			{
+				namespace: "test-namespace-1",
+				name:      "test-name-1",
+				version:   1,
+				plaintext: "test-plaintext-1",
+				dataKeyId: "test-data-key-id-1",
+			},
+			{
+				namespace: "test-namespace-1",
+				name:      "test-name-2",
+				version:   1,
+				plaintext: "test-plaintext-2",
+				dataKeyId: "test-data-key-id-1",
+			},
+			{
+				namespace: "test-namespace-2",
+				name:      "test-name-3",
+				version:   1,
+				plaintext: "test-plaintext-3",
+				dataKeyId: "test-data-key-id-2",
+			},
+		}
+
+		// Seed with data in the legacy format
+		for _, tc := range testCases {
+			err := createLegacyEncryptedData(t, sut, enc, tc.namespace, tc.name, tc.version, tc.plaintext, tc.dataKeyId)
+			require.NoError(t, err)
+		}
+
+		// Run the migration and blindy trust it
+		rowsAffected, err := sut.EncryptedValueMigrationExecutor.Execute(t.Context())
+		require.NoError(t, err)
+		require.Equal(t, len(testCases), rowsAffected)
+
+		// Now validate that the data is in the new format
+		encryptedValues, err := sut.GlobalEncryptedValueStorage.ListAll(t.Context(), contracts.ListOpts{}, nil)
+		require.NoError(t, err)
+		require.Len(t, encryptedValues, 3)
+
+		for _, tc := range testCases {
+			ev, err := sut.EncryptedValueStorage.Get(t.Context(), xkube.Namespace(tc.namespace), tc.name, tc.version)
+			require.NoError(t, err)
+
+			// Decrypt the encrypted data and check for equality
+			decrypted, err := enc.Decrypt(t.Context(), ev.EncryptedData, tc.dataKeyId)
+			require.NoError(t, err)
+			require.Equal(t, tc.dataKeyId, ev.DataKeyID)
+			require.Equal(t, tc.plaintext, string(decrypted))
+		}
+	})
+
+	t.Run("error conditions - handles corrupt data gracefully", func(t *testing.T) {
+		t.Parallel()
+
+		tracer := noop.NewTracerProvider().Tracer("test")
+		sut := testutils.Setup(t)
+
+		t.Run("global store list error", func(t *testing.T) {
+			mockGlobalStore := &mockGlobalEncryptedValueStorage{
+				listAllError: errors.New("database connection failed"),
+			}
+
+			migrationExecutor, err := encryption.ProvideEncryptedValueMigrationExecutor(
+				sut.Database,
+				tracer,
+				sut.EncryptedValueStorage,
+				mockGlobalStore,
+			)
+			require.NoError(t, err)
+
+			rowsAffected, err := migrationExecutor.Execute(t.Context())
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "listing all encrypted values")
+			require.Equal(t, 0, rowsAffected)
+		})
+
+		t.Run("corrupt data - missing key delimiter", func(t *testing.T) {
+			mockGlobalStore := &mockGlobalEncryptedValueStorage{
+				encryptedValues: []*contracts.EncryptedValue{
+					{
+						Namespace: "test-ns",
+						Name:      "test-name",
+						Version:   1,
+						EncryptedPayload: contracts.EncryptedPayload{
+							EncryptedData: []byte("corrupt-data-without-delimiter"),
+							DataKeyID:     "", // Empty to trigger migration
+						},
+					},
+				},
+			}
+
+			migrationExecutor, err := encryption.ProvideEncryptedValueMigrationExecutor(
+				sut.Database,
+				tracer,
+				sut.EncryptedValueStorage,
+				mockGlobalStore,
+			)
+			require.NoError(t, err)
+
+			rowsAffected, err := migrationExecutor.Execute(t.Context())
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "could not find valid key id in encrypted payload")
+			require.Equal(t, 0, rowsAffected)
+		})
+
+		t.Run("corrupt data - empty encrypted data", func(t *testing.T) {
+			mockGlobalStore := &mockGlobalEncryptedValueStorage{
+				encryptedValues: []*contracts.EncryptedValue{
+					{
+						Namespace: "test-ns",
+						Name:      "test-name",
+						Version:   1,
+						EncryptedPayload: contracts.EncryptedPayload{
+							EncryptedData: []byte("#dGVzdA#"), // Valid key but no encrypted data after delimiter
+							DataKeyID:     "",                 // Empty to trigger migration
+						},
+					},
+				},
+			}
+
+			migrationExecutor, err := encryption.ProvideEncryptedValueMigrationExecutor(
+				sut.Database,
+				tracer,
+				sut.EncryptedValueStorage,
+				mockGlobalStore,
+			)
+			require.NoError(t, err)
+
+			rowsAffected, err := migrationExecutor.Execute(t.Context())
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "encrypted data is empty")
+			require.Equal(t, 0, rowsAffected)
+		})
+
+		t.Run("corrupt data - invalid base64 key", func(t *testing.T) {
+			mockGlobalStore := &mockGlobalEncryptedValueStorage{
+				encryptedValues: []*contracts.EncryptedValue{
+					{
+						Namespace: "test-ns",
+						Name:      "test-name",
+						Version:   1,
+						EncryptedPayload: contracts.EncryptedPayload{
+							EncryptedData: []byte("#invalid-base64!@#$%^&*()#somedata"),
+							DataKeyID:     "", // Empty to trigger migration
+						},
+					},
+				},
+			}
+
+			migrationExecutor, err := encryption.ProvideEncryptedValueMigrationExecutor(
+				sut.Database,
+				tracer,
+				sut.EncryptedValueStorage,
+				mockGlobalStore,
+			)
+			require.NoError(t, err)
+
+			rowsAffected, err := migrationExecutor.Execute(t.Context())
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "decoding key id")
+			require.Equal(t, 0, rowsAffected)
+		})
+
+		t.Run("update failure", func(t *testing.T) {
+			mockGlobalStore := &mockGlobalEncryptedValueStorage{
+				encryptedValues: []*contracts.EncryptedValue{
+					{
+						Namespace: "nonexistent-ns",
+						Name:      "nonexistent-name",
+						Version:   999,
+						EncryptedPayload: contracts.EncryptedPayload{
+							EncryptedData: []byte("#dGVzdA#someencrypteddata"),
+							DataKeyID:     "", // Empty to trigger migration
+						},
+					},
+				},
+			}
+
+			migrationExecutor, err := encryption.ProvideEncryptedValueMigrationExecutor(
+				sut.Database,
+				tracer,
+				sut.EncryptedValueStorage,
+				mockGlobalStore,
+			)
+			require.NoError(t, err)
+
+			rowsAffected, err := migrationExecutor.Execute(t.Context())
+			require.Error(t, err)
+			require.Contains(t, err.Error(), "updating encrypted value")
+			require.Equal(t, 0, rowsAffected)
+		})
+	})
+}
+
+// Helper function that bypasses interfaces and creates data in the legacy format directly in the database.
+// The format is "#{encoded_key_id}#{encrypted_data}".
+func createLegacyEncryptedData(t *testing.T, sut testutils.Sut, enc cipher.Cipher, namespace, name string, version int64, plaintext string, dataKeyId string) error {
+	t.Helper()
+
+	encryptedData, err := enc.Encrypt(t.Context(), []byte(plaintext), dataKeyId)
+	require.NoError(t, err)
+
+	// Encode using the legacy format
+	const keyIdDelimiter = '#'
+	prefix := make([]byte, base64.RawStdEncoding.EncodedLen(len(dataKeyId))+2)
+	base64.RawStdEncoding.Encode(prefix[1:], []byte(dataKeyId))
+	prefix[0] = keyIdDelimiter
+	prefix[len(prefix)-1] = keyIdDelimiter
+
+	blob := make([]byte, len(prefix)+len(encryptedData))
+	copy(blob, prefix)
+	copy(blob[len(prefix):], encryptedData)
+
+	createdTime := time.Now().Unix()
+
+	encryptedValue := &encryption.EncryptedValue{
+		Namespace:     namespace,
+		Name:          name,
+		Version:       version,
+		EncryptedData: blob,
+		DataKeyID:     "",
+		Created:       createdTime,
+		Updated:       createdTime,
+	}
+
+	req := struct {
+		sqltemplate.SQLTemplate
+		Row *encryption.EncryptedValue
+	}{
+		SQLTemplate: sqltemplate.New(sqltemplate.DialectForDriver(sut.Database.DriverName())),
+		Row:         encryptedValue,
+	}
+	tmpl, err := template.ParseFiles("data/encrypted_value_create.sql")
+	if err != nil {
+		return fmt.Errorf("parsing template: %w", err)
+	}
+
+	query, err := sqltemplate.Execute(tmpl, req)
+	if err != nil {
+		return fmt.Errorf("executing template: %w", err)
+	}
+
+	res, err := sut.Database.ExecContext(t.Context(), query, req.GetArgs()...)
+	if err != nil {
+		return fmt.Errorf("inserting row: %w", err)
+	}
+
+	if rowsAffected, err := res.RowsAffected(); err != nil {
+		return fmt.Errorf("getting rows affected: %w", err)
+	} else if rowsAffected != 1 {
+		return fmt.Errorf("expected 1 row affected, got %d", rowsAffected)
+	}
+
+	return nil
+}
+
 func TestStateMachine(t *testing.T) {
 	t.Parallel()
 
@@ -212,10 +532,14 @@ func TestStateMachine(t *testing.T) {
 				ns := namespaceGen.Draw(t, "ns")
 				name := nameGen.Draw(t, "name")
 				version := versionGen.Draw(t, "version")
+				dataKeyId := rapid.String().Draw(t, "dataKeyId")
 				plaintext := rapid.String().Draw(t, "plaintext")
 
-				_, modelErr := m.create(ns, name, version, []byte(plaintext))
-				_, err := sut.EncryptedValueStorage.Create(t.Context(), ns, name, version, []byte(plaintext))
+				_, modelErr := m.create(ns, name, version, []byte(plaintext), dataKeyId)
+				_, err := sut.EncryptedValueStorage.Create(t.Context(), xkube.Namespace(ns), name, version, contracts.EncryptedPayload{
+					DataKeyID:     dataKeyId,
+					EncryptedData: []byte(plaintext),
+				})
 				if modelErr != nil || err != nil {
 					require.ErrorIs(t, err, modelErr)
 					return
@@ -225,10 +549,14 @@ func TestStateMachine(t *testing.T) {
 				ns := namespaceGen.Draw(t, "ns")
 				name := nameGen.Draw(t, "name")
 				version := versionGen.Draw(t, "version")
+				dataKeyId := rapid.String().Draw(t, "dataKeyId")
 				plaintext := rapid.String().Draw(t, "plaintext")
 
-				modelErr := m.update(ns, name, version, []byte(plaintext))
-				err := sut.EncryptedValueStorage.Update(t.Context(), ns, name, version, []byte(plaintext))
+				modelErr := m.update(ns, name, version, []byte(plaintext), dataKeyId)
+				err := sut.EncryptedValueStorage.Update(t.Context(), xkube.Namespace(ns), name, version, contracts.EncryptedPayload{
+					DataKeyID:     dataKeyId,
+					EncryptedData: []byte(plaintext),
+				})
 				if modelErr != nil || err != nil {
 					require.ErrorIs(t, err, modelErr)
 					return
@@ -240,7 +568,7 @@ func TestStateMachine(t *testing.T) {
 				version := versionGen.Draw(t, "version")
 
 				modelValue, modelErr := m.get(ns, name, version)
-				value, err := sut.EncryptedValueStorage.Get(t.Context(), ns, name, version)
+				value, err := sut.EncryptedValueStorage.Get(t.Context(), xkube.Namespace(ns), name, version)
 				if modelErr != nil || err != nil {
 					require.ErrorIs(t, err, modelErr)
 					return
@@ -258,7 +586,7 @@ func TestStateMachine(t *testing.T) {
 				version := versionGen.Draw(t, "version")
 
 				modelErr := m.delete(ns, name, version)
-				err := sut.EncryptedValueStorage.Delete(t.Context(), ns, name, version)
+				err := sut.EncryptedValueStorage.Delete(t.Context(), xkube.Namespace(ns), name, version)
 				if modelErr != nil || err != nil {
 					require.ErrorIs(t, err, modelErr)
 					return
@@ -290,18 +618,19 @@ type entry struct {
 	name          string
 	version       int64
 	encryptedData []byte
+	dataKeyId     string
 }
 
 func newModel() *model {
 	return &model{}
 }
 
-func (m *model) create(namespace, name string, version int64, encryptedData []byte) (*contracts.EncryptedValue, error) {
+func (m *model) create(namespace, name string, version int64, encryptedData []byte, dataKeyId string) (*contracts.EncryptedValue, error) {
 	v, err := m.get(namespace, name, version)
 	if err != nil && !errors.Is(err, encryption.ErrEncryptedValueNotFound) {
 		return nil, err
 	}
-	// The entry being creted already exists
+	// The entry being created already exists
 	if v != nil {
 		return nil, encryption.ErrEncryptedValueAlreadyExists
 	}
@@ -311,20 +640,25 @@ func (m *model) create(namespace, name string, version int64, encryptedData []by
 		name:          name,
 		version:       version,
 		encryptedData: encryptedData,
+		dataKeyId:     dataKeyId,
 	})
 	return &contracts.EncryptedValue{
-		Namespace:     namespace,
-		Name:          name,
-		Version:       version,
-		EncryptedData: encryptedData,
-		Created:       1,
-		Updated:       1,
+		Namespace: namespace,
+		Name:      name,
+		Version:   version,
+		EncryptedPayload: contracts.EncryptedPayload{
+			DataKeyID:     dataKeyId,
+			EncryptedData: encryptedData,
+		},
+		Created: 1,
+		Updated: 1,
 	}, nil
 }
-func (m *model) update(namespace, name string, version int64, encryptedData []byte) error {
+func (m *model) update(namespace, name string, version int64, encryptedData []byte, dataKeyId string) error {
 	for _, v := range m.entries {
 		if v.namespace == namespace && v.name == name && v.version == version {
 			v.encryptedData = encryptedData
+			v.dataKeyId = dataKeyId
 			return nil
 		}
 	}
@@ -336,12 +670,15 @@ func (m *model) get(namespace, name string, version int64) (*contracts.Encrypted
 	for _, v := range m.entries {
 		if v.namespace == namespace && v.name == name && v.version == version {
 			return &contracts.EncryptedValue{
-				Namespace:     namespace,
-				Name:          name,
-				Version:       version,
-				EncryptedData: v.encryptedData,
-				Created:       1,
-				Updated:       1,
+				Namespace: namespace,
+				Name:      name,
+				Version:   version,
+				EncryptedPayload: contracts.EncryptedPayload{
+					DataKeyID:     v.dataKeyId,
+					EncryptedData: v.encryptedData,
+				},
+				Created: 1,
+				Updated: 1,
 			}, nil
 		}
 	}
@@ -354,3 +691,25 @@ func (m *model) delete(namespace, name string, version int64) error {
 	})
 	return nil
 }
+
+// mockGlobalEncryptedValueStorage is a mock implementation of contracts.GlobalEncryptedValueStorage
+// used for testing error conditions in the migration executor
+type mockGlobalEncryptedValueStorage struct {
+	encryptedValues []*contracts.EncryptedValue
+	listAllError    error
+	countAllError   error
+}
+
+func (m *mockGlobalEncryptedValueStorage) ListAll(ctx context.Context, opts contracts.ListOpts, untilTime *int64) ([]*contracts.EncryptedValue, error) {
+	if m.listAllError != nil {
+		return nil, m.listAllError
+	}
+	return m.encryptedValues, nil
+}
+
+func (m *mockGlobalEncryptedValueStorage) CountAll(ctx context.Context, untilTime *int64) (int64, error) {
+	if m.countAllError != nil {
+		return 0, m.countAllError
+	}
+	return int64(len(m.encryptedValues)), nil
+}

pkg/storage/secret/encryption/query.go

@@ -74,6 +74,7 @@ type updateEncryptedValue struct {
 	Name          string
 	Version       int64
 	EncryptedData []byte
+	DataKeyID     string
 	Updated       int64
 }
 

pkg/storage/secret/encryption/query_test.go

@@ -24,6 +24,7 @@ func TestEncryptedValueQueries(t *testing.T) {
 							Name:          "n1",
 							Version:       1,
 							EncryptedData: []byte("secret"),
+							DataKeyID:     "test-data-key-id",
 							Created:       1234,
 							Updated:       5678,
 						},
@@ -50,6 +51,7 @@ func TestEncryptedValueQueries(t *testing.T) {
 						Name:          "n1",
 						Version:       1,
 						EncryptedData: []byte("secret"),
+						DataKeyID:     "test-data-key-id",
 						Updated:       5679,
 					},
 				},

pkg/storage/secret/encryption/testdata/mysql--encrypted_value_create-create.sql

@@ -3,6 +3,7 @@ INSERT INTO `secret_encrypted_value` (
   `name`,
   `version`,
   `encrypted_data`,
+  `data_key_id`,
   `created`,
   `updated`
 ) VALUES (
@@ -10,6 +11,7 @@ INSERT INTO `secret_encrypted_value` (
   'n1',
   1,
   '[115 101 99 114 101 116]',
+  'test-data-key-id',
   1234,
   5678
 );

pkg/storage/secret/encryption/testdata/mysql--encrypted_value_list_all-list_all.sql

@@ -3,6 +3,7 @@ SELECT
   `name`,
   `version`,
   `encrypted_data`,
+  `data_key_id`,
   `created`,
   `updated`
 FROM

pkg/storage/secret/encryption/testdata/mysql--encrypted_value_list_all-list_all_until_time.sql

@@ -3,6 +3,7 @@ SELECT
   `name`,
   `version`,
   `encrypted_data`,
+  `data_key_id`,
   `created`,
   `updated`
 FROM

pkg/storage/secret/encryption/testdata/mysql--encrypted_value_list_all-list_limit_10_offset_0.sql

@@ -3,6 +3,7 @@ SELECT
   `name`,
   `version`,
   `encrypted_data`,
+  `data_key_id`,
   `created`,
   `updated`
 FROM

pkg/storage/secret/encryption/testdata/mysql--encrypted_value_list_all-list_limit_10_offset_2.sql

@@ -3,6 +3,7 @@ SELECT
   `name`,
   `version`,
   `encrypted_data`,
+  `data_key_id`,
   `created`,
   `updated`
 FROM

pkg/storage/secret/encryption/testdata/mysql--encrypted_value_read-read.sql

@@ -3,6 +3,7 @@ SELECT
   `name`,
   `version`,
   `encrypted_data`,
+  `data_key_id`,
   `created`,
   `updated`
 FROM

pkg/storage/secret/encryption/testdata/mysql--encrypted_value_update-update.sql

@@ -2,6 +2,7 @@ UPDATE
   `secret_encrypted_value`
 SET
   `encrypted_data` = '[115 101 99 114 101 116]',
+  `data_key_id` = 'test-data-key-id',
   `updated` = 5679
 WHERE 
   `namespace` = 'ns' AND

pkg/storage/secret/encryption/testdata/postgres--encrypted_value_create-create.sql

@@ -3,6 +3,7 @@ INSERT INTO "secret_encrypted_value" (
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 ) VALUES (
@@ -10,6 +11,7 @@ INSERT INTO "secret_encrypted_value" (
   'n1',
   1,
   '[115 101 99 114 101 116]',
+  'test-data-key-id',
   1234,
   5678
 );

pkg/storage/secret/encryption/testdata/postgres--encrypted_value_list_all-list_all.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/postgres--encrypted_value_list_all-list_all_until_time.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/postgres--encrypted_value_list_all-list_limit_10_offset_0.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/postgres--encrypted_value_list_all-list_limit_10_offset_2.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/postgres--encrypted_value_read-read.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/postgres--encrypted_value_update-update.sql

@@ -2,6 +2,7 @@ UPDATE
   "secret_encrypted_value"
 SET
   "encrypted_data" = '[115 101 99 114 101 116]',
+  "data_key_id" = 'test-data-key-id',
   "updated" = 5679
 WHERE 
   "namespace" = 'ns' AND

pkg/storage/secret/encryption/testdata/sqlite--encrypted_value_create-create.sql

@@ -3,6 +3,7 @@ INSERT INTO "secret_encrypted_value" (
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 ) VALUES (
@@ -10,6 +11,7 @@ INSERT INTO "secret_encrypted_value" (
   'n1',
   1,
   '[115 101 99 114 101 116]',
+  'test-data-key-id',
   1234,
   5678
 );

pkg/storage/secret/encryption/testdata/sqlite--encrypted_value_list_all-list_all.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/sqlite--encrypted_value_list_all-list_all_until_time.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/sqlite--encrypted_value_list_all-list_limit_10_offset_0.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/sqlite--encrypted_value_list_all-list_limit_10_offset_2.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/sqlite--encrypted_value_read-read.sql

@@ -3,6 +3,7 @@ SELECT
   "name",
   "version",
   "encrypted_data",
+  "data_key_id",
   "created",
   "updated"
 FROM

pkg/storage/secret/encryption/testdata/sqlite--encrypted_value_update-update.sql

@@ -2,6 +2,7 @@ UPDATE
   "secret_encrypted_value"
 SET
   "encrypted_data" = '[115 101 99 114 101 116]',
+  "data_key_id" = 'test-data-key-id',
   "updated" = 5679
 WHERE 
   "namespace" = 'ns' AND

pkg/storage/secret/metadata/decrypt_store.go

@@ -134,7 +134,7 @@ func (s *decryptStorage) Decrypt(ctx context.Context, namespace xkube.Namespace,
 		return "", fmt.Errorf("failed to get keeper for config: %v (%w)", err, contracts.ErrDecryptFailed)
 	}
 
-	exposedValue, err := keeper.Expose(ctx, keeperConfig, namespace.String(), name, sv.Status.Version)
+	exposedValue, err := keeper.Expose(ctx, keeperConfig, namespace, name, sv.Status.Version)
 	if err != nil {
 		return "", fmt.Errorf("failed to expose secret: %v (%w)", err, contracts.ErrDecryptFailed)
 	}

pkg/storage/secret/migrator/migrator.go

@@ -200,4 +200,15 @@ func (*SecretDB) AddMigration(mg *migrator.Migrator) {
 	mg.AddMigration("add lease_created index to "+TableNameSecureValue, migrator.NewAddIndexMigration(secureValueTable, &migrator.Index{
 		Cols: []string{"lease_created"},
 	}))
+
+	mg.AddMigration("add data_key_id column to "+TableNameEncryptedValue, migrator.NewAddColumnMigration(encryptedValueTable, &migrator.Column{
+		Name:     "data_key_id",
+		Type:     migrator.DB_NVarchar,
+		Length:   100,
+		Nullable: false,
+		Default:  "''",
+	}))
+	mg.AddMigration("add data_key_id index to "+TableNameEncryptedValue, migrator.NewAddIndexMigration(encryptedValueTable, &migrator.Index{
+		Cols: []string{"data_key_id"},
+	}))
 }