Email migration (#64)

* proto: define OIDCToEmail migration

* rpc: implement OIDC-to-Email migration

* rpc: add and update tests

* etc: enable email migration on dev

config/migrations.go

@@ -2,10 +2,16 @@ package config
 
 type MigrationsConfig struct {
 	OIDCToStytch []OIDCToStytchConfig `toml:"oidc_to_stytch"`
+	Email        EmailMigrationConfig `toml:"oidc_to_email"`
 }
 
 type OIDCToStytchConfig struct {
 	SequenceProject uint64 `toml:"sequence_project"`
 	StytchProject   string `toml:"stytch_project"`
 	FromIssuer      string `toml:"from_issuer"`
 }
+
+type EmailMigrationConfig struct {
+	Enabled      bool   `toml:"enabled"`
+	IssuerPrefix string `toml:"issuer_prefix"`
+}

data/account.go

@@ -223,7 +223,7 @@ func (t *AccountTable) ListByProjectAndIdentity(ctx context.Context, page Page,
 	if identityType != proto.IdentityType_None {
 		identCond = string(identityType) + ":"
 		if issuer != "" {
-			identCond += issuer + "#"
+			identCond += issuer
 		}
 
 		*input.KeyConditionExpression += " and begins_with(#I, :identCond)"

etc/waas-auth.dev.conf

@@ -58,3 +58,8 @@ QwIDAQAB
     sequence_project = 694
     stytch_project = "project-test-c6241c64-de15-412a-a843-09966c98de57"
     from_issuer = "https://oidc-wrapper.sequence.info"
+
+[migrations.oidc_to_email]
+    enabled = true
+    issuer_prefix = "https://cognito-idp.ca-central-1.amazonaws.com/"
+

proto/authenticator.ridl

@@ -200,6 +200,7 @@ struct Page
 
 enum Migration: string
   - OIDCToStytch
+  - OIDCToEmail
 
 
 ##

proto/clients/authenticator.gen.ts

@@ -1,5 +1,5 @@
 /* eslint-disable */
-// sequence-waas-authenticator v0.1.0 2434bf308eeece8d32c65c08f787c7d152d5d199
+// sequence-waas-authenticator v0.1.0 fdf39b8b0bdcc44e72cae2ad1827d0d3d870334d
 // --
 // Code generated by [email protected] with typescript generator. DO NOT EDIT.
 //
@@ -12,7 +12,7 @@ export const WebRPCVersion = "v1"
 export const WebRPCSchemaVersion = "v0.1.0"
 
 // Schema hash generated from your RIDL schema
-export const WebRPCSchemaHash = "2434bf308eeece8d32c65c08f787c7d152d5d199"
+export const WebRPCSchemaHash = "fdf39b8b0bdcc44e72cae2ad1827d0d3d870334d"
 
 //
 // Types
@@ -87,7 +87,8 @@ export interface IntentResponse {
 }
 
 export enum Migration {
-  OIDCToStytch = 'OIDCToStytch'
+  OIDCToStytch = 'OIDCToStytch',
+  OIDCToEmail = 'OIDCToEmail'
 }
 
 export interface Version {

rpc/auth_test.go

@@ -72,6 +72,7 @@ func TestEmailAuth(t *testing.T) {
 			},
 			assertRegisterSessionFn: func(t *testing.T, p assertionParams, sess *proto.Session, res *proto.IntentResponse, err error) {
 				expectedIdentity := newEmailIdentity(fmt.Sprintf("user+%[email protected]", p.tenant.ProjectID))
+				expectedIdentity.Email = expectedIdentity.Subject
 				require.NoError(t, err)
 				assert.Equal(t, expectedIdentity, sess.Identity)
 			},
@@ -106,6 +107,7 @@ func TestEmailAuth(t *testing.T) {
 					return
 				}
 				expectedIdentity := newEmailIdentity(fmt.Sprintf("user+%[email protected]", p.tenant.ProjectID))
+				expectedIdentity.Email = expectedIdentity.Subject
 				require.NoError(t, err)
 				assert.Equal(t, expectedIdentity, sess.Identity)
 			},

rpc/helpers_test.go

@@ -358,7 +358,7 @@ func newTenantWithAuthConfig(t *testing.T, enc *enclave.Enclave, authCfg proto.A
 	}, payload
 }
 
-func newAccount(t *testing.T, tnt *data.Tenant, enc *enclave.Enclave, identity proto.Identity, wallet *ethwallet.Wallet) *data.Account {
+func newAccount(t *testing.T, tnt *data.Tenant, enc *enclave.Enclave, identity proto.Identity, wallet *ethwallet.Wallet, optEmail ...string) *data.Account {
 	att, err := enc.GetAttestation(context.Background(), nil)
 	require.NoError(t, err)
 
@@ -379,6 +379,9 @@ func newAccount(t *testing.T, tnt *data.Tenant, enc *enclave.Enclave, identity p
 	require.NoError(t, err)
 
 	email := "[email protected]"
+	if len(optEmail) > 0 {
+		email = optEmail[0]
+	}
 	return &data.Account{
 		ProjectID:          tnt.ProjectID,
 		Identity:           data.Identity(identity),
@@ -408,7 +411,6 @@ func newEmailIdentity(email string) proto.Identity {
 	return proto.Identity{
 		Type:    proto.IdentityType_Email,
 		Subject: email,
-		Email:   email,
 	}
 }
 

rpc/migration/oidc_to_email.go

@@ -0,0 +1,176 @@
+package migration
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/0xsequence/waas-authenticator/config"
+	"github.com/0xsequence/waas-authenticator/data"
+	"github.com/0xsequence/waas-authenticator/proto"
+	"github.com/0xsequence/waas-authenticator/rpc/attestation"
+	"github.com/0xsequence/waas-authenticator/rpc/auth/email"
+	"github.com/0xsequence/waas-authenticator/rpc/crypto"
+	"github.com/0xsequence/waas-authenticator/rpc/tenant"
+)
+
+type OIDCToEmail struct {
+	accounts *data.AccountTable
+	tenants  *data.TenantTable
+	config   config.EmailMigrationConfig
+}
+
+func (m *OIDCToEmail) OnRegisterSession(ctx context.Context, originalAccount *data.Account) error {
+	att := attestation.FromContext(ctx)
+	tntData := tenant.FromContext(ctx)
+
+	if originalAccount.ProjectID != tntData.ProjectID {
+		return errors.New("project id does not match")
+	}
+	if originalAccount.Identity.Type != proto.IdentityType_OIDC {
+		return nil
+	}
+	if !strings.HasPrefix(originalAccount.Identity.Issuer, m.config.IssuerPrefix) {
+		return nil
+	}
+
+	normEmail := email.Normalize(originalAccount.Email)
+	migratedIdentity := proto.Identity{
+		Type:    proto.IdentityType_Email,
+		Subject: normEmail,
+		Email:   normEmail,
+	}
+
+	_, accountFound, err := m.accounts.Get(ctx, tntData.ProjectID, migratedIdentity)
+	if err != nil {
+		return fmt.Errorf("failed to retrieve account: %w", err)
+	}
+	if accountFound {
+		return nil
+	}
+
+	accData := &proto.AccountData{
+		ProjectID: tntData.ProjectID,
+		UserID:    originalAccount.UserID,
+		Identity:  migratedIdentity.String(),
+		CreatedAt: originalAccount.CreatedAt,
+	}
+	encryptedKey, algorithm, ciphertext, err := crypto.EncryptData(ctx, att, tntData.KMSKeys[0], accData)
+	if err != nil {
+		return fmt.Errorf("encrypting account data: %w", err)
+	}
+
+	account := &data.Account{
+		ProjectID:          tntData.ProjectID,
+		Identity:           data.Identity(migratedIdentity),
+		UserID:             accData.UserID,
+		Email:              migratedIdentity.Email,
+		ProjectScopedEmail: fmt.Sprintf("%d|%s", tntData.ProjectID, migratedIdentity.Email),
+		EncryptedKey:       encryptedKey,
+		Algorithm:          algorithm,
+		Ciphertext:         ciphertext,
+		CreatedAt:          accData.CreatedAt,
+	}
+	if err := m.accounts.Create(ctx, account); err != nil {
+		return fmt.Errorf("saving account: %w", err)
+	}
+	return nil
+}
+
+func (m *OIDCToEmail) NextBatch(ctx context.Context, projectID uint64, page data.Page) ([]string, data.Page, error) {
+	items := make([]string, 0, page.Limit)
+	for {
+		accounts, page, err := m.accounts.ListByProjectAndIdentity(ctx, page, projectID, proto.IdentityType_OIDC, m.config.IssuerPrefix)
+		if err != nil {
+			return nil, page, err
+		}
+
+		for _, acc := range accounts {
+			normEmail := email.Normalize(acc.Email)
+			migratedIdentity := proto.Identity{
+				Type:    proto.IdentityType_Email,
+				Subject: normEmail,
+				Email:   normEmail,
+			}
+			_, found, err := m.accounts.Get(ctx, acc.ProjectID, migratedIdentity)
+			if err != nil {
+				return nil, page, err
+			}
+			if !found {
+				items = append(items, acc.Identity.String())
+			}
+		}
+
+		if len(accounts) < int(page.Limit) || len(items) >= int(page.Limit) {
+			return items, page, nil
+		}
+	}
+}
+
+func (m *OIDCToEmail) ProcessItems(ctx context.Context, tenant *proto.TenantData, items []string) (*Result, error) {
+	if len(items) > 100 {
+		return nil, fmt.Errorf("can only process 100 items at a time")
+	}
+
+	att := attestation.FromContext(ctx)
+	res := NewResult()
+
+	identities := make([]proto.Identity, len(items))
+	for i, item := range items {
+		if err := identities[i].FromString(item); err != nil {
+			res.Errorf(item, "parsing identity: %w", err)
+			continue
+		}
+		if identities[i].Type != proto.IdentityType_OIDC || !strings.HasPrefix(identities[i].Issuer, m.config.IssuerPrefix) {
+			res.Errorf(item, "incorrect identity: %s", identities[i].String())
+			continue
+		}
+	}
+
+	originalAccounts, err := m.accounts.GetBatch(ctx, tenant.ProjectID, identities)
+	if err != nil {
+		return nil, fmt.Errorf("getting accounts: %w", err)
+	}
+
+	for _, originalAccount := range originalAccounts {
+		item := originalAccount.Identity.String()
+		normEmail := email.Normalize(originalAccount.Email)
+		migratedIdentity := proto.Identity{
+			Type:    proto.IdentityType_Email,
+			Subject: normEmail,
+			Email:   normEmail,
+		}
+		accData := &proto.AccountData{
+			ProjectID: tenant.ProjectID,
+			UserID:    originalAccount.UserID,
+			Identity:  migratedIdentity.String(),
+			CreatedAt: originalAccount.CreatedAt,
+		}
+		encryptedKey, algorithm, ciphertext, err := crypto.EncryptData(ctx, att, tenant.KMSKeys[0], accData)
+		if err != nil {
+			res.Errorf(item, "encrypting account data: %w", err)
+			continue
+		}
+
+		account := &data.Account{
+			ProjectID:          tenant.ProjectID,
+			Identity:           data.Identity(migratedIdentity),
+			UserID:             accData.UserID,
+			Email:              migratedIdentity.Email,
+			ProjectScopedEmail: fmt.Sprintf("%d|%s", tenant.ProjectID, migratedIdentity.Email),
+			EncryptedKey:       encryptedKey,
+			Algorithm:          algorithm,
+			Ciphertext:         ciphertext,
+			CreatedAt:          accData.CreatedAt,
+		}
+		if err := m.accounts.Create(ctx, account); err != nil {
+			res.Errorf(item, "saving account: %w", err)
+			continue
+		}
+
+		res.AddItem(item)
+	}
+
+	return res, nil
+}

rpc/migration/oidc_to_stytch.go

@@ -88,7 +88,7 @@ func (m *OIDCToStytch) NextBatch(ctx context.Context, projectID uint64, page dat
 
 	items := make([]string, 0, page.Limit)
 	for {
-		accounts, page, err := m.accounts.ListByProjectAndIdentity(ctx, page, projectID, proto.IdentityType_OIDC, cfg.FromIssuer)
+		accounts, page, err := m.accounts.ListByProjectAndIdentity(ctx, page, projectID, proto.IdentityType_OIDC, cfg.FromIssuer+"#")
 		if err != nil {
 			return nil, page, err
 		}

rpc/migration/runner.go

@@ -33,6 +33,12 @@ func NewRunner(cfg config.MigrationsConfig, accounts *data.AccountTable) *Runner
 		}
 		r.migrations[proto.Migration_OIDCToStytch] = m
 	}
+	if cfg.Email.Enabled {
+		r.migrations[proto.Migration_OIDCToEmail] = &OIDCToEmail{
+			accounts: accounts,
+			config:   cfg.Email,
+		}
+	}
 	return r
 }