From fcc5670ae2c87f02538ba284b47b20fb6789f773 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Sun, 4 Oct 2026 14:43:21 +0100 Subject: [PATCH] agent: keep ratchet hashes for connection lifetime, up to some cap, to prevent replay of expired ratchets (#1893) * agent: keep ratchet hashes for connection lifetime, to prevent replay of expired ratchets * cap * simpler * simpler2 * split * one query * simplify * performance * migration * fix schema * comment * improve --------- Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com> --- simplexmq.cabal | 3 +- src/Simplex/Messaging/Agent.hs | 5 ++- .../Messaging/Agent/Store/AgentStore.hs | 38 +++++++++++++------ .../Agent/Store/Postgres/Migrations/App.hs | 4 +- ..._index.hs => M20260929_ratchet_indexes.hs} | 12 +++--- .../Migrations/agent_postgres_schema.sql | 4 ++ .../Agent/Store/SQLite/Migrations/App.hs | 4 +- .../Migrations/M20260929_ratchet_indexes.hs | 19 ++++++++++ .../Store/SQLite/Migrations/agent_schema.sql | 4 ++ tests/AgentTests/FunctionalAPITests.hs | 29 ++++++++++++-- tests/AgentTests/SQLiteTests.hs | 16 ++++++++ 11 files changed, 113 insertions(+), 25 deletions(-) rename src/Simplex/Messaging/Agent/Store/Postgres/Migrations/{M20260929_skipped_messages_index.hs => M20260929_ratchet_indexes.hs} (56%) create mode 100644 src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs diff --git a/simplexmq.cabal b/simplexmq.cabal index 78fd61a1f..7244d4de0 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -195,7 +195,7 @@ library Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260712_address_dr_rpc Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260823_snd_files_entitlement Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260919_ratchet_verify_codes - Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_skipped_messages_index + Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_ratchet_indexes else exposed-modules: Simplex.Messaging.Agent.Store.SQLite @@ -251,6 +251,7 @@ library Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260712_address_dr_rpc Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260823_snd_files_entitlement Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260919_ratchet_verify_codes + Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260929_ratchet_indexes Simplex.Messaging.Agent.Store.SQLite.Util if flag(client_postgres) || flag(server_postgres) exposed-modules: diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 1cf455114..d7cd4d453 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -3404,6 +3404,9 @@ subscriber c@AgentClient {msgQ, subQ} = run $ forever $ do run a = a `catchOwn` \e -> notify $ CRITICAL True $ "Agent subscriber stopped: " <> show e notify err = atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR err) +maxRatchetKeyHashes :: Int +maxRatchetKeyHashes = 100 + cleanupManager :: AgentClient -> AM' () cleanupManager c@AgentClient {subQ} = do AgentConfig {initialCleanupDelay, cleanupInterval = int, storedMsgDataTTL = ttl, cleanupBatchSize = limit} <- @@ -3413,7 +3416,7 @@ cleanupManager c@AgentClient {subQ} = do run ERR deleteConns run ERR $ withStore' c $ \db -> deleteRcvMsgHashesExpired db ttl limit run ERR $ withStore' c $ \db -> deleteSndMsgsExpired db ttl limit - run ERR $ withStore' c $ \db -> deleteRatchetKeyHashesExpired db ttl limit + run ERR $ withStore' c $ \db -> deleteRatchetKeyHashesExpired db ttl maxRatchetKeyHashes run ERR $ withStore' c (`deleteExpiredNtfTokensToDelete` ttl) run RFERR deleteRcvFilesExpired run RFERR deleteRcvFilesDeleted diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 38b454c2b..25ea50495 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/src/Simplex/Messaging/Agent/Store/AgentStore.hs @@ -2818,21 +2818,35 @@ checkRatchetKeyHashExists db connId hash = (connId, Binary hash) deleteRatchetKeyHashesExpired :: DB.Connection -> NominalDiffTime -> Int -> IO () -deleteRatchetKeyHashesExpired db ttl limit = do +deleteRatchetKeyHashesExpired db ttl maxConnHashes = do cutoffTs <- addUTCTime (-ttl) <$> getCurrentTime +#if defined(dbPostgres) DB.execute db - [sql| - DELETE FROM processed_ratchet_key_hashes - WHERE processed_ratchet_key_hash_id IN ( - SELECT processed_ratchet_key_hash_id - FROM processed_ratchet_key_hashes - WHERE created_at < ? - ORDER BY created_at ASC - LIMIT ? - ) - |] - (cutoffTs, limit) + ("DELETE FROM processed_ratchet_key_hashes h USING (" <> maxExcessIdsQuery <> ") e WHERE h.conn_id = e.conn_id AND h.processed_ratchet_key_hash_id <= e.max_excess_id AND h.created_at < ?") + (maxConnHashes, maxConnHashes, cutoffTs) +#else + maxExcessIds <- DB.query db maxExcessIdsQuery (maxConnHashes, maxConnHashes) + DB.executeMany + db + "DELETE FROM processed_ratchet_key_hashes WHERE conn_id = ? AND processed_ratchet_key_hash_id <= ? AND created_at < ?" + (map (\(connId :: ConnId, maxExcessId :: Int64) -> (connId, maxExcessId, cutoffTs)) maxExcessIds) +#endif + where + maxExcessIdsQuery :: Query + maxExcessIdsQuery = + [sql| + SELECT conn_id, ( + SELECT processed_ratchet_key_hash_id + FROM processed_ratchet_key_hashes + WHERE conn_id = c.conn_id + ORDER BY processed_ratchet_key_hash_id DESC + LIMIT 1 OFFSET ? + ) AS max_excess_id + FROM processed_ratchet_key_hashes c + GROUP BY conn_id + HAVING COUNT(*) > ? + |] -- | returns all connection queues, the first queue is the primary one getRcvQueuesByConnId_ :: DB.Connection -> ConnId -> IO (Maybe (NonEmpty RcvQueue)) diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs index 0173c4fb8..26c7a80da 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs @@ -16,7 +16,7 @@ import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260411_service_certs import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260712_address_dr_rpc import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260823_snd_files_entitlement import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260919_ratchet_verify_codes -import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_skipped_messages_index +import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_ratchet_indexes import Simplex.Messaging.Agent.Store.Shared (Migration (..)) schemaMigrations :: [(String, Text, Maybe Text)] @@ -33,7 +33,7 @@ schemaMigrations = ("20260712_address_dr_rpc", m20260712_address_dr_rpc, Just down_m20260712_address_dr_rpc), ("20260823_snd_files_entitlement", m20260823_snd_files_entitlement, Just down_m20260823_snd_files_entitlement), ("20260919_ratchet_verify_codes", m20260919_ratchet_verify_codes, Just down_m20260919_ratchet_verify_codes), - ("20260929_skipped_messages_index", m20260929_skipped_messages_index, Just down_m20260929_skipped_messages_index) + ("20260929_ratchet_indexes", m20260929_ratchet_indexes, Just down_m20260929_ratchet_indexes) ] -- | The list of migrations in ascending order by date diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_skipped_messages_index.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs similarity index 56% rename from src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_skipped_messages_index.hs rename to src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs index 7f465cec7..37b81f3eb 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_skipped_messages_index.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs @@ -1,21 +1,23 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE QuasiQuotes #-} -module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_skipped_messages_index where +module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_ratchet_indexes where import Data.Text (Text) import Text.RawString.QQ (r) -m20260929_skipped_messages_index :: Text -m20260929_skipped_messages_index = +m20260929_ratchet_indexes :: Text +m20260929_ratchet_indexes = [r| DROP INDEX idx_skipped_messages_conn_id; CREATE INDEX idx_skipped_messages_conn_id ON skipped_messages(conn_id, skipped_message_id); +CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes(conn_id, processed_ratchet_key_hash_id); |] -down_m20260929_skipped_messages_index :: Text -down_m20260929_skipped_messages_index = +down_m20260929_ratchet_indexes :: Text +down_m20260929_ratchet_indexes = [r| +DROP INDEX idx_processed_ratchet_key_hashes_conn_id; DROP INDEX idx_skipped_messages_conn_id; CREATE INDEX idx_skipped_messages_conn_id ON skipped_messages(conn_id); |] diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql index c5be00a6e..e7d90ae27 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql @@ -1169,6 +1169,10 @@ CREATE INDEX idx_ntf_tokens_ntf_host_ntf_port ON smp_agent_test_protocol_schema. +CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, processed_ratchet_key_hash_id); + + + CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (created_at); diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/App.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/App.hs index e50d8bd43..bae3b6afc 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/App.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/App.hs @@ -52,6 +52,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260411_service_certs import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260712_address_dr_rpc import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260823_snd_files_entitlement import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260919_ratchet_verify_codes +import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260929_ratchet_indexes import Simplex.Messaging.Agent.Store.Shared (Migration (..)) schemaMigrations :: [(String, Query, Maybe Query)] @@ -103,7 +104,8 @@ schemaMigrations = ("m20260411_service_certs", m20260411_service_certs, Just down_m20260411_service_certs), ("m20260712_address_dr_rpc", m20260712_address_dr_rpc, Just down_m20260712_address_dr_rpc), ("m20260823_snd_files_entitlement", m20260823_snd_files_entitlement, Just down_m20260823_snd_files_entitlement), - ("m20260919_ratchet_verify_codes", m20260919_ratchet_verify_codes, Just down_m20260919_ratchet_verify_codes) + ("m20260919_ratchet_verify_codes", m20260919_ratchet_verify_codes, Just down_m20260919_ratchet_verify_codes), + ("m20260929_ratchet_indexes", m20260929_ratchet_indexes, Just down_m20260929_ratchet_indexes) ] -- | The list of migrations in ascending order by date diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs new file mode 100644 index 000000000..d2ca60a35 --- /dev/null +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs @@ -0,0 +1,19 @@ +{-# LANGUAGE QuasiQuotes #-} + +module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260929_ratchet_indexes where + +import Database.SQLite.Simple (Query) +import Database.SQLite.Simple.QQ (sql) + +-- idx_skipped_messages_conn_id is not changed: SQLite index entries include rowid (skipped_message_id) +m20260929_ratchet_indexes :: Query +m20260929_ratchet_indexes = + [sql| +CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes(conn_id, processed_ratchet_key_hash_id); + |] + +down_m20260929_ratchet_indexes :: Query +down_m20260929_ratchet_indexes = + [sql| +DROP INDEX idx_processed_ratchet_key_hashes_conn_id; + |] diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index de0c1494b..47fd154a0 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -641,6 +641,10 @@ CREATE INDEX idx_connections_deleted ON connections(deleted); CREATE INDEX idx_connections_service_request_expires_at ON connections( service_request_expires_at ); +CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes( + conn_id, + processed_ratchet_key_hash_id +); CREATE TRIGGER tr_rcv_queue_insert AFTER INSERT ON rcv_queues FOR EACH ROW diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 8194e9ccd..ac9ae7a2c 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -71,7 +71,7 @@ import qualified Data.ByteString.Char8 as B import Data.Either (isRight) import Data.Int (Int64) import Data.List (find, isPrefixOf, isSuffixOf) -import Data.List.NonEmpty (NonEmpty) +import Data.List.NonEmpty (NonEmpty (..)) import qualified Data.Map as M import Data.Maybe (isJust, isNothing) import qualified Data.Set as S @@ -87,12 +87,12 @@ import SMPAgentClient import SMPClient import Simplex.Messaging.Agent hiding (acceptContact, createConnection, deleteConnection, deleteConnections, getConnShortLink, joinConnection, sendMessage, setConnShortLink, subscribeConnection, suspendConnection) import qualified Simplex.Messaging.Agent as A -import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), ServerQueueInfo (..), UserNetworkInfo (..), UserNetworkType (..), waitForUserNetwork) +import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), ServerQueueInfo (..), UserNetworkInfo (..), UserNetworkType (..), sendAgentMessage, waitForUserNetwork) import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), Env (..), InitialAgentServers (..), createAgentStore) import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ, SENT) import qualified Simplex.Messaging.Agent.Protocol as A import Simplex.Messaging.Agent.Store (Connection' (..), SomeConn' (..), StoredRcvQueue (..)) -import Simplex.Messaging.Agent.Store.AgentStore (getConn) +import Simplex.Messaging.Agent.Store.AgentStore (deleteRatchetKeyHashesExpired, getConn, getRatchetX3dhKeys) import Simplex.Messaging.Agent.Store.Common (DBStore (..), withTransaction) import Simplex.Messaging.Agent.Store.Interface import qualified Simplex.Messaging.Agent.Store.DB as DB @@ -463,6 +463,8 @@ functionalAPITests ps = do testRatchetSyncSuspendForeground ps it "should synchronize ratchets when clients start synchronization simultaneously" $ testRatchetSyncSimultaneous ps + it "should ignore replayed ratchet key after expired hashes are deleted" $ + testRatchetSyncReplayedKey ps #endif describe "Subscription mode OnlyCreate" $ do it "messages delivered only when polled" $ @@ -2748,6 +2750,27 @@ testRatchetSyncSimultaneous ps = do disposeAgentClient bob disposeAgentClient bob2 +testRatchetSyncReplayedKey :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncReplayedKey ps = withAgentClients2 $ \alice bob -> do + (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ -> + setupDesynchronizedRatchet alice bob + ("", "", DOWN _ _) <- nGet alice + ("", "", DOWN _ _) <- nGet bob2 + _ <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn False + Right pks <- withTransaction (store $ agentEnv bob2) (`getRatchetX3dhKeys` aliceId) + Right (SomeConn _ (DuplexConnection _ _ (sq :| _))) <- withTransaction (store $ agentEnv bob2) (`getConn` aliceId) + withSmpServerStoreMsgLogOn ps testPort $ \_ -> do + concurrently_ + (getInAnyOrder alice [ratchetSyncP' bobId RSAgreed, serverUpP]) + (getInAnyOrder bob2 [ratchetSyncP' aliceId RSAgreed, serverUpP]) + get alice =##> ratchetSyncP bobId RSOk + get bob2 =##> ratchetSyncP aliceId RSOk + withTransaction (store $ agentEnv alice) $ \db -> deleteRatchetKeyHashesExpired db 0 100 + let keyMsg = AgentRatchetKey {agentVersion = currentSMPAgentVersion, e2eEncryption = CR.mkRcvE2ERatchetParams CR.currentE2EEncryptVersion pks, info = ""} + Right _ <- runReaderT (runExceptT $ sendAgentMessage bob2 sq SMP.noMsgFlags $ smpEncode keyMsg) (agentEnv bob2) + runRight_ $ exchangeGreetingsMsgIds alice bobId 10 bob2 aliceId 7 + disposeAgentClient bob2 + getMsg :: AgentClient -> ConnId -> ExceptT AgentErrorType IO a -> ExceptT AgentErrorType IO a getMsg c cId action = do liftIO $ noMessages c "nothing should be delivered before GET" diff --git a/tests/AgentTests/SQLiteTests.hs b/tests/AgentTests/SQLiteTests.hs index 0a418cabb..8a5fd14d7 100644 --- a/tests/AgentTests/SQLiteTests.hs +++ b/tests/AgentTests/SQLiteTests.hs @@ -137,6 +137,7 @@ storeTests = do testCreateRcvMsg testCreateSndMsg testCreateRcvAndSndMsgs + describe "deleteRatchetKeyHashesExpired" testDeleteRatchetKeyHashesExpired it "should keep only the newest skipped message keys" testGetSkippedMsgKeys describe "Work items" $ do it "should getPendingQueueMsg" testGetPendingQueueMsg @@ -602,6 +603,21 @@ testCreateRcvAndSndMsgs = testCreateSndMsg_ db "snd_hash_1" connId sq $ mkSndMsgData (InternalId 5) (InternalSndId 2) "snd_hash_2" testCreateSndMsg_ db "snd_hash_2" connId sq $ mkSndMsgData (InternalId 6) (InternalSndId 3) "snd_hash_3" +testDeleteRatchetKeyHashesExpired :: SpecWith DBStore +testDeleteRatchetKeyHashesExpired = + it "should delete expired ratchet key hashes except the newest in each connection" . withStoreTransaction $ \db -> do + g <- C.newRandom + Right connId <- createNewConn db g cData1 {connId = ""} SCMInvitation + Right connId' <- createNewConn db g cData1 {connId = ""} SCMContact + let hashes = ["h1", "h2", "h3", "h4", "h5", "h6"] + forM_ hashes $ addProcessedRatchetKeyHash db connId + forM_ (take 4 hashes) $ addProcessedRatchetKeyHash db connId' + deleteRatchetKeyHashesExpired db 86400 4 + mapM (checkRatchetKeyHashExists db connId) hashes `shouldReturn` replicate 6 True + deleteRatchetKeyHashesExpired db 0 4 + mapM (checkRatchetKeyHashExists db connId) hashes `shouldReturn` [False, False, True, True, True, True] + mapM (checkRatchetKeyHashExists db connId') (take 4 hashes) `shouldReturn` replicate 4 True + testGetSkippedMsgKeys :: DBStore -> Expectation testGetSkippedMsgKeys st = do g <- C.newRandom