fix: prevent implicit conversions to/from EncryptedColumn (#2668)

Make EncryptedColumn a struct so we can't accidentally implicitly
convert `[]byte` to/from it. As a side-effect, discovered a few columns
in the DB that weren't the correct types.

backend/controller/controller.go

@@ -253,7 +253,7 @@ func New(ctx context.Context, conn *sql.DB, config Config, devel bool) (*Service
 	svc.routes.Store(map[string][]dal.Route{})
 	svc.schema.Store(&schema.Schema{})
 
-	cronSvc := cronjobs.New(ctx, key, svc.config.Advertise.Host, conn)
+	cronSvc := cronjobs.New(ctx, key, svc.config.Advertise.Host, encryptionSrv, conn)
 	svc.cronJobs = cronSvc
 
 	pubSub := pubsub.New(ctx, db, svc.tasks, svc)

backend/controller/cronjobs/cronjobs.go

@@ -10,9 +10,11 @@ import (
 
 	"github.com/TBD54566975/ftl/backend/controller/cronjobs/dal"
 	parentdal "github.com/TBD54566975/ftl/backend/controller/dal"
+	encryptionsvc "github.com/TBD54566975/ftl/backend/controller/encryption"
 	schemapb "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/schema"
 	"github.com/TBD54566975/ftl/backend/schema"
 	"github.com/TBD54566975/ftl/internal/cron"
+	"github.com/TBD54566975/ftl/internal/encryption"
 	"github.com/TBD54566975/ftl/internal/log"
 	"github.com/TBD54566975/ftl/internal/model"
 )
@@ -21,18 +23,20 @@ type Service struct {
 	key           model.ControllerKey
 	requestSource string
 	dal           dal.DAL
+	encryption    *encryptionsvc.Service
 	clock         clock.Clock
 }
 
-func New(ctx context.Context, key model.ControllerKey, requestSource string, conn *sql.DB) *Service {
-	return NewForTesting(ctx, key, requestSource, *dal.New(conn), clock.New())
+func New(ctx context.Context, key model.ControllerKey, requestSource string, encryption *encryptionsvc.Service, conn *sql.DB) *Service {
+	return NewForTesting(ctx, key, requestSource, encryption, *dal.New(conn), clock.New())
 }
 
-func NewForTesting(ctx context.Context, key model.ControllerKey, requestSource string, dal dal.DAL, clock clock.Clock) *Service {
+func NewForTesting(ctx context.Context, key model.ControllerKey, requestSource string, encryption *encryptionsvc.Service, dal dal.DAL, clock clock.Clock) *Service {
 	svc := &Service{
 		key:           key,
 		requestSource: requestSource,
 		dal:           dal,
+		encryption:    encryption,
 		clock:         clock,
 	}
 	return svc
@@ -174,11 +178,16 @@ func (s *Service) scheduleCronJob(ctx context.Context, tx *dal.DAL, job model.Cr
 
 	logger.Tracef("Scheduling cron job %q async_call execution at %s", job.Key, nextAttemptForJob)
 	origin := &parentdal.AsyncOriginCron{CronJobKey: job.Key}
+	var request encryption.EncryptedColumn[encryption.AsyncSubKey]
+	err = s.encryption.Encrypt([]byte(`{}`), &request)
+	if err != nil {
+		return fmt.Errorf("failed to encrypt request for job %q: %w", job.Key, err)
+	}
 	id, err := tx.CreateAsyncCall(ctx, dal.CreateAsyncCallParams{
 		ScheduledAt: nextAttemptForJob,
 		Verb:        schema.RefKey{Module: job.Verb.Module, Name: job.Verb.Name},
 		Origin:      origin.String(),
-		Request:     []byte(`{}`),
+		Request:     request,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to create async call for job %q: %w", job.Key, err)

backend/controller/cronjobs/cronjobs_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/alecthomas/assert/v2"
 	"github.com/alecthomas/types/either"
+	"github.com/alecthomas/types/optional"
 	"github.com/benbjohnson/clock"
 
 	"github.com/TBD54566975/ftl/backend/controller/cronjobs/dal"
@@ -24,7 +25,6 @@ import (
 )
 
 func TestNewCronJobsForModule(t *testing.T) {
-	t.Parallel()
 	ctx := log.ContextWithNewDefaultLogger(context.Background())
 	ctx, cancel := context.WithCancel(ctx)
 	t.Cleanup(cancel)
@@ -36,7 +36,8 @@ func TestNewCronJobsForModule(t *testing.T) {
 	conn := sqltest.OpenForTesting(ctx, t)
 	dal := dal.New(conn)
 
-	encryption, err := encryption.New(ctx, conn, ftlencryption.NewBuilder())
+	uri := "fake-kms://CK6YwYkBElQKSAowdHlwZS5nb29nbGVhcGlzLmNvbS9nb29nbGUuY3J5cHRvLnRpbmsuQWVzR2NtS2V5EhIaEJy4TIQgfCuwxA3ZZgChp_wYARABGK6YwYkBIAE"
+	encryption, err := encryption.New(ctx, conn, ftlencryption.NewBuilder().WithKMSURI(optional.Some(uri)))
 	assert.NoError(t, err)
 
 	parentDAL := parentdal.New(ctx, conn, encryption)
@@ -52,7 +53,7 @@ func TestNewCronJobsForModule(t *testing.T) {
 
 	// Progress so that start_time is valid
 	clk.Add(time.Second)
-	cjs := NewForTesting(ctx, key, "test.com", *dal, clk)
+	cjs := NewForTesting(ctx, key, "test.com", encryption, *dal, clk)
 	// All jobs need to be scheduled
 	expectUnscheduledJobs(t, dal, clk, 2)
 	unscheduledJobs, err := dal.GetUnscheduledCronJobs(ctx, clk.Now())
@@ -70,8 +71,8 @@ func TestNewCronJobsForModule(t *testing.T) {
 	for _, job := range jobsToCreate {
 		j, err := dal.GetCronJobByKey(ctx, job.Key)
 		assert.NoError(t, err)
-		assert.Equal(t, job.StartTime, j.StartTime)
-		assert.Equal(t, j.NextExecution, clk.Now().Add(time.Second))
+		assert.Equal(t, j.StartTime, job.StartTime)
+		assert.Equal(t, clk.Now().Add(time.Second), j.NextExecution)
 
 		p, err := dal.IsCronJobPending(ctx, job.Key, job.StartTime)
 		assert.NoError(t, err)
@@ -82,10 +83,10 @@ func TestNewCronJobsForModule(t *testing.T) {
 	for i, job := range jobsToCreate {
 		call, _, err := parentDAL.AcquireAsyncCall(ctx)
 		assert.NoError(t, err)
-		assert.Equal(t, call.Verb, job.Verb.ToRefKey())
-		assert.Equal(t, call.Origin.String(), fmt.Sprintf("cron:%s", job.Key))
-		assert.Equal(t, call.Request, []byte("{}"))
-		assert.Equal(t, call.QueueDepth, int64(len(jobsToCreate)-i)) // widdling down queue
+		assert.Equal(t, job.Verb.ToRefKey(), call.Verb)
+		assert.Equal(t, fmt.Sprintf("cron:%s", job.Key), call.Origin.String())
+		assert.Equal(t, []byte("{}"), call.Request)
+		assert.Equal(t, int64(len(jobsToCreate)-i), call.QueueDepth) // widdling down queue
 
 		p, err := dal.IsCronJobPending(ctx, job.Key, job.StartTime)
 		assert.NoError(t, err)
@@ -116,10 +117,10 @@ func TestNewCronJobsForModule(t *testing.T) {
 	for i, job := range jobsToCreate {
 		call, _, err := parentDAL.AcquireAsyncCall(ctx)
 		assert.NoError(t, err)
-		assert.Equal(t, call.Verb, job.Verb.ToRefKey())
-		assert.Equal(t, call.Origin.String(), fmt.Sprintf("cron:%s", job.Key))
-		assert.Equal(t, call.Request, []byte("{}"))
-		assert.Equal(t, call.QueueDepth, int64(len(jobsToCreate)-i)) // widdling down queue
+		assert.Equal(t, job.Verb.ToRefKey(), call.Verb)
+		assert.Equal(t, fmt.Sprintf("cron:%s", job.Key), call.Origin.String())
+		assert.Equal(t, []byte("{}"), call.Request)
+		assert.Equal(t, int64(len(jobsToCreate)-i), call.QueueDepth) // widdling down queue
 
 		assert.Equal(t, call.ScheduledAt, clk.Now())
 

backend/controller/dal/fsm.go

@@ -34,7 +34,7 @@ import (
 func (d *DAL) StartFSMTransition(ctx context.Context, fsm schema.RefKey, instanceKey string, destinationState schema.RefKey, request []byte, encrypted bool, retryParams schema.RetryParams) (err error) {
 	var encryptedRequest encryption.EncryptedAsyncColumn
 	if encrypted {
-		encryptedRequest = encryption.EncryptedAsyncColumn(request)
+		encryptedRequest.Set(request)
 	} else {
 		err = d.encryption.Encrypt(request, &encryptedRequest)
 		if err != nil {
@@ -139,9 +139,16 @@ func (d *DAL) PopNextFSMEvent(ctx context.Context, fsm schema.RefKey, instanceKe
 		}
 		return optional.None[NextFSMEvent](), err
 	}
+
+	var decryptedRequest json.RawMessage
+	err = d.encryption.DecryptJSON(&next.Request, &decryptedRequest)
+	if err != nil {
+		return optional.None[NextFSMEvent](), fmt.Errorf("failed to decrypt FSM request: %w", err)
+	}
+
 	return optional.Some(NextFSMEvent{
 		DestinationState: next.NextState,
-		Request:          next.Request,
+		Request:          decryptedRequest,
 		RequestType:      next.RequestType,
 	}), nil
 }

backend/controller/dal/pubsub.go

@@ -98,6 +98,11 @@ func (d *DAL) ProgressSubscriptions(ctx context.Context, eventConsumptionDelay t
 			observability.PubSub.PropagationFailed(ctx, "GetNextEventForSubscription", subscription.Topic.Payload, nextCursor.Caller, subscriptionRef(subscription), optional.None[schema.RefKey]())
 			return 0, fmt.Errorf("failed to get next cursor: %w", libdal.TranslatePGError(err))
 		}
+		payload, ok := nextCursor.Payload.Get()
+		if !ok {
+			observability.PubSub.PropagationFailed(ctx, "GetNextEventForSubscription-->Payload.Get", subscription.Topic.Payload, nextCursor.Caller, subscriptionRef(subscription), optional.None[schema.RefKey]())
+			return 0, fmt.Errorf("could not find payload to progress subscription: %w", libdal.TranslatePGError(err))
+		}
 		nextCursorKey, ok := nextCursor.Event.Get()
 		if !ok {
 			observability.PubSub.PropagationFailed(ctx, "GetNextEventForSubscription-->Event.Get", subscription.Topic.Payload, nextCursor.Caller, subscriptionRef(subscription), optional.None[schema.RefKey]())
@@ -131,7 +136,7 @@ func (d *DAL) ProgressSubscriptions(ctx context.Context, eventConsumptionDelay t
 			ScheduledAt:       time.Now(),
 			Verb:              subscriber.Sink,
 			Origin:            origin.String(),
-			Request:           nextCursor.Payload, // already encrypted
+			Request:           payload, // already encrypted
 			RemainingAttempts: subscriber.RetryAttempts,
 			Backoff:           subscriber.Backoff,
 			MaxBackoff:        subscriber.MaxBackoff,

backend/controller/encryption/dal/dal.go

@@ -83,18 +83,18 @@ func (d *DAL) VerifyEncryptor(ctx context.Context, encryptor encryption.DataEncr
 	if err != nil {
 		return fmt.Errorf("failed to verify timeline subkey: %w", err)
 	}
-	if newTimeline != nil {
+	if newTimeline.Ok() {
 		needsUpdate = true
-		row.VerifyTimeline = optional.Some(newTimeline)
+		row.VerifyTimeline = newTimeline
 	}
 
 	newAsync, err := verifySubkey(encryptor, row.VerifyAsync)
 	if err != nil {
 		return fmt.Errorf("failed to verify async subkey: %w", err)
 	}
-	if newAsync != nil {
+	if newAsync.Ok() {
 		needsUpdate = true
-		row.VerifyAsync = optional.Some(newAsync)
+		row.VerifyAsync = newAsync
 	}
 
 	if !needsUpdate {
@@ -115,25 +115,30 @@ func (d *DAL) VerifyEncryptor(ctx context.Context, encryptor encryption.DataEncr
 
 // verifySubkey checks if the subkey is set and if not, sets it to a verification string.
 // returns (nil, nil) if verified and not changed
-func verifySubkey[SK encryption.SubKey](encryptor encryption.DataEncryptor, encrypted optional.Option[encryption.EncryptedColumn[SK]]) (encryption.EncryptedColumn[SK], error) {
+func verifySubkey[SK encryption.SubKey](
+	encryptor encryption.DataEncryptor,
+	encrypted optional.Option[encryption.EncryptedColumn[SK]],
+) (optional.Option[encryption.EncryptedColumn[SK]], error) {
+	type EC = encryption.EncryptedColumn[SK]
+
 	verifyField, ok := encrypted.Get()
 	if !ok {
 		err := encryptor.Encrypt([]byte(verification), &verifyField)
 		if err != nil {
-			return nil, fmt.Errorf("failed to encrypt verification sanity string: %w", err)
+			return optional.None[EC](), fmt.Errorf("failed to encrypt verification sanity string: %w", err)
 		}
-		return verifyField, nil
+		return optional.Some(verifyField), nil
 	}
 
 	decrypted, err := encryptor.Decrypt(&verifyField)
 	if err != nil {
-		return nil, fmt.Errorf("failed to decrypt verification sanity string: %w", err)
+		return optional.None[EC](), fmt.Errorf("failed to decrypt verification sanity string: %w", err)
 	}
 
 	if string(decrypted) != verification {
-		return nil, fmt.Errorf("decrypted verification string does not match expected value")
+		return optional.None[EC](), fmt.Errorf("decrypted verification string does not match expected value")
 	}
 
 	// verified, no need to update
-	return nil, nil
+	return optional.None[EC](), nil
 }

backend/controller/sql/schema/20240913035022_encrypted_fsm_next_request.sql

@@ -0,0 +1,6 @@
+-- migrate:up
+
+ALTER TABLE fsm_next_event
+    ALTER COLUMN request TYPE encrypted_async;
+
+-- migrate:down

backend/controller/sql/schema/20240913041619_encrypted_topic_events_payload.sql

@@ -0,0 +1,6 @@
+-- migrate:up
+
+ALTER TABLE topic_events
+  ALTER COLUMN payload TYPE encrypted_async;
+
+-- migrate:down