From 794cf349da7bad04ebcc8827c9ee674dcf79a4b6 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Tue, 4 Aug 2026 13:46:58 +0200 Subject: [PATCH] WPB-22969: migrate password-reset to PostGreSQL Add Postgres + DualWrite interpreters and a background-worker copy loop for PasswordResetCodeStore, selected at startup by the shared passwordReset StorageLocation. Mirrors the brig DomainRegistrationStore migration and the canonical CodeStore worked example; preserves dual expiry (TTL->expires_at and the timeout column). --- ...PB-22969-migration-postgres-password-reset | 1 + .../background-worker/configmap.yaml | 1 + charts/wire-server/values.yaml | 5 + .../src/developer/reference/config-options.md | 10 + hack/helm_vars/common.yaml.gotmpl | 1 + hack/helm_vars/wire-server/values.yaml.gotmpl | 1 + integration/integration.cabal | 1 + .../test/Test/Migration/PasswordReset.hs | 75 +++++++ libs/wire-api/src/Wire/API/User/Password.hs | 15 ++ .../Wire/API/Roundtrip/PostgresMarshall.hs | 3 + .../20260804090521-password-reset.sql | 8 + .../Wire/PasswordResetCodeStore/DualWrite.hs | 52 +++++ .../Wire/PasswordResetCodeStore/Migration.hs | 183 ++++++++++++++++++ .../Wire/PasswordResetCodeStore/Postgres.hs | 120 ++++++++++++ .../src/Wire/PostgresMigrationOpts.hs | 4 +- libs/wire-subsystems/wire-subsystems.cabal | 3 + postgres-schema.sql | 28 ++- .../background-worker.integration.yaml | 2 + .../src/Wire/BackgroundWorker.hs | 10 +- .../src/Wire/BackgroundWorker/Options.hs | 1 + .../src/Wire/PostgresMigrations.hs | 19 ++ .../Wire/BackendNotificationPusherSpec.hs | 6 +- .../background-worker/test/Test/Wire/Util.hs | 3 +- services/brig/brig.integration.yaml | 1 + .../brig/src/Brig/CanonicalInterpreter.hs | 24 ++- 25 files changed, 569 insertions(+), 8 deletions(-) create mode 100644 changelog.d/5-internal/gdifolco-WPB-22969-migration-postgres-password-reset create mode 100644 integration/test/Test/Migration/PasswordReset.hs create mode 100644 libs/wire-subsystems/postgres-migrations/20260804090521-password-reset.sql create mode 100644 libs/wire-subsystems/src/Wire/PasswordResetCodeStore/DualWrite.hs create mode 100644 libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Migration.hs create mode 100644 libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Postgres.hs diff --git a/changelog.d/5-internal/gdifolco-WPB-22969-migration-postgres-password-reset b/changelog.d/5-internal/gdifolco-WPB-22969-migration-postgres-password-reset new file mode 100644 index 00000000000..eeae983ad2c --- /dev/null +++ b/changelog.d/5-internal/gdifolco-WPB-22969-migration-postgres-password-reset @@ -0,0 +1 @@ +Migration of password reset from cassandra to postgres diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index d4fe2a63202..637f0fc5f61 100644 --- a/charts/wire-server/templates/background-worker/configmap.yaml +++ b/charts/wire-server/templates/background-worker/configmap.yaml @@ -84,6 +84,7 @@ data: migrateConversationCodes: {{ .migrateConversationCodes }} migrateTeamFeatures: {{ .migrateTeamFeatures }} migrateDomainRegistration: {{ .migrateDomainRegistration }} + migratePasswordReset: {{ .migratePasswordReset }} migrationOptions: {{ toYaml .migrationOptions | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index ae71d850795..83815cb1188 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -90,6 +90,7 @@ galley: teamFeatures: cassandra domainRegistration: cassandra user: cassandra + passwordReset: cassandra settings: httpPoolSize: 128 maxTeamSize: 10000 @@ -1017,6 +1018,10 @@ background-worker: # It's important to set `settings.postgresMigration.domainRegistration` to `migration-to-postgresql` # before starting the migration. migrateDomainRegistration: false + # This will start the migration of password reset data. + # It's important to set `settings.postgresMigration.passwordReset` to `migration-to-postgresql` + # before starting the migration. + migratePasswordReset: false backendNotificationPusher: pushBackoffMinWait: 10000 # in microseconds, so 10ms diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md index 066637dbb07..03607e3b6e7 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -2106,12 +2106,14 @@ galley: teamFeatures: postgresql domainRegistration: postgresql user: postgresql + passwordReset: postgresql background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migratePasswordReset: false ``` #### Migration for existing installations @@ -2142,6 +2144,7 @@ The current settings and their background-worker flags are: - `conversationCodes` -> `migrateConversationCodes` - `teamFeatures` -> `migrateTeamFeatures` - `domainRegistration` -> `migrateDomainRegistration` +- `passwordReset` -> `migratePasswordReset` **Migration pattern per migration setting** @@ -2161,12 +2164,14 @@ The current settings and their background-worker flags are: conversationCodes: migration-to-postgresql teamFeatures: migration-to-postgresql domainRegistration: cassandra + passwordReset: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migratePasswordReset: false ``` This change should restart the affected pods, and new writes will follow the @@ -2181,6 +2186,7 @@ The current settings and their background-worker flags are: migrateConversationCodes: true migrateTeamFeatures: true migrateDomainRegistration: true + migratePasswordReset: true ``` During migration, Cassandra rows are not deleted. Writes and migration share @@ -2196,6 +2202,7 @@ The current settings and their background-worker flags are: - `conversationCodes`: `wire_conv_codes_migration_finished` - `teamFeatures`: `wire_team_features_migration_finished` - `domainRegistration`: `wire_domain_registration_migration_finished` + - `passwordReset`: `wire_password_reset_migration_finished` 3. Cut over reads and writes to PostgreSQL for the selected migration setting(s). This configuration must be used from now on for every new @@ -2209,12 +2216,14 @@ The current settings and their background-worker flags are: conversationCodes: postgresql teamFeatures: postgresql domainRegistration: cassandra + passwordReset: cassandra background-worker: config: migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false + migratePasswordReset: false ``` **How to run migrations independently or in batches** @@ -2303,6 +2312,7 @@ migrateConversations: false migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false +migratePasswordReset: false # migration settings migrationOptions: diff --git a/hack/helm_vars/common.yaml.gotmpl b/hack/helm_vars/common.yaml.gotmpl index 2276355e2a9..c96984cfeb1 100644 --- a/hack/helm_vars/common.yaml.gotmpl +++ b/hack/helm_vars/common.yaml.gotmpl @@ -18,6 +18,7 @@ conversationStore: {{ $preferredStore }} conversationCodesStore: {{ $preferredStore }} teamFeaturesStore: {{ $preferredStore }} domainRegistration: {{ $preferredStore }} +passwordReset: {{ $preferredStore }} userStore: {{ $preferredStore }} {{- if (eq (env "UPLOAD_XML_S3_BASE_URL") "") }} diff --git a/hack/helm_vars/wire-server/values.yaml.gotmpl b/hack/helm_vars/wire-server/values.yaml.gotmpl index bd6cdadf2a8..2c1a914127b 100644 --- a/hack/helm_vars/wire-server/values.yaml.gotmpl +++ b/hack/helm_vars/wire-server/values.yaml.gotmpl @@ -305,6 +305,7 @@ galley: conversationCodes: {{ .Values.conversationCodesStore }} teamFeatures: {{ .Values.teamFeaturesStore }} domainRegistration: {{ .Values.domainRegistration }} + passwordReset: {{ .Values.passwordReset }} user: {{ .Values.userStore }} settings: maxConvAndTeamSize: 16 diff --git a/integration/integration.cabal b/integration/integration.cabal index c36b9c4e730..c6184af0980 100644 --- a/integration/integration.cabal +++ b/integration/integration.cabal @@ -180,6 +180,7 @@ library Test.Migration.Conversation Test.Migration.ConversationCodes Test.Migration.DomainRegistration + Test.Migration.PasswordReset Test.Migration.TeamFeatures Test.Migration.Util Test.MLS diff --git a/integration/test/Test/Migration/PasswordReset.hs b/integration/test/Test/Migration/PasswordReset.hs new file mode 100644 index 00000000000..522d05d83c3 --- /dev/null +++ b/integration/test/Test/Migration/PasswordReset.hs @@ -0,0 +1,75 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Test.Migration.PasswordReset (testPasswordResetMigration) where + +import API.Brig +import API.BrigInternal (getPasswordResetCode) +import Control.Monad.Codensity +import Control.Monad.Reader +import SetupHelpers +import Test.Migration.Util (waitForMigration) +import Testlib.Prelude +import Testlib.ResourcePool + +-- | Drives the password-reset store through the full cutover lifecycle +-- (cassandra -> migration-to-postgresql -> postgresql). A reset code written to +-- Cassandra before migration must be served from Postgres after the cutover and +-- still complete the reset, proving the row was backfilled. +testPasswordResetMigration :: (HasCallStack) => App () +testPasswordResetMigration = do + resourcePool <- asks (.resourcePool) + runCodensity (acquireResources 1 resourcePool) $ \[backend] -> do + let domain = backend.berDomain + + -- Cassandra: create a user and initiate a password reset (writes the code to Cassandra) + (email, key, code) <- + runCodensity (startDynamicBackend backend (conf "cassandra" False)) $ \_ -> do + user <- randomUser domain def + email <- user %. "email" & asString + passwordReset domain email >>= assertSuccess + getResetData domain email + + -- migration-to-postgresql (worker off): the code is still served from Cassandra + runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" False)) $ \_ -> + checkCode domain email key code + + -- migration-to-postgresql (worker on): backfill the code to Postgres and wait for completion + runCodensity (startDynamicBackend backend (conf "migration-to-postgresql" True)) $ \_ -> + waitForMigration domain counterName + + -- postgresql: the migrated code is served from Postgres and completes the reset + runCodensity (startDynamicBackend backend (conf "postgresql" False)) $ \_ -> do + checkCode domain email key code + let newPassword = "shiny-new-password" + completePasswordReset domain key code newPassword >>= assertSuccess + login domain email newPassword >>= assertSuccess + where + conf db runMigration = + def + { galleyCfg = setField "postgresMigration.passwordReset" db, + backgroundWorkerCfg = setField "migratePasswordReset" runMigration + } + counterName = "^wire_password_reset_migration_finished" + getResetData dom email = + bindResponse (getPasswordResetCode dom email) $ \resp -> do + resp.status `shouldMatchInt` 200 + (,) <$> (resp.json %. "key" & asString) <*> (resp.json %. "code" & asString) + checkCode dom email key code = do + (key', code') <- getResetData dom email + key' `shouldMatch` key + code' `shouldMatch` code diff --git a/libs/wire-api/src/Wire/API/User/Password.hs b/libs/wire-api/src/Wire/API/User/Password.hs index 079ccde88fa..b18f8ec2293 100644 --- a/libs/wire-api/src/Wire/API/User/Password.hs +++ b/libs/wire-api/src/Wire/API/User/Password.hs @@ -48,12 +48,15 @@ import Data.Proxy (Proxy (Proxy)) import Data.Range (Ranged (..)) import Data.Schema as Schema import Data.Text.Ascii +import Data.Text qualified as Text +import Data.Text.Encoding qualified as Text import Data.Tuple.Extra import Imports import Servant (FromHttpApiData (..)) import Wire.API.User.EmailAddress import Wire.API.User.Phone import Wire.Arbitrary (Arbitrary, GenericUniform (..)) +import Wire.API.PostgresMarshall (PostgresMarshall (postgresMarshall), PostgresUnmarshall (postgresUnmarshall)) -------------------------------------------------------------------------------- -- NewPasswordReset @@ -206,6 +209,18 @@ newtype PasswordResetCode = PasswordResetCode deriving instance C.Cql PasswordResetCode +instance PostgresMarshall Text PasswordResetKey where + postgresMarshall = Text.decodeUtf8 . toByteString' + +instance PostgresMarshall Text PasswordResetCode where + postgresMarshall = Text.decodeUtf8 . toByteString' + +instance PostgresUnmarshall Text PasswordResetKey where + postgresUnmarshall = mapLeft Text.pack . runParser parser . Text.encodeUtf8 + +instance PostgresUnmarshall Text PasswordResetCode where + postgresUnmarshall = mapLeft Text.pack . runParser parser . Text.encodeUtf8 + -------------------------------------------------------------------------------- -- DEPRECATED diff --git a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs index 191f55bffd0..27d5a1e5884 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/PostgresMarshall.hs @@ -35,12 +35,15 @@ import Wire.API.Password.Scrypt (encodeScryptPassword) import Wire.API.PostgresMarshall import Wire.API.Team.Feature import Wire.Arbitrary qualified as Arbitrary () +import Wire.API.User.Password (PasswordResetCode, PasswordResetKey) tests :: T.TestTree tests = T.localOption (T.Timeout (60 * 1000000) "60s") . T.testGroup "PostgresMarshall roundtrip tests" $ [ testRoundTrip @Text @Code.Key, testRoundTrip @Text @Code.Value, + testRoundTrip @Text @PasswordResetKey, + testRoundTrip @Text @PasswordResetCode, testRoundTrip @ByteString @Password.Password, testRoundTrip @Int32 @FeatureStatus, testRoundTrip @Int32 @LockStatus, diff --git a/libs/wire-subsystems/postgres-migrations/20260804090521-password-reset.sql b/libs/wire-subsystems/postgres-migrations/20260804090521-password-reset.sql new file mode 100644 index 00000000000..3884708a59f --- /dev/null +++ b/libs/wire-subsystems/postgres-migrations/20260804090521-password-reset.sql @@ -0,0 +1,8 @@ +CREATE TABLE IF NOT EXISTS password_reset ( + key text PRIMARY KEY, + code text NOT NULL, + "user" uuid NOT NULL, + retries int4, + timeout timestamptz, + expires_at timestamptz NOT NULL +); diff --git a/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/DualWrite.hs b/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/DualWrite.hs new file mode 100644 index 00000000000..8fe1041c884 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/DualWrite.hs @@ -0,0 +1,52 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.PasswordResetCodeStore.DualWrite + ( interpretPasswordResetCodeStoreToCassandraAndPostgres, + ) +where + +import Cassandra (MonadClient) +import Imports +import Polysemy +import Wire.PasswordResetCodeStore +import Wire.PasswordResetCodeStore.Cassandra (passwordResetCodeStoreToCassandra) +import Wire.PasswordResetCodeStore.Postgres qualified as Postgres +import Wire.Postgres (PGConstraints) + +-- | During migration, Cassandra remains the source of truth for reads while +-- writes are mirrored to Postgres. The generation constructors are pure (they +-- only draw randomness); they run exactly once — invoking both interpreters +-- would produce two independent random codes. +interpretPasswordResetCodeStoreToCassandraAndPostgres :: + forall m r. + ( MonadClient m, + Member (Embed m) r, + PGConstraints r + ) => + InterpreterFor PasswordResetCodeStore r +interpretPasswordResetCodeStoreToCassandraAndPostgres = interpret $ \case + GenerateEmailCode -> Postgres.genEmailCode + GeneratePhoneCode -> Postgres.genPhoneCode + CodeSelect prk -> + passwordResetCodeStoreToCassandra @m $ codeSelect prk + CodeInsert prk prqd ttl -> do + passwordResetCodeStoreToCassandra @m $ codeInsert prk prqd ttl + Postgres.interpretPasswordResetCodeStoreToPostgres $ codeInsert prk prqd ttl + CodeDelete prk -> do + passwordResetCodeStoreToCassandra @m $ codeDelete prk + Postgres.interpretPasswordResetCodeStoreToPostgres $ codeDelete prk diff --git a/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Migration.hs b/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Migration.hs new file mode 100644 index 00000000000..6e832a53848 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Migration.hs @@ -0,0 +1,183 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.PasswordResetCodeStore.Migration + ( migratePasswordResetLoop, + ) +where + +import Cassandra +import Data.ByteString.Conversion +import Data.Conduit +import Data.Conduit.List qualified as C +import Data.Id (UserId) +import Data.IORef qualified as IORef +import Data.Text qualified as T +import Data.Time +import Hasql.Pool.Extended qualified as Hasql +import Imports +import Polysemy +import Polysemy.Async +import Polysemy.Conc (interpretRace) +import Polysemy.Conc qualified as Conc +import Polysemy.Conc.Effect.Race hiding (Timeout) +import Polysemy.Input +import Polysemy.Resource (Resource, bracket, resourceToIOFinal) +import Polysemy.State +import Polysemy.TinyLog +import Prometheus qualified +import System.Logger qualified as Log +import UnliftIO qualified +import Wire.API.User.Password +import Wire.Migration +import Wire.PasswordResetCodeStore +import Wire.PasswordResetCodeStore.Postgres qualified as Postgres +import Wire.Postgres +import Wire.Sem.Logger (mapLogger) +import Wire.Sem.Logger.TinyLog (loggerToTinyLog) + +type EffectStack = + [ State Int, + Input ClientState, + Input Hasql.Pool, + Resource, + Async, + Race, + TinyLog, + Embed IO, + Final IO + ] + +migratePasswordResetLoop :: + MigrationOptions -> + ClientState -> + Hasql.Pool -> + Log.Logger -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + IO () +migratePasswordResetLoop migOpts cassClient pgPool logger migCounter migFinished migFailed migDuration = + migrationLoop + logger + "password reset" + migFinished + migFailed + (interpreter cassClient pgPool logger "password reset") + (migrateAllPasswordReset migOpts migCounter migDuration) + +interpreter :: ClientState -> Hasql.Pool -> Log.Logger -> ByteString -> Sem EffectStack a -> IO (Int, a) +interpreter cassClient pgPool logger name = + runFinal + . embedToFinal + . loggerToTinyLog logger + . mapLogger (Log.field "migration" (Log.val name) .) + . raiseUnder + . interpretRace + . asyncToIOFinal + . resourceToIOFinal + . runInputConst pgPool + . runInputConst cassClient + . runState 0 + +migrateAllPasswordReset :: + ( Member (Input Hasql.Pool) r, + Member (Input ClientState) r, + Member (Embed IO) r, + Member TinyLog r, + Member (State Int) r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + ConduitM () Void (Sem r) () +migrateAllPasswordReset migOpts migCounter migDuration = do + lift $ info $ Log.msg (Log.val "migrateAllPasswordReset") + withCount (paginateSem selectAllPasswordReset (paramsP LocalQuorum () migOpts.pageSize) x5) + .| logRetrievedPage migOpts.pageSize id + .| C.mapM_ (traverse_ (\row@(key, _, _, _, _, _) -> handleErrors (toByteString' key) (migratePasswordResetRow migOpts migCounter migDuration row))) + +migratePasswordResetRow :: + ( PGConstraints r, + Member TinyLog r, + Member Resource r, + Member Race r + ) => + MigrationOptions -> + Prometheus.Counter -> + Prometheus.Vector Text Prometheus.Histogram -> + (PasswordResetKey, PasswordResetCode, UserId, Maybe Int32, Maybe UTCTime, Int32) -> + Sem r () +migratePasswordResetRow migOpts migCounter migDuration (key, code, uid, mRetries, mTimeout, ttl) = + when (ttl > 0) $ case (mRetries, mTimeout) of + (Just retries, Just codeTimeout) -> migrateRow retries codeTimeout + -- retries/timeout are always present on rows written by CodeInsert (Identity), so a + -- null here means a malformed/legacy row: skip it (logged) rather than block the + -- migration. Its reset code is not backfilled, so the affected user re-initiates the + -- reset after cutover. + _ -> + warn $ + Log.msg (Log.val "password reset row with null retries/timeout, skipping") + . Log.field "key" (show key) + where + migrateRow retries codeTimeout = do + let prqd = + PRQueryData + { prqdCode = code, + prqdUser = uid, + prqdRetries = Identity retries, + prqdTimeout = Identity codeTimeout + } + keyText = T.pack (show key) + outcomeRef <- liftIO $ IORef.newIORef @Text "error" + bracket + (liftIO getCurrentTime) + (observeDuration migDuration outcomeRef) + ( const $ do + timeoutResult <- + Conc.timeout (migOpts.timeout <$ handleTimeout) migOpts.timeout $ + Postgres.interpretPasswordResetCodeStoreToPostgres $ + codeInsert key prqd ttl + case timeoutResult of + Left timedOutAfter -> do + markOutcome outcomeRef "timeout" + liftIO . UnliftIO.throwIO $ MigrationTimedOut keyText timedOutAfter + Right () -> do + markOutcome outcomeRef "success" + liftIO $ Prometheus.incCounter migCounter + ) + handleTimeout = + err $ + Log.msg (Log.val "password reset code migration timed out") + . Log.field "key" (show key) + . Log.field "timeout" (show migOpts.timeout) + markOutcome ref outcome = liftIO $ IORef.writeIORef ref outcome + observeDuration metric outcomeRef start = do + outcome <- liftIO $ IORef.readIORef outcomeRef + end <- liftIO getCurrentTime + liftIO $ Prometheus.withLabel metric outcome (`Prometheus.observe` realToFrac (diffUTCTime end start)) + +selectAllPasswordReset :: + PrepQuery + R + () + (PasswordResetKey, PasswordResetCode, UserId, Maybe Int32, Maybe UTCTime, Int32) +selectAllPasswordReset = + "SELECT key, code, user, retries, timeout, ttl(code) FROM password_reset" diff --git a/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Postgres.hs b/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Postgres.hs new file mode 100644 index 00000000000..09200a93ea6 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/PasswordResetCodeStore/Postgres.hs @@ -0,0 +1,120 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . +{-# LANGUAGE RecordWildCards #-} + +module Wire.PasswordResetCodeStore.Postgres + ( interpretPasswordResetCodeStoreToPostgres, + genEmailCode, + genPhoneCode, + ) +where + +import Data.Id (UserId) +import Data.Text (pack) +import Data.Text.Ascii (encodeBase64Url, unsafeFromText) +import Data.Time.Clock (UTCTime) +import Hasql.Statement qualified as Hasql +import Hasql.TH +import Imports +import OpenSSL.BN (randIntegerZeroToNMinusOne) +import OpenSSL.Random (randBytes) +import Polysemy +import Text.Printf (printf) +import Wire.API.PostgresMarshall +import Wire.API.User.Password +import Wire.PasswordResetCodeStore (PasswordResetCodeStore (..), PRQueryData (..)) +import Wire.Postgres + +interpretPasswordResetCodeStoreToPostgres :: + (PGConstraints r) => + InterpreterFor PasswordResetCodeStore r +interpretPasswordResetCodeStoreToPostgres = interpret $ \case + GenerateEmailCode -> genEmailCode + GeneratePhoneCode -> genPhoneCode + CodeSelect prk -> codeSelect prk + CodeInsert prk prqd ttl -> codeInsert prk prqd ttl + CodeDelete prk -> codeDelete prk + +-- | 24 random bytes, base64url-encoded (mirrors the Cassandra interpreter). +genEmailCode :: (Member (Embed IO) r) => Sem r PasswordResetCode +genEmailCode = PasswordResetCode . encodeBase64Url <$> embed @IO (randBytes 24) + +-- | A 6-digit, zero-padded code (mirrors the Cassandra interpreter). +genPhoneCode :: (Member (Embed IO) r) => Sem r PasswordResetCode +genPhoneCode = + PasswordResetCode . unsafeFromText . pack . printf "%06d" + <$> embed @IO (randIntegerZeroToNMinusOne 1000000) + +codeSelect :: + (PGConstraints r) => + PasswordResetKey -> + Sem r (Maybe (PRQueryData Maybe)) +codeSelect prk = do + mRow <- runStatement prk select + pure $ fmap toRecord mRow + where + toRecord :: + (PasswordResetCode, UserId, Maybe Int32, Maybe UTCTime) -> + PRQueryData Maybe + toRecord (prqdCode, prqdUser, prqdRetries, prqdTimeout) = + PRQueryData {..} + select :: + Hasql.Statement PasswordResetKey (Maybe (PasswordResetCode, UserId, Maybe Int32, Maybe UTCTime)) + select = + dimapPG + [maybeStatement|SELECT (code :: text), ("user" :: uuid), (retries :: int4?), (timeout :: timestamptz?) + FROM password_reset + WHERE key = ($1 :: text) AND expires_at > now () + |] + +codeInsert :: + (PGConstraints r) => + PasswordResetKey -> + PRQueryData Identity -> + Int32 -> + Sem r () +codeInsert prk (PRQueryData prc uid n ut) ttl = + runStatement (prk, prc, uid, Just (runIdentity n), Just (runIdentity ut), ttl) insert + where + insert :: + Hasql.Statement (PasswordResetKey, PasswordResetCode, UserId, Maybe Int32, Maybe UTCTime, Int32) () + insert = + lmapPG + [resultlessStatement|INSERT INTO password_reset (key, code, "user", retries, timeout, expires_at) + VALUES + ($1 :: text, $2 :: text, $3 :: uuid, $4 :: int4?, $5 :: timestamptz?, now() + make_interval(secs => $6 :: int)) + ON CONFLICT (key) DO UPDATE + SET code = ($2 :: text), + "user" = ($3 :: uuid), + retries = ($4 :: int4?), + timeout = ($5 :: timestamptz?), + expires_at = now() + make_interval(secs => $6 :: int) + |] + +codeDelete :: + (PGConstraints r) => + PasswordResetKey -> + Sem r () +codeDelete prk = + runStatement prk delete + where + delete :: Hasql.Statement PasswordResetKey () + delete = + lmapPG + [resultlessStatement|DELETE FROM password_reset + WHERE key = ($1 :: text) + |] diff --git a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs index 327862f7cd5..1036843e927 100644 --- a/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs +++ b/libs/wire-subsystems/src/Wire/PostgresMigrationOpts.hs @@ -56,7 +56,8 @@ data PostgresMigrationOpts = PostgresMigrationOpts conversationCodes :: StorageLocation, teamFeatures :: StorageLocation, domainRegistration :: StorageLocation, - user :: StorageLocation + user :: StorageLocation, + passwordReset :: StorageLocation } deriving (Show) @@ -68,3 +69,4 @@ instance FromJSON PostgresMigrationOpts where <*> o .: "teamFeatures" <*> o .: "domainRegistration" <*> o .: "user" + <*> o .: "passwordReset" diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 943f8fa0294..529e875e18a 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -414,6 +414,9 @@ library Wire.ParseException Wire.PasswordResetCodeStore Wire.PasswordResetCodeStore.Cassandra + Wire.PasswordResetCodeStore.DualWrite + Wire.PasswordResetCodeStore.Migration + Wire.PasswordResetCodeStore.Postgres Wire.PasswordStore Wire.PasswordStore.Cassandra Wire.Postgres diff --git a/postgres-schema.sql b/postgres-schema.sql index 948f1d62594..32796b6be15 100644 --- a/postgres-schema.sql +++ b/postgres-schema.sql @@ -9,8 +9,8 @@ \restrict 79bbfb4630959c48307653a5cd3d83f2582b3c2210f75f10d79e3ebf0015620 --- Dumped from database version 17.9 --- Dumped by pg_dump version 17.9 +-- Dumped from database version 17.10 +-- Dumped by pg_dump version 17.10 SET statement_timeout = 0; SET lock_timeout = 0; @@ -342,6 +342,22 @@ CREATE TABLE public.mls_history_client ( ALTER TABLE public.mls_history_client OWNER TO "wire-server"; +-- +-- Name: password_reset; Type: TABLE; Schema: public; Owner: wire-server +-- + +CREATE TABLE public.password_reset ( + key text NOT NULL, + code text NOT NULL, + "user" uuid NOT NULL, + retries integer, + timeout timestamp with time zone, + expires_at timestamp with time zone NOT NULL +); + + +ALTER TABLE public.password_reset OWNER TO "wire-server"; + -- -- Name: remote_conversation_local_member; Type: TABLE; Schema: public; Owner: wire-server -- @@ -601,6 +617,14 @@ ALTER TABLE ONLY public.mls_history_client ADD CONSTRAINT mls_history_client_pkey PRIMARY KEY (group_id, id); +-- +-- Name: password_reset password_reset_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server +-- + +ALTER TABLE ONLY public.password_reset + ADD CONSTRAINT password_reset_pkey PRIMARY KEY (key); + + -- -- Name: remote_conversation_local_member remote_conversation_local_member_pkey; Type: CONSTRAINT; Schema: public; Owner: wire-server -- diff --git a/services/background-worker/background-worker.integration.yaml b/services/background-worker/background-worker.integration.yaml index e264ce14016..f4cd9da2922 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -58,6 +58,7 @@ migrationOptions: migrateConversationCodes: false migrateTeamFeatures: false migrateDomainRegistration: false +migratePasswordReset: false # Background jobs consumer configuration for integration backgroundJobs: @@ -91,4 +92,5 @@ postgresMigration: conversationCodes: postgresql teamFeatures: postgresql domainRegistration: postgresql + passwordReset: postgresql user: postgresql diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index b57ba12df40..f7954b27782 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -78,6 +78,13 @@ run opts galleyOpts = do withNamedLogger "migrate-domain-registration" $ Migrations.domainRegistration opts.migrationOptions else pure $ pure () + cleanupPasswordResetMigration <- + if opts.migratePasswordReset + then + runAppT env $ + withNamedLogger "migrate-password-reset" $ + Migrations.passwordReset opts.migrationOptions + else pure $ pure () cleanupJobs <- runAppT env $ withNamedLogger "background-job-consumer" $ @@ -89,13 +96,14 @@ run opts galleyOpts = do let cleanup = void $ runConcurrently $ - (,,,,,,,) + (,,,,,,,,) <$> Concurrently cleanupDeadUserNotifWatcher <*> Concurrently cleanupBackendNotifPusher <*> Concurrently cleanupConvMigration <*> Concurrently cleanUpConvCodesMigration <*> Concurrently cleanupTeamFeaturesMigration <*> Concurrently cleanupDomainRegistrationMigration + <*> Concurrently cleanupPasswordResetMigration <*> Concurrently cleanupJobRunner <*> Concurrently cleanupJobs diff --git a/services/background-worker/src/Wire/BackgroundWorker/Options.hs b/services/background-worker/src/Wire/BackgroundWorker/Options.hs index 61df5d5d14f..da7e245c938 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs @@ -55,6 +55,7 @@ data Opts = Opts migrateConversationCodes :: !Bool, migrateTeamFeatures :: !Bool, migrateDomainRegistration :: !Bool, + migratePasswordReset :: !Bool, jobs :: JobConfig, meetingsCleanup :: MeetingsCleanupConfig, backgroundJobs :: BackgroundJobsConfig diff --git a/services/background-worker/src/Wire/PostgresMigrations.hs b/services/background-worker/src/Wire/PostgresMigrations.hs index 604cab0140c..74a7ecd54b4 100644 --- a/services/background-worker/src/Wire/PostgresMigrations.hs +++ b/services/background-worker/src/Wire/PostgresMigrations.hs @@ -26,6 +26,7 @@ import Wire.BackgroundWorker.Util import Wire.CodeStore.Migration import Wire.ConversationStore.Migration import Wire.DomainRegistrationStore.Migration +import Wire.PasswordResetCodeStore.Migration import Wire.Migration (MigrationOptions) import Wire.TeamFeatureStore.Migration @@ -107,3 +108,21 @@ domainRegistration migOpts = do pure $ do Log.info logger $ Log.msg (Log.val "cancelling domain registration migration") cancel migrationLoop + +passwordReset :: MigrationOptions -> AppT IO CleanupAction +passwordReset migOpts = do + cassClient <- asks (.cassandraBrig) + pgPool <- asks (.hasqlPool) + logger <- asks (.logger) + Log.info logger $ Log.msg (Log.val "starting password reset migration") + count <- register $ counter $ Prometheus.Info "wire_password_reset_migrated_to_pg" "Number of password reset keys migrated to Postgresql" + finished <- register $ counter $ Prometheus.Info "wire_password_reset_migration_finished" "Whether the password reset migration to Postgresql is finished successfully" + failed <- register $ counter $ Prometheus.Info "wire_password_reset_migration_failed" "Whether the password reset migration to Postgresql has failed" + duration <- register $ vector "outcome" $ histogram (Prometheus.Info "wire_password_reset_migration_duration_seconds" "Duration of password reset migration attempts") defaultBuckets + + migrationLoop <- async . lift $ migratePasswordResetLoop migOpts cassClient pgPool logger count finished failed duration + + Log.info logger $ Log.msg (Log.val "started password reset migration") + pure $ do + Log.info logger $ Log.msg (Log.val "cancelling password reset migration") + cancel migrationLoop diff --git a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs index 7222120d93a..547406b26b9 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -371,7 +371,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + passwordReset = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined @@ -435,7 +436,8 @@ spec = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + passwordReset = CassandraStorage } gundeckEndpoint = undefined brigEndpoint = undefined diff --git a/services/background-worker/test/Test/Wire/Util.hs b/services/background-worker/test/Test/Wire/Util.hs index 5d89532bfec..aac06443512 100644 --- a/services/background-worker/test/Test/Wire/Util.hs +++ b/services/background-worker/test/Test/Wire/Util.hs @@ -50,7 +50,8 @@ testEnv = do conversationCodes = CassandraStorage, teamFeatures = CassandraStorage, domainRegistration = CassandraStorage, - user = CassandraStorage + user = CassandraStorage, + passwordReset = CassandraStorage } statuses <- newIORef mempty backendNotificationMetrics <- mkBackendNotificationMetrics diff --git a/services/brig/brig.integration.yaml b/services/brig/brig.integration.yaml index 8be11f028bd..392a8bf86ff 100644 --- a/services/brig/brig.integration.yaml +++ b/services/brig/brig.integration.yaml @@ -175,6 +175,7 @@ postgresMigration: conversationCodes: postgresql teamFeatures: postgresql domainRegistration: postgresql + passwordReset: postgresql user: postgresql optSettings: diff --git a/services/brig/src/Brig/CanonicalInterpreter.hs b/services/brig/src/Brig/CanonicalInterpreter.hs index a39da9d12a3..c75b52e8939 100644 --- a/services/brig/src/Brig/CanonicalInterpreter.hs +++ b/services/brig/src/Brig/CanonicalInterpreter.hs @@ -119,6 +119,8 @@ import Wire.NotificationSubsystem.Interpreter (defaultNotificationSubsystemConfi import Wire.ParseException import Wire.PasswordResetCodeStore (PasswordResetCodeStore) import Wire.PasswordResetCodeStore.Cassandra (interpretClientToIO, passwordResetCodeStoreToCassandra) +import Wire.PasswordResetCodeStore.DualWrite (interpretPasswordResetCodeStoreToCassandraAndPostgres) +import Wire.PasswordResetCodeStore.Postgres (interpretPasswordResetCodeStoreToPostgres) import Wire.PasswordStore (PasswordStore) import Wire.PasswordStore.Cassandra (interpretPasswordStore) import Wire.PostgresMigrationOpts @@ -420,6 +422,26 @@ runBrigToIO e (AppT ma) = do PostgresqlStorage -> interpretUserStorePostgres MigrationToPostgresql -> error "Migration not implemented for user" + -- PasswordResetCodeStore must be peeled at this slot in the stack (between + -- interpretGalleyAPIAccessToRpc and randomToIO), which precedes the global + -- runInputConst e.hasqlPool / mapError postgresUsageErrorToHttpError providers. + -- The Postgres-backed branches therefore supply Input Pool and Error UsageError + -- locally (raiseUnder2 + runInputConst + mapError) so they stay self-contained; + -- UsageError is mapped to HttpError (handled by rethrowHttpErrorIO further up). + passwordResetCodeStoreInterpreter = + case e.postgresMigration.passwordReset of + CassandraStorage -> passwordResetCodeStoreToCassandra @Cas.Client + PostgresqlStorage -> + runInputConst e.hasqlPool + . mapError postgresUsageErrorToHttpError + . interpretPasswordResetCodeStoreToPostgres + . raiseUnder2 + MigrationToPostgresql -> + runInputConst e.hasqlPool + . mapError postgresUsageErrorToHttpError + . interpretPasswordResetCodeStoreToCassandraAndPostgres @Cas.Client + . raiseUnder2 + ( either throwM pure <=< ( runFinal . unsafelyPerformConcurrency @@ -440,7 +462,7 @@ runBrigToIO e (AppT ma) = do . emailSendingInterpreter e . interpretSparAPIAccessToRpc e.sparEndpoint . interpretGalleyAPIAccessToRpc e.disabledVersions e.galleyEndpoint - . passwordResetCodeStoreToCassandra @Cas.Client + . passwordResetCodeStoreInterpreter . randomToIO . runDelay . nowToIOAction e.currentTime