diff --git a/protocol/agent-protocol.md b/protocol/agent-protocol.md index 00b099e40..f30b33b92 100644 --- a/protocol/agent-protocol.md +++ b/protocol/agent-protocol.md @@ -208,7 +208,7 @@ This syntax of decrypted SMP client message body is defined by `decryptedAgentMe Decrypted SMP message client body can be one of 4 types: - `agentConnInfo` - used by the initiating party when confirming reply queue - sent in `agentConfirmation` envelope. - `agentConnInfoReply` - used by accepting party, includes reply queue(s) in the initial confirmation - sent in `agentConfirmation` envelope. -- `agentRatchetInfo` - used to pass additional information when renegotiating double ratchet encryption - sent in `agentRatchetKey` envelope. +- `agentRatchetInfo` - used to pass additional information when renegotiating double ratchet encryption - sent in `agentRatchetKey` envelope. A key sent in reply to another key includes the hash of that key; agents do not reply to such keys. - `agentMessage` - all other agent messages. `agentMessage` contains these parts: @@ -233,7 +233,8 @@ connInfo = *OCTET agentConnInfoReply = %s"D" smpQueues connInfo smpQueues = length 1*newQueueInfo ; NonEmpty list of reply queues agentRatchetInfo = %s"R" ratchetInfo -ratchetInfo = *OCTET +ratchetInfo = [answeredKeyHash *OCTET] ; bytes after answeredKeyHash are ignored +answeredKeyHash = %s"0" / (%s"1" shortString) ; "0" in a key that starts renegotiation, otherwise SHA-256 of the two raw public keys of the answered key agentMessage = %s"M" agentMsgHeader aMessage agentMsgHeader = agentMsgId prevMsgHash diff --git a/simplexmq.cabal b/simplexmq.cabal index 39b40dd33..7244d4de0 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -195,6 +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_ratchet_indexes else exposed-modules: Simplex.Messaging.Agent.Store.SQLite @@ -250,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 be34deafb..067031899 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -2435,62 +2435,66 @@ enqueueMessageB c reqs = do cfg <- asks config (_, reqMids) <- unsafeWithStore c $ \db -> do mapAccumLM (\ids r -> storeSentMsg db cfg ids r `E.catchAny` \e -> (ids,) <$> handleInternal e) IM.empty reqs - forME reqMids $ \((csqs_, _, _, _), InternalId msgId, pqSecr) -> forM csqs_ $ \(cData, sq :| sqs) -> do - submitPendingMsg c sq - let sqs' = filter (isActiveSndQ cData) sqs - pure ((msgId, pqSecr), if null sqs' then Nothing else Just (sqs', msgId)) + forME reqMids $ submitSentMsg c where - storeSentMsg :: - DB.Connection -> - AgentConfig -> - IntMap (Maybe Int64, AMessage) -> - Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage) -> - IO (IntMap (Maybe Int64, AMessage), Either AgentErrorType ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption)) - storeSentMsg db cfg aMessageIds = \case - Left e -> pure (aMessageIds, Left e) - Right req@(csqs_, pqEnc_, msgFlags, mbr) -> case mbr of - VRValue i_ aMessage -> case i_ >>= (`IM.lookup` aMessageIds) of - Just _ -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg duplicate saved message body") - Nothing -> do - (mbId_, r) <- case csqs_ of - Left e -> pure (Nothing, Left e) - Right (cData, sq :| _) -> do - mbId <- createSndMsgBody db aMessage - (Just mbId,) <$> storeSentMsg_ cData sq mbId aMessage - let aMessageIds' = maybe id (`IM.insert` (mbId_, aMessage)) i_ aMessageIds - pure (aMessageIds', r) - VRRef i -> case csqs_ of - Left e -> pure $ (aMessageIds, Left e) - Right (cData, sq :| _) -> case IM.lookup i aMessageIds of - Just (Just mbId, aMessage) -> (aMessageIds,) <$> storeSentMsg_ cData sq mbId aMessage - Just (Nothing, aMessage) -> do - mbId <- createSndMsgBody db aMessage - let aMessageIds' = IM.insert i (Just mbId, aMessage) aMessageIds - (aMessageIds',) <$> storeSentMsg_ cData sq mbId aMessage - Nothing -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg missing saved message body id") - where - storeSentMsg_ cData@ConnData {connId} sq sndMsgBodyId aMessage = fmap (first storeError) $ runExceptT $ do - let AgentConfig {e2eEncryptVRange} = cfg - internalTs <- liftIO getCurrentTime - (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId - -- We need to do pre-flight encoding that is not stored in database - -- to calculate its hash and remember it on connection (createSndMsg -> updateSndMsgHash) - -- to enable next enqueue. - -- (As encoding is different per connection, we can't store shared body, so it's repeated on delivery) - let agentMsgStr = encodeAgentMsgStr aMessage internalSndId prevMsgHash - internalHash = C.sha256Hash agentMsgStr - currentE2EVersion = maxVersion e2eEncryptVRange - (mek, paddedLen, pqEnc) <- agentRatchetEncryptHeader db cData e2eEncAgentMsgLength pqEnc_ currentE2EVersion - withExceptT (SEAgentError . cryptoError) $ CR.rcCheckCanPad paddedLen agentMsgStr - let msgType = aMessageType aMessage - -- msgBody is empty, because snd_messages record is linked to snd_message_bodies - msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody = "", pqEncryption = pqEnc, internalHash, prevMsgHash, sndMsgPrepData_ = Just SndMsgPrepData {encryptKey = mek, paddedLen, sndMsgBodyId}} - liftIO $ createSndMsg db connId msgData - liftIO $ createSndMsgDelivery db sq internalId - pure (req, internalId, pqEnc) handleInternal :: E.SomeException -> IO (Either AgentErrorType b) handleInternal = pure . Left . INTERNAL . show +submitSentMsg :: AgentClient -> ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption) -> AM' (Either AgentErrorType ((AgentMsgId, PQEncryption), Maybe ([SndQueue], AgentMsgId))) +submitSentMsg c ((csqs_, _, _, _), InternalId msgId, pqSecr) = forM csqs_ $ \(cData, sq :| sqs) -> do + submitPendingMsg c sq + let sqs' = filter (isActiveSndQ cData) sqs + pure ((msgId, pqSecr), if null sqs' then Nothing else Just (sqs', msgId)) + +storeSentMsg :: + DB.Connection -> + AgentConfig -> + IntMap (Maybe Int64, AMessage) -> + Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage) -> + IO (IntMap (Maybe Int64, AMessage), Either AgentErrorType ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption)) +storeSentMsg db cfg aMessageIds = \case + Left e -> pure (aMessageIds, Left e) + Right req@(csqs_, pqEnc_, msgFlags, mbr) -> case mbr of + VRValue i_ aMessage -> case i_ >>= (`IM.lookup` aMessageIds) of + Just _ -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg duplicate saved message body") + Nothing -> do + (mbId_, r) <- case csqs_ of + Left e -> pure (Nothing, Left e) + Right (cData, sq :| _) -> do + mbId <- createSndMsgBody db aMessage + (Just mbId,) <$> storeSentMsg_ cData sq mbId aMessage + let aMessageIds' = maybe id (`IM.insert` (mbId_, aMessage)) i_ aMessageIds + pure (aMessageIds', r) + VRRef i -> case csqs_ of + Left e -> pure $ (aMessageIds, Left e) + Right (cData, sq :| _) -> case IM.lookup i aMessageIds of + Just (Just mbId, aMessage) -> (aMessageIds,) <$> storeSentMsg_ cData sq mbId aMessage + Just (Nothing, aMessage) -> do + mbId <- createSndMsgBody db aMessage + let aMessageIds' = IM.insert i (Just mbId, aMessage) aMessageIds + (aMessageIds',) <$> storeSentMsg_ cData sq mbId aMessage + Nothing -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg missing saved message body id") + where + storeSentMsg_ cData@ConnData {connId} sq sndMsgBodyId aMessage = fmap (first storeError) $ runExceptT $ do + let AgentConfig {e2eEncryptVRange} = cfg + internalTs <- liftIO getCurrentTime + (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId + -- We need to do pre-flight encoding that is not stored in database + -- to calculate its hash and remember it on connection (createSndMsg -> updateSndMsgHash) + -- to enable next enqueue. + -- (As encoding is different per connection, we can't store shared body, so it's repeated on delivery) + let agentMsgStr = encodeAgentMsgStr aMessage internalSndId prevMsgHash + internalHash = C.sha256Hash agentMsgStr + currentE2EVersion = maxVersion e2eEncryptVRange + (mek, paddedLen, pqEnc) <- agentRatchetEncryptHeader db cData e2eEncAgentMsgLength pqEnc_ currentE2EVersion + withExceptT (SEAgentError . cryptoError) $ CR.rcCheckCanPad paddedLen agentMsgStr + let msgType = aMessageType aMessage + -- msgBody is empty, because snd_messages record is linked to snd_message_bodies + msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody = "", pqEncryption = pqEnc, internalHash, prevMsgHash, sndMsgPrepData_ = Just SndMsgPrepData {encryptKey = mek, paddedLen, sndMsgBodyId}} + liftIO $ createSndMsg db connId msgData + liftIO $ createSndMsgDelivery db sq internalId + pure (req, internalId, pqEnc) + encodeAgentMsgStr :: AMessage -> InternalSndId -> PrevSndMsgHash -> ByteString encodeAgentMsgStr aMessage internalSndId prevMsgHash = do let privHeader = APrivHeader (unSndId internalSndId) prevMsgHash @@ -2865,15 +2869,18 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni SomeConn _ (DuplexConnection cData@ConnData {pqSupport} rqs sqs) | ratchetSyncAllowed cData || force -> do -- check queues are not switching? - when (pqSupport' /= pqSupport) $ withStore' c $ \db -> setConnPQSupport db connId pqSupport' let cData' = cData {pqSupport = pqSupport'} :: ConnData - AgentConfig {e2eEncryptVRange} <- asks config + AgentConfig {e2eEncryptVRange, smpAgentVRange} <- asks config g <- asks random (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eEncryptVRange) pqSupport' - enqueueRatchetKeyMsgs c cData' sqs e2eParams - withStore' c $ \db -> do - setConnRatchetSync db connId RSStarted - setRatchetX3dhKeys db connId pks + msgId <- withStore c $ \db -> runExceptT $ do + msgId <- storeRatchetKey db (L.head sqs) (maxVersion smpAgentVRange) e2eParams Nothing + liftIO $ do + when (pqSupport' /= pqSupport) $ setConnPQSupport db connId pqSupport' + setConnRatchetSync db connId RSStarted + setRatchetX3dhKeys db connId pks + pure msgId + lift $ enqueueRatchetKeyMsgs c cData' sqs msgId let cData'' = cData' {ratchetSyncState = RSStarted} :: ConnData conn' = DuplexConnection cData'' rqs sqs connectionStats c conn' @@ -3404,6 +3411,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 +3423,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 @@ -3610,9 +3620,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar _ -> prohibited "handshake: incorrect state" >> ack (Just e2eDh, Nothing) -> do decryptClientMessage e2eDh clientMsg >>= \case - (SMP.PHEmpty, AgentRatchetKey {agentVersion, e2eEncryption}) -> do + (SMP.PHEmpty, AgentRatchetKey {agentVersion, e2eEncryption, info}) -> do conn' <- updateConnVersion conn cData agentVersion - qDuplex conn' "AgentRatchetKey" $ \a -> newRatchetKey e2eEncryption a >> ack + qDuplex conn' "AgentRatchetKey" $ \a -> newRatchetKey e2eEncryption info a >> ack (SMP.PHEmpty, AgentMsgEnvelope {agentVersion, encAgentMessage}) -> do conn' <- updateConnVersion conn cData agentVersion -- primary queue is set as Active in helloMsg, below is to set additional queues Active @@ -3655,7 +3665,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar qDuplexAckDel conn'' name a = qDuplex conn'' name a >> ackDel msgId resetRatchetSync :: AM (Connection c) resetRatchetSync - | rss `notElem` ([RSOk, RSStarted] :: [RatchetSyncState]) = do + | rss `notElem` ([RSOk, RSStarted] :: [RatchetSyncState]) = ifM ((RSStarted ==) <$> getRatchetSyncState) (pure conn') $ do let cData'' = (toConnData conn') {ratchetSyncState = RSOk} :: ConnData conn'' = updateConnection cData'' conn' cStats <- connectionStats c conn'' @@ -3696,7 +3706,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar notifySync :: AM () notifySync = qDuplex conn' "AGENT A_CRYPTO error" $ \connDuplex -> do let rss' = cryptoErrToSyncState e - when (rss `elem` ([RSOk, RSAllowed, RSRequired] :: [RatchetSyncState])) $ do + whenM ((`elem` ([RSOk, RSAllowed, RSRequired] :: [RatchetSyncState])) <$> getRatchetSyncState) $ do let cData'' = (toConnData conn') {ratchetSyncState = rss'} :: ConnData conn'' = updateConnection cData'' connDuplex cStats <- connectionStats c conn'' @@ -3792,6 +3802,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar enqueueCmd :: InternalCommand -> AM () enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand + getRatchetSyncState :: AM RatchetSyncState + getRatchetSyncState = withStore c (`getConnRatchetSync` connId) + unexpected :: BrokerMsg -> AM () unexpected r = do logServer "<--" c srv rId $ "unexpected: " <> bshow r @@ -4160,42 +4173,47 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar DuplexConnection {} -> action conn' _ -> qError $ name <> ": message must be sent to duplex connection" - newRatchetKey :: CR.RcvE2ERatchetParams 'C.X448 -> Connection 'CDuplex -> AM () - newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) = + newRatchetKey :: CR.RcvE2ERatchetParams 'C.X448 -> ByteString -> Connection 'CDuplex -> AM () + newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) info conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) = unlessM ratchetExists $ do AgentConfig {e2eEncryptVRange} <- asks config unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwE $ AGENT A_VERSION) - keys <- getSendRatchetKeys + keys_ <- getSendRatchetKeys let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eEncryptVRange} - initRatchet rcVs keys - notifyAgreed + case keys_ of + Just (keys, replyKey_) -> initRatchet rcVs keys replyKey_ >> notifyAgreed + Nothing -> withStore' c $ \db -> addProcessedRatchetKeyHash db connId rkHashRcv where rkHashRcv = rkHash k1Rcv k2Rcv rkHash k1 k2 = C.sha256Hash $ C.pubKeyBytes k1 <> C.pubKeyBytes k2 + answeredKeyHash_ = case smpDecode info of + Right (AgentRatchetInfo RatchetInfo {answeredKeyHash}) -> answeredKeyHash + _ -> Nothing ratchetExists :: AM Bool ratchetExists = withStore' c $ \db -> do exists <- checkRatchetKeyHashExists db connId rkHashRcv - unless exists $ addProcessedRatchetKeyHash db connId rkHashRcv pure exists - getSendRatchetKeys :: AM (CR.RcvE2EPrivRatchetParams 'C.X448) - getSendRatchetKeys = case rss of - RSOk -> sendReplyKey -- receiving client - RSAllowed -> sendReplyKey - RSRequired -> sendReplyKey - RSStarted -> withStore c (`getRatchetX3dhKeys` connId) -- initiating client - RSAgreed -> do - withStore' c $ \db -> setConnRatchetSync db connId RSRequired + getSendRatchetKeys :: AM (Maybe (CR.RcvE2EPrivRatchetParams 'C.X448, Maybe (CR.RcvE2ERatchetParams 'C.X448))) + getSendRatchetKeys = getRatchetSyncState >>= \rss' -> case (rss', answeredKeyHash_) of + (RSStarted, Nothing) -> Just . (,Nothing) <$> withStore c (`getRatchetX3dhKeys` connId) -- initiating client + (RSStarted, Just h) -> fmap (,Nothing) . answeredKeys h <$> withStore c (`getRatchetX3dhKeys` connId) + (_, Nothing) -> Just <$> sendReplyKey -- receiving client + (RSAgreed, Just _) -> do + withStore' c $ \db -> addProcessedRatchetKeyHash db connId rkHashRcv >> setConnRatchetSync db connId RSRequired notifyRatchetSyncError -- can communicate for other client to reset to RSRequired -- - need to add new AgentMsgEnvelope, AgentMessage, AgentMessageType -- - need to deduplicate on receiving side throwE $ AGENT (A_CRYPTO RATCHET_SYNC) + (_, Just _) -> pure Nothing where + answeredKeys h keys@(pk1, pk2, _) + | rkHash (C.publicKey pk1) (C.publicKey pk2) == h = Just keys + | otherwise = Nothing sendReplyKey = do g <- asks random (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g e2eVersion pqSupport - enqueueRatchetKeyMsgs c cData' sqs e2eParams - pure pks + pure (pks, Just e2eParams) notifyRatchetSyncError = do let cData'' = cData' {ratchetSyncState = RSRequired} :: ConnData conn'' = updateConnection cData'' conn' @@ -4207,23 +4225,33 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar conn'' = updateConnection cData'' conn' cStats <- connectionStats c conn'' notify $ RSYNC RSAgreed Nothing cStats - recreateRatchet :: CR.Ratchet 'C.X448 -> AM () - recreateRatchet rc = withStore' c $ \db -> do - setConnRatchetSync db connId RSAgreed - deleteRatchet db connId - createRatchet db connId rc + recreateRatchet :: Maybe (CR.RcvE2ERatchetParams 'C.X448) -> Maybe AMessage -> CR.Ratchet 'C.X448 -> AM () + recreateRatchet replyKey_ eready_ rc = do + cfg@AgentConfig {smpAgentVRange} <- asks config + (msgId_, r_) <- withStore c $ \db -> runExceptT $ do + msgId_ <- forM replyKey_ $ \e2eParams -> storeRatchetKey db (L.head sqs) (maxVersion smpAgentVRange) e2eParams (Just rkHashRcv) + liftIO $ do + addProcessedRatchetKeyHash db connId rkHashRcv + setConnRatchetSync db connId RSAgreed + deleteRatchet db connId + createRatchet db connId rc + r_ <- forM eready_ $ \msg -> liftIO $ storeSentMsg db cfg IM.empty (Right (Right (cData', sqs), Nothing, SMP.MsgFlags {notification = True}, vrValue msg)) >>= either E.throwIO pure . snd + pure (msgId_, r_) + forM_ msgId_ $ lift . enqueueRatchetKeyMsgs c cData' sqs + forM_ r_ $ \r -> lift $ do + r' <- submitSentMsg c r + enqueueSavedMessageB c $ mapMaybe snd $ rights [r'] -- compare public keys `k1` in AgentRatchetKey messages sent by self and other party -- to determine ratchet initilization ordering - initRatchet :: CR.RatchetVersions -> CR.RcvE2EPrivRatchetParams 'C.X448 -> AM () - initRatchet rcVs (pk1, pk2, pKem) + initRatchet :: CR.RatchetVersions -> CR.RcvE2EPrivRatchetParams 'C.X448 -> Maybe (CR.RcvE2ERatchetParams 'C.X448) -> AM () + initRatchet rcVs (pk1, pk2, pKem) replyKey_ | rkHash (C.publicKey pk1) (C.publicKey pk2) <= rkHashRcv = do rcParams <- liftError cryptoError $ CR.pqX3dhRcv (pk1, pk2, pKem) e2eOtherPartyParams - recreateRatchet $ CR.initRcvRatchet rcVs pk2 rcParams pqSupport + recreateRatchet replyKey_ Nothing $ CR.initRcvRatchet rcVs pk2 rcParams pqSupport | otherwise = do (_, rcDHRs) <- atomically . C.generateKeyPair =<< asks random rcParams <- liftEitherWith cryptoError $ CR.pqX3dhSnd (pk1, pk2, CR.APRKP CR.SRKSProposed <$> pKem) e2eOtherPartyParams - recreateRatchet $ CR.initSndRatchet rcVs k2Rcv rcDHRs rcParams - void . enqueueMessages' c cData' sqs SMP.MsgFlags {notification = True} $ EREADY lastExternalSndId + recreateRatchet replyKey_ (Just $ EREADY lastExternalSndId) $ CR.initSndRatchet rcVs k2Rcv rcDHRs rcParams checkMsgIntegrity :: PrevExternalSndId -> ExternalSndId -> PrevRcvMsgHash -> ByteString -> MsgIntegrity checkMsgIntegrity prevExtSndId extSndId internalPrevMsgHash receivedPrevMsgHash @@ -4336,32 +4364,25 @@ storeConfirmation c cData@ConnData {connId, pqSupport, connAgentVersion = v} sq liftIO $ createSndMsg db connId msgData liftIO $ createSndMsgDelivery db sq internalId -enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM () -enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do - msgId <- enqueueRatchetKey c sq e2eEncryption - mapM_ (lift . enqueueSavedMessage c msgId) $ filter (isActiveSndQ cData) sqs +enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> InternalId -> AM' () +enqueueRatchetKeyMsgs c cData (sq :| sqs) (InternalId msgId) = do + submitPendingMsg c sq + mapM_ (enqueueSavedMessage c msgId) $ filter (isActiveSndQ cData) sqs -enqueueRatchetKey :: AgentClient -> SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM AgentMsgId -enqueueRatchetKey c sq@SndQueue {connId} e2eEncryption = do - aVRange <- asks $ smpAgentVRange . config - msgId <- storeRatchetKey $ maxVersion aVRange - lift $ submitPendingMsg c sq - pure $ unId msgId - where - storeRatchetKey :: VersionSMPA -> AM InternalId - storeRatchetKey agentVersion = withStore c $ \db -> runExceptT $ do - internalTs <- liftIO getCurrentTime - (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId - let agentMsg = AgentRatchetInfo "" - agentMsgStr = smpEncode agentMsg - internalHash = C.sha256Hash agentMsgStr - let msgBody = smpEncode $ AgentRatchetKey {agentVersion, e2eEncryption, info = agentMsgStr} - msgType = agentMessageType agentMsg - -- this message is e2e encrypted with queue key, not with double ratchet - msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, pqEncryption = PQEncOff, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, sndMsgPrepData_ = Nothing} - liftIO $ createSndMsg db connId msgData - liftIO $ createSndMsgDelivery db sq internalId - pure internalId +storeRatchetKey :: DB.Connection -> SndQueue -> VersionSMPA -> CR.RcvE2ERatchetParams 'C.X448 -> Maybe ByteString -> ExceptT StoreError IO InternalId +storeRatchetKey db sq@SndQueue {connId} agentVersion e2eEncryption answeredKeyHash_ = do + internalTs <- liftIO getCurrentTime + (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId + let agentMsg = AgentRatchetInfo RatchetInfo {answeredKeyHash = answeredKeyHash_} + agentMsgStr = smpEncode agentMsg + internalHash = C.sha256Hash agentMsgStr + let msgBody = smpEncode $ AgentRatchetKey {agentVersion, e2eEncryption, info = agentMsgStr} + msgType = agentMessageType agentMsg + -- this message is e2e encrypted with queue key, not with double ratchet + msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, pqEncryption = PQEncOff, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, sndMsgPrepData_ = Nothing} + liftIO $ createSndMsg db connId msgData + liftIO $ createSndMsgDelivery db sq internalId + pure internalId -- encoded AgentMessage -> encoded EncAgentMessage agentRatchetEncrypt :: DB.Connection -> ConnData -> ByteString -> (PQSupport -> Int) -> Maybe PQEncryption -> CR.VersionE2E -> ExceptT StoreError IO (ByteString, PQEncryption) @@ -4386,7 +4407,7 @@ agentRatchetDecrypt g db connId encAgentMsg = do agentRatchetDecrypt' :: TVar ChaChaDRG -> DB.Connection -> ConnId -> CR.RatchetX448 -> ByteString -> ExceptT StoreError IO (ByteString, PQEncryption) agentRatchetDecrypt' g db connId rc encAgentMsg = do - skipped <- liftIO $ getSkippedMsgKeys db connId + skipped <- liftIO $ getSkippedMsgKeys db connId CR.maxSkippedMsgKeys (agentMsgBody_, rc', skippedDiff) <- withExceptT (SEAgentError . cryptoError) $ CR.rcDecrypt g rc skipped encAgentMsg agentMsgBody <- liftEither $ first (SEAgentError . cryptoError) agentMsgBody_ liftIO $ updateRatchet db connId rc' skippedDiff diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index afe698769..5f7977e45 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -79,6 +79,7 @@ module Simplex.Messaging.Agent.Protocol SMPConfirmation (..), AgentMsgEnvelope (..), AgentMessage (..), + RatchetInfo (..), RequestSignature (..), AgentMessageType (..), APrivHeader (..), @@ -928,7 +929,7 @@ data AgentMessage | -- AgentConnInfoReply is used by accepting party in duplexHandshake mode (v2), allowing to include reply queue(s) in the initial confirmation. -- It made removed REPLY message unnecessary. AgentConnInfoReply (NonEmpty SMPQueueInfo) ConnInfo - | AgentRatchetInfo ByteString + | AgentRatchetInfo RatchetInfo | AgentMessage APrivHeader AMessage | AgentServiceRequest (NonEmpty SMPQueueInfo) (Maybe RequestSignature) MsgBody | AgentServiceResponse MsgBody @@ -939,7 +940,7 @@ instance Encoding AgentMessage where smpEncode = \case AgentConnInfo cInfo -> smpEncode ('I', Tail cInfo) AgentConnInfoReply smpQueues cInfo -> smpEncode ('D', smpQueues, Tail cInfo) -- 'D' stands for "duplex" - AgentRatchetInfo info -> smpEncode ('R', Tail info) + AgentRatchetInfo info -> smpEncode ('R', info) AgentMessage hdr aMsg -> smpEncode ('M', hdr, aMsg) AgentServiceRequest qs sig_ body -> smpEncode ('A', qs, sig_, Tail body) AgentServiceResponse body -> smpEncode ('P', Tail body) @@ -948,13 +949,23 @@ instance Encoding AgentMessage where smpP >>= \case 'I' -> AgentConnInfo . unTail <$> smpP 'D' -> AgentConnInfoReply <$> smpP <*> (unTail <$> smpP) - 'R' -> AgentRatchetInfo . unTail <$> smpP + 'R' -> AgentRatchetInfo <$> smpP 'M' -> AgentMessage <$> smpP <*> smpP 'A' -> AgentServiceRequest <$> smpP <*> smpP <*> (unTail <$> smpP) 'P' -> AgentServiceResponse . unTail <$> smpP 'J' -> AgentRejection . unTail <$> smpP _ -> fail "bad AgentMessage" +data RatchetInfo = RatchetInfo {answeredKeyHash :: Maybe ByteString} + deriving (Eq, Show) + +instance Encoding RatchetInfo where + smpEncode RatchetInfo {answeredKeyHash} = smpEncode answeredKeyHash + smpP = do + answeredKeyHash <- smpP <|> pure Nothing + _ <- A.takeByteString + pure RatchetInfo {answeredKeyHash} + -- internal type for storing message type in the database data AgentMessageType = AM_CONN_INFO diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 311cdc14e..21c087d46 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/src/Simplex/Messaging/Agent/Store/AgentStore.hs @@ -76,6 +76,7 @@ module Simplex.Messaging.Agent.Store.AgentStore getExpiredServiceConns, deleteExpiredServiceRequests, getDeletedWaitingDeliveryConnIds, + getConnRatchetSync, setConnRatchetSync, addProcessedRatchetKeyHash, checkRatchetKeyHashExists, @@ -1594,12 +1595,16 @@ getRatchet_ q db connId = where ratchet = maybe (Left SERatchetNotFound) Right . fromOnly -getSkippedMsgKeys :: DB.Connection -> ConnId -> IO SkippedMsgKeys -getSkippedMsgKeys db connId = - skipped <$> DB.query db "SELECT header_key, msg_n, msg_key FROM skipped_messages WHERE conn_id = ?" (Only connId) +getSkippedMsgKeys :: DB.Connection -> ConnId -> Int -> IO SkippedMsgKeys +getSkippedMsgKeys db connId maxKeys = do + (keys, oldKeys) <- splitAt maxKeys <$> DB.query db "SELECT skipped_message_id, header_key, msg_n, msg_key FROM skipped_messages WHERE conn_id = ? ORDER BY skipped_message_id DESC LIMIT ?" (connId, maxKeys + 1) + case oldKeys of + (skippedMsgId :: Int64, _, _, _) : _ -> DB.execute db "DELETE FROM skipped_messages WHERE conn_id = ? AND skipped_message_id <= ?" (connId, skippedMsgId) + [] -> pure () + pure $ skipped keys where skipped = foldl' addSkippedKey M.empty - addSkippedKey smks (hk, msgN, mk) = M.alter (Just . addMsgKey) hk smks + addSkippedKey smks (_, hk, msgN, mk) = M.alter (Just . addMsgKey) hk smks where addMsgKey = maybe (M.singleton msgN mk) (M.insert msgN mk) @@ -2793,6 +2798,11 @@ getDeletedWaitingDeliveryConnIds :: DB.Connection -> IO [ConnId] getDeletedWaitingDeliveryConnIds db = map fromOnly <$> DB.query_ db "SELECT conn_id FROM connections WHERE deleted_at_wait_delivery IS NOT NULL" +getConnRatchetSync :: DB.Connection -> ConnId -> IO (Either StoreError RatchetSyncState) +getConnRatchetSync db connId = + firstRow fromOnly SEConnNotFound $ + DB.query db "SELECT ratchet_sync_state FROM connections WHERE conn_id = ?" (Only connId) + setConnRatchetSync :: DB.Connection -> ConnId -> RatchetSyncState -> IO () setConnRatchetSync db connId ratchetSyncState = DB.execute db "UPDATE connections SET ratchet_sync_state = ? WHERE conn_id = ?" (ratchetSyncState, connId) @@ -2814,21 +2824,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 f34b120e2..26c7a80da 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs @@ -16,6 +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_ratchet_indexes import Simplex.Messaging.Agent.Store.Shared (Migration (..)) schemaMigrations :: [(String, Text, Maybe Text)] @@ -31,7 +32,8 @@ schemaMigrations = ("20260411_service_certs", m20260411_service_certs, Just down_m20260411_service_certs), ("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) + ("20260919_ratchet_verify_codes", m20260919_ratchet_verify_codes, Just down_m20260919_ratchet_verify_codes), + ("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_ratchet_indexes.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs new file mode 100644 index 000000000..511064cf5 --- /dev/null +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs @@ -0,0 +1,34 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE QuasiQuotes #-} + +module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_ratchet_indexes where + +import Data.Text (Text) +import Text.RawString.QQ (r) + +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); + +DELETE FROM processed_ratchet_key_hashes AS h +WHERE EXISTS ( + SELECT 1 FROM processed_ratchet_key_hashes d + WHERE d.conn_id = h.conn_id AND d.hash = h.hash AND d.processed_ratchet_key_hash_id < h.processed_ratchet_key_hash_id +); + +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); +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 :: Text +down_m20260929_ratchet_indexes = + [r| +DROP INDEX idx_processed_ratchet_key_hashes_conn_id; +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); +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 a0d408c36..55a9ab78d 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,11 +1169,15 @@ 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); -CREATE INDEX idx_processed_ratchet_key_hashes_hash ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, hash); +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, hash); @@ -1241,7 +1245,7 @@ CREATE UNIQUE INDEX idx_server_certs_user_id_host_port ON smp_agent_test_protoco -CREATE INDEX idx_skipped_messages_conn_id ON smp_agent_test_protocol_schema.skipped_messages USING btree (conn_id); +CREATE INDEX idx_skipped_messages_conn_id ON smp_agent_test_protocol_schema.skipped_messages USING btree (conn_id, skipped_message_id); 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..a1e70a2e2 --- /dev/null +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs @@ -0,0 +1,29 @@ +{-# 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| +DELETE FROM processed_ratchet_key_hashes AS h +WHERE EXISTS ( + SELECT 1 FROM processed_ratchet_key_hashes d + WHERE d.conn_id = h.conn_id AND d.hash = h.hash AND d.processed_ratchet_key_hash_id < h.processed_ratchet_key_hash_id +); + +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); +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; +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); + |] 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..e118a17eb 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -573,10 +573,6 @@ CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hash conn_id, hash ); -CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes( - conn_id, - hash -); CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages( conn_id, rcpt_internal_id @@ -641,6 +637,14 @@ CREATE INDEX idx_connections_deleted ON connections(deleted); CREATE INDEX idx_connections_service_request_expires_at ON connections( service_request_expires_at ); +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes( + conn_id, + hash +); +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/src/Simplex/Messaging/Crypto/Ratchet.hs b/src/Simplex/Messaging/Crypto/Ratchet.hs index fe74d772a..067fedd89 100644 --- a/src/Simplex/Messaging/Crypto/Ratchet.hs +++ b/src/Simplex/Messaging/Crypto/Ratchet.hs @@ -72,6 +72,7 @@ module Simplex.Messaging.Crypto.Ratchet rcEncryptHeader, rcEncryptMsg, rcDecrypt, + maxSkippedMsgKeys, -- used in tests MsgHeader (..), RatchetInitParams (..), @@ -962,6 +963,9 @@ type DecryptResult a = (Either CryptoError ByteString, Ratchet a, SkippedMsgDiff maxSkip :: Word32 maxSkip = 512 +maxSkippedMsgKeys :: Int +maxSkippedMsgKeys = 2000 + rcDecrypt :: forall a. (AlgorithmI a, DhAlgorithm a) => diff --git a/tests/AgentTests/ConnectionRequestTests.hs b/tests/AgentTests/ConnectionRequestTests.hs index d76d8d305..40ca12cf9 100644 --- a/tests/AgentTests/ConnectionRequestTests.hs +++ b/tests/AgentTests/ConnectionRequestTests.hs @@ -362,6 +362,10 @@ connectionRequestTests = smpEncodingTest $ AgentServiceRequest [qInfo] Nothing "service request payload" smpEncodingTest $ AgentServiceResponse "service response payload" smpEncodingTest $ AgentRejection "rejected: not allowed" + it "should serialize and parse ratchet key info" $ do + smpDecode "R" `shouldBe` Right (AgentRatchetInfo RatchetInfo {answeredKeyHash = Nothing}) + smpDecode "R1\3abcdef" `shouldBe` Right (AgentRatchetInfo RatchetInfo {answeredKeyHash = Just "abc"}) + smpEncodingTest $ AgentRatchetInfo RatchetInfo {answeredKeyHash = Just "0123456789abcdef0123456789abcdef"} where smpEncodingTest :: (Encoding a, Eq a, Show a, HasCallStack) => a -> Expectation smpEncodingTest a = smpDecode (smpEncode a) `shouldBe` Right a diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 8194e9ccd..727c47d5a 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -70,8 +70,8 @@ import Data.ByteString.Char8 (ByteString) 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 (find, isInfixOf, isPrefixOf, isSuffixOf) +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,16 @@ 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 + it "should synchronize ratchets when synchronization is forced again" $ + testRatchetSyncRepeated ps + it "should not mark ratchet key as processed when ratchet recreation fails" $ + testRatchetSyncFailedKeyNotProcessed ps + it "should not store reply ratchet key when ratchet recreation fails" $ + testRatchetSyncFailedRecreationNoReply ps + it "should not store ratchet key when starting synchronization fails" $ + testRatchetSyncStartFailedNoKey ps #endif describe "Subscription mode OnlyCreate" $ do it "messages delivered only when polled" $ @@ -2748,6 +2758,117 @@ 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 + +testRatchetSyncRepeated :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncRepeated ps = withAgentClients2 $ \alice bob -> do + (aliceId, bobId, bob2) <- startRatchetSyncOffline ps alice bob + ConnectionStats {ratchetSyncState = rss2} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn True + rss2 `shouldBe` RSStarted + + withSmpServerStoreMsgLogOn ps testPort $ \_ -> do + concurrently_ + (getInAnyOrder alice [ratchetSyncP' bobId RSAgreed, serverUpP]) + (getInAnyOrder bob2 [ratchetSyncP' aliceId RSAgreed, serverUpP]) + runRight_ $ do + get alice =##> ratchetSyncP bobId RSAgreed + get alice =##> ratchetSyncP bobId RSOk + get bob2 =##> ratchetSyncP aliceId RSOk + msgId <- sendMessage alice bobId SMP.noMsgFlags "hello" + get alice ##> ("", bobId, SENT msgId) + get bob2 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False + ackMessage bob2 aliceId 8 Nothing + map fst <$> processedRatchetKeyHashes bob2 `shouldReturn` [aliceId, aliceId] + disposeAgentClient bob2 + +testRatchetSyncFailedKeyNotProcessed :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncFailedKeyNotProcessed ps = withAgentClients2 $ \alice bob -> do + (aliceId, bobId, bob2) <- startRatchetSyncOffline ps alice bob + withTransaction (store $ agentEnv bob2) $ \db -> + DB.execute_ db "UPDATE ratchets SET x3dh_priv_key_1 = NULL" + + withSmpServerStoreMsgLogOn ps testPort $ \_ -> + concurrently_ + (getInAnyOrder alice [ratchetSyncP' bobId RSAgreed, serverUpP]) + (getInAnyOrder bob2 [x3dhKeysNotFoundP aliceId, serverUpP]) + map fst <$> processedRatchetKeyHashes alice `shouldReturn` [bobId] + processedRatchetKeyHashes bob2 `shouldReturn` [] + disposeAgentClient bob2 + where + x3dhKeysNotFoundP :: ConnId -> ATransmission -> Bool + x3dhKeysNotFoundP cId = \case + (_, cId', AEvt SAEConn (ERR (A.INTERNAL e))) -> cId' == cId && "SEX3dhKeysNotFound" `isPrefixOf` e + _ -> False + +testRatchetSyncFailedRecreationNoReply :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncFailedRecreationNoReply ps = withAgentClients2 $ \alice bob -> do + (_, bobId, bob2) <- startRatchetSyncOffline ps alice bob + aliceSndMsgs <- sndMessages alice + withTransaction (store $ agentEnv alice) $ \db -> + DB.execute_ db "CREATE TRIGGER fail_ratchet_insert BEFORE INSERT ON ratchets BEGIN SELECT RAISE(ABORT, 'ratchet insert failed'); END" + withSmpServerStoreMsgLogOn ps testPort $ \_ -> + concurrently_ + (getInAnyOrder alice [ratchetInsertFailedP bobId, serverUpP]) + (getInAnyOrder bob2 [serverUpP]) + processedRatchetKeyHashes alice `shouldReturn` [] + sndMessages alice `shouldReturn` aliceSndMsgs + disposeAgentClient bob2 + where + ratchetInsertFailedP :: ConnId -> ATransmission -> Bool + ratchetInsertFailedP cId = \case + (_, cId', AEvt SAEConn (ERR (A.INTERNAL e))) -> cId' == cId && "ratchet insert failed" `isInfixOf` e + _ -> False + +testRatchetSyncStartFailedNoKey :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncStartFailedNoKey ps = withAgentClients2 $ \alice bob -> do + (aliceId, _, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ -> + setupDesynchronizedRatchet alice bob + bobSndMsgs <- sndMessages bob2 + withTransaction (store $ agentEnv bob2) $ \db -> + DB.execute_ db "CREATE TRIGGER fail_ratchet_update BEFORE UPDATE ON ratchets BEGIN SELECT RAISE(ABORT, 'ratchet update failed'); END" + Left (A.INTERNAL e) <- runExceptT $ synchronizeRatchet bob2 aliceId PQSupportOff False + e `shouldContain` "ratchet update failed" + ConnectionStats {ratchetSyncState} <- runRight $ getConnectionServers bob2 aliceId + ratchetSyncState `shouldBe` RSRequired + withTransaction (store $ agentEnv bob2) (`DB.query_` "SELECT conn_id, pq_support FROM connections") `shouldReturn` [(aliceId, PQSupportOn)] + sndMessages bob2 `shouldReturn` bobSndMsgs + disposeAgentClient bob2 + +startRatchetSyncOffline :: HasCallStack => (ASrvTransport, AStoreType) -> AgentClient -> AgentClient -> IO (ConnId, ConnId, AgentClient) +startRatchetSyncOffline ps alice bob = do + (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ -> + setupDesynchronizedRatchet alice bob + ("", "", DOWN _ _) <- nGet alice + ("", "", DOWN _ _) <- nGet bob2 + ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn False + ratchetSyncState `shouldBe` RSStarted + pure (aliceId, bobId, bob2) + +processedRatchetKeyHashes :: AgentClient -> IO [(ConnId, ByteString)] +processedRatchetKeyHashes c = withTransaction (store $ agentEnv c) (`DB.query_` "SELECT conn_id, hash FROM processed_ratchet_key_hashes") + +sndMessages :: AgentClient -> IO [(ConnId, Int64)] +sndMessages c = withTransaction (store $ agentEnv c) (`DB.query_` "SELECT conn_id, internal_id FROM snd_messages ORDER BY conn_id, internal_id") + 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 c22eddd7a..8a5fd14d7 100644 --- a/tests/AgentTests/SQLiteTests.hs +++ b/tests/AgentTests/SQLiteTests.hs @@ -20,12 +20,13 @@ import Control.Concurrent.Async (concurrently_) import Control.Concurrent.MVar import Control.Concurrent.STM import Control.Exception (SomeException) -import Control.Monad (replicateM_) +import Control.Monad (forM_, replicateM_) import Control.Monad.Trans.Except import Crypto.Random (ChaChaDRG) import Data.ByteArray (ScrubbedBytes) import Data.ByteString.Char8 (ByteString) import Data.List (isInfixOf) +import qualified Data.Map.Strict as M import qualified Data.Text as T import Data.Text.Encoding (encodeUtf8) import Data.Time @@ -51,6 +52,7 @@ import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Crypto.File (CryptoFile (..)) import Simplex.Messaging.Crypto.Ratchet (pattern IKPQOn) import qualified Simplex.Messaging.Crypto.Ratchet as CR +import Simplex.Messaging.Encoding (Encoding (..)) import Simplex.Messaging.Encoding.String (StrEncoding (..)) import Simplex.Messaging.Protocol (EntityId (..), QueueMode (..), SubscriptionMode (..), pattern VersionSMPC) import qualified Simplex.Messaging.Protocol as SMP @@ -135,6 +137,8 @@ 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 it "should getPendingServerCommand" testGetPendingServerCommand @@ -599,6 +603,42 @@ 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 + withTransaction st $ \db -> do + Right connId <- createNewConn db g cData1 {connId = ""} SCMInvitation + Right connId' <- createNewConn db g cData1 {connId = ""} SCMInvitation + createSkippedKeys db connId' + createSkippedKeys db connId + M.map M.keys <$> getSkippedMsgKeys db connId 4 + `shouldReturn` M.singleton (C.Key "header_key") [1, 2, 3, 4] + getMsgNs db connId `shouldReturn` [1 .. 4] + getMsgNs db connId' `shouldReturn` [1 .. 10] + where + createSkippedKeys :: DB.Connection -> ConnId -> IO () + createSkippedKeys db connId = do + DB.execute db "INSERT INTO ratchets (conn_id) VALUES (?)" (Only connId) + forM_ ([10, 9 .. 1] :: [Int]) $ \msgN -> + DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, "header_key" :: ByteString, msgN, smpEncode ("key" :: ByteString, "iv" :: ByteString)) + getMsgNs :: DB.Connection -> ConnId -> IO [Int] + getMsgNs db connId = map fromOnly <$> DB.query db "SELECT msg_n FROM skipped_messages WHERE conn_id = ? ORDER BY msg_n" (Only connId) + testCloseReopenStore :: IO () testCloseReopenStore = do st <- createStore' diff --git a/tests/AgentTests/SchemaDump.hs b/tests/AgentTests/SchemaDump.hs index d9aa79513..ee401bf19 100644 --- a/tests/AgentTests/SchemaDump.hs +++ b/tests/AgentTests/SchemaDump.hs @@ -48,6 +48,7 @@ schemaDumpTest = do it "verify strict tables" testVerifyStrict it "should NOT create user record for new database" testUsersMigrationNew it "should create user record for old database" testUsersMigrationOld + it "should remove duplicate ratchet key hashes before adding unique index" testRatchetKeyHashesUniqueMigration testVerifySchemaDump :: IO () testVerifySchemaDump = do @@ -114,12 +115,28 @@ testUsersMigrationOld = do `shouldReturn` ([Only (1 :: Int)]) closeDBStore st' +testRatchetKeyHashesUniqueMigration :: IO () +testRatchetKeyHashesUniqueMigration = do + let beforeUnique = takeWhile (("m20260929_ratchet_indexes" /=) . name) appMigrations + Right st <- createDBStore (DBOpts testDB [] "" False True TQOff) beforeUnique (MigrationConfig MCError Nothing) + withTransaction' st $ \db -> do + SQL.execute_ db "INSERT INTO users (user_id) VALUES (1)" + SQL.execute_ db "INSERT INTO connections (conn_id, conn_mode, user_id) VALUES (x'01', 'INV', 1), (x'02', 'INV', 1)" + SQL.execute_ db "INSERT INTO processed_ratchet_key_hashes (conn_id, hash) VALUES (x'01', x'aa'), (x'01', x'aa'), (x'01', x'bb'), (x'02', x'aa'), (x'01', x'aa')" + closeDBStore st + Right st' <- createDBStore (DBOpts testDB [] "" False True TQOff) appMigrations (MigrationConfig MCYesUp Nothing) + withTransaction' st' (`SQL.query_` "SELECT processed_ratchet_key_hash_id FROM processed_ratchet_key_hashes ORDER BY processed_ratchet_key_hash_id") + `shouldReturn` [Only (1 :: Int), Only 3, Only 4] + closeDBStore st' + skipComparisonForDownMigrations :: [String] skipComparisonForDownMigrations = [ -- on down migration idx_messages_internal_snd_id_ts index moves down to the end of the file "m20230814_indexes", -- snd_secure and last_broker_ts columns swap order on down migration - "m20250322_short_links" + "m20250322_short_links", + -- on down migration idx_processed_ratchet_key_hashes_hash index moves down to the end of the file + "m20260929_ratchet_indexes" ] getSchema :: FilePath -> FilePath -> IO String