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 <[email protected]>
2026-10-12 02:22:45 +00:00 · 2026-10-04 14:43:21 +01:00
parent 64c5d19fc4
commit fcc5670ae2
11 changed files with 113 additions and 25 deletions
@@ -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:
@@ -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
@@ -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))
@@ -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
@@ -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);
 |]
@@ -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);


@@ -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
@@ -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;
+  |]
@@ -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
@@ -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"
@@ -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