From b329eb9a86b2c271ed264614c2161e59e6524191 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sun, 20 Aug 2023 18:45:16 +0100 Subject: [PATCH] agent: fix async JOIN (#827) * agent: fix async JOIN * refactor --- src/Simplex/Messaging/Agent.hs | 143 +++++++++++--------- src/Simplex/Messaging/Agent/Store/SQLite.hs | 2 +- tests/AgentTests/FunctionalAPITests.hs | 42 ++++++ tests/ServerTests.hs | 4 +- tests/XFTPAgent.hs | 4 +- 5 files changed, 127 insertions(+), 68 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index db090e5bf..d7a261734 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -223,7 +223,7 @@ createConnection c userId enableNtfs cMode clientData = withAgentEnv c $ newConn -- | Join SMP agent connection (JOIN command) joinConnection :: AgentErrorMonad m => AgentClient -> UserId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId -joinConnection c userId enableNtfs = withAgentEnv c .: joinConn c userId "" False enableNtfs +joinConnection c userId enableNtfs = withAgentEnv c .: joinConn c userId "" enableNtfs -- | Allow connection to continue after CONF notification (LET command) allowConnection :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m () @@ -425,7 +425,7 @@ processCommand :: forall m. AgentMonad m => AgentClient -> (EntityId, APartyCmd processCommand c (connId, APC e cmd) = second (APC e) <$> case cmd of NEW enableNtfs (ACM cMode) -> second (INV . ACR cMode) <$> newConn c userId connId enableNtfs cMode Nothing - JOIN enableNtfs (ACR _ cReq) connInfo -> (,OK) <$> joinConn c userId connId False enableNtfs cReq connInfo + JOIN enableNtfs (ACR _ cReq) connInfo -> (,OK) <$> joinConn c userId connId enableNtfs cReq connInfo LET confId ownCInfo -> allowConnection' c connId confId ownCInfo $> (connId, OK) ACPT invId ownCInfo -> (,OK) <$> acceptContact' c connId True invId ownCInfo RJCT invId -> rejectContact' c connId invId $> (connId, OK) @@ -581,16 +581,16 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData srv = do withStore' c $ \db -> createRatchetX3dhKeys db connId pk1 pk2 pure (connId, CRInvitationUri crData $ toVersionRangeT e2eRcvParams e2eEncryptVRange) -joinConn :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId -joinConn c userId connId asyncMode enableNtfs cReq cInfo = do +joinConn :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId +joinConn c userId connId enableNtfs cReq cInfo = do srv <- case cReq of CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _ -> getNextServer c userId [qServer q] _ -> getSMPServer c userId - joinConnSrv c userId connId asyncMode enableNtfs cReq cInfo srv + joinConnSrv c userId connId enableNtfs cReq cInfo srv -joinConnSrv :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServerWithAuth -> m ConnId -joinConnSrv c userId connId asyncMode enableNtfs (CRInvitationUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)} e2eRcvParamsUri) cInfo srv = do +startJoinInvitation :: AgentMonad m => UserId -> ConnId -> Bool -> ConnectionRequestUri 'CMInvitation -> m (Compatible Version, ConnData, SndQueue, CR.Ratchet 'C.X448, CR.E2ERatchetParams 'C.X448) +startJoinInvitation userId connId enableNtfs (CRInvitationUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)} e2eRcvParamsUri) = do AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config case ( qUri `compatibleVersion` smpClientVRange, e2eRcvParamsUri `compatibleVersion` e2eEncryptVRange, @@ -603,31 +603,29 @@ joinConnSrv c userId connId asyncMode enableNtfs (CRInvitationUri ConnReqUriData q <- newSndQueue userId "" qInfo let duplexHS = connAgentVersion /= 1 cData = ConnData {userId, connId, connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk} - connId' <- setUpConn asyncMode cData q rc - let sq = (q :: SndQueue) {connId = connId'} - cData' = (cData :: ConnData) {connId = connId'} - tryError (confirmQueue aVersion c cData' sq srv cInfo $ Just e2eSndParams) >>= \case - Right _ -> do - unless duplexHS . void $ enqueueMessage c cData' sq SMP.noMsgFlags HELLO - pure connId' - Left e -> do - -- possible improvement: recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md - unless asyncMode $ withStore' c (`deleteConn` connId') - throwError e - where - setUpConn True _ sq rc = - withStore c $ \db -> runExceptT $ do - void . ExceptT $ updateNewConnSnd db connId sq - liftIO $ createRatchet db connId rc - pure connId - setUpConn False cData sq rc = do - g <- asks idsDrg - withStore c $ \db -> runExceptT $ do - connId' <- ExceptT $ createSndConn db g cData sq - liftIO $ createRatchet db connId' rc - pure connId' + pure (aVersion, cData, q, rc, e2eSndParams) _ -> throwError $ AGENT A_VERSION -joinConnSrv c userId connId False enableNtfs (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)}) cInfo srv = do + +joinConnSrv :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServerWithAuth -> m ConnId +joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo srv = do + (aVersion, cData@ConnData {connAgentVersion}, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv + g <- asks idsDrg + connId' <- withStore c $ \db -> runExceptT $ do + connId' <- ExceptT $ createSndConn db g cData q + liftIO $ createRatchet db connId' rc + pure connId' + let sq = (q :: SndQueue) {connId = connId'} + cData' = (cData :: ConnData) {connId = connId'} + duplexHS = connAgentVersion /= 1 + tryError (confirmQueue aVersion c cData' sq srv cInfo $ Just e2eSndParams) >>= \case + Right _ -> do + unless duplexHS . void $ enqueueMessage c cData' sq SMP.noMsgFlags HELLO + pure connId' + Left e -> do + -- possible improvement: recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md + withStore' c (`deleteConn` connId') + throwError e +joinConnSrv c userId connId enableNtfs (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)}) cInfo srv = do aVRange <- asks $ smpAgentVRange . config clientVRange <- asks $ smpClientVRange . config case ( qUri `compatibleVersion` clientVRange, @@ -638,7 +636,15 @@ joinConnSrv c userId connId False enableNtfs (CRContactUri ConnReqUriData {crAge sendInvitation c userId qInfo vrsn cReq cInfo pure connId' _ -> throwError $ AGENT A_VERSION -joinConnSrv _c _userId _connId True _enableNtfs (CRContactUri _) _cInfo _srv = do + +joinConnSrvAsync :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServerWithAuth -> m () +joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo srv = do + (aVersion, cData, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv + withStore c $ \db -> runExceptT $ do + void . ExceptT $ updateNewConnSnd db connId q + liftIO $ createRatchet db connId rc + confirmQueueAsync aVersion c cData q srv cInfo $ Just e2eSndParams +joinConnSrvAsync _c _userId _connId _enableNtfs (CRContactUri _) _cInfo _srv = do throwError $ CMD PROHIBITED createReplyQueue :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> m SMPQueueInfo @@ -673,7 +679,7 @@ acceptContact' c connId enableNtfs invId ownConnInfo = withConnLock c connId "ac withStore c (`getConn` contactConnId) >>= \case SomeConn _ (ContactConnection ConnData {userId} _) -> do withStore' c $ \db -> acceptInvitation db invId ownConnInfo - joinConn c userId connId False enableNtfs connReq ownConnInfo `catchAgentError` \err -> do + joinConn c userId connId enableNtfs connReq ownConnInfo `catchAgentError` \err -> do withStore' c (`unacceptInvitation` invId) throwError err _ -> throwError $ CMD PROHIBITED @@ -890,7 +896,7 @@ runCommandProcessing c@AgentClient {subQ} server_ = do let initUsed = [qServer q] usedSrvs <- newTVarIO initUsed tryCommand . withNextSrv c userId usedSrvs initUsed $ \srv -> do - void $ joinConnSrv c userId connId True enableNtfs cReq connInfo srv + joinConnSrvAsync c userId connId enableNtfs cReq connInfo srv notify OK LET confId ownCInfo -> withServer' . tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK ACK msgId rcptInfo_ -> withServer' . tryCommand $ ackMessage' c connId msgId rcptInfo_ >> notify OK @@ -1087,6 +1093,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl withRetryLock2 ri' qLock $ \riState loop -> do resp <- tryError $ case msgType of AM_CONN_INFO -> sendConfirmation c sq msgBody + AM_CONN_INFO_REPLY -> sendConfirmation c sq msgBody _ -> sendAgentMessage c sq msgFlags msgBody case resp of Left e -> do @@ -1138,12 +1145,8 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl retrySndOp c $ loop riMode Right () -> do case msgType of - AM_CONN_INFO -> do - withStore' c $ \db -> do - setSndQueueStatus db sq Confirmed - when (isJust rq_) $ removeConfirmations db connId - unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO - AM_CONN_INFO_REPLY -> pure () + AM_CONN_INFO -> setConfirmed + AM_CONN_INFO_REPLY -> setConfirmed AM_RATCHET_INFO -> pure () AM_REPLY_ -> pure () AM_HELLO_ -> do @@ -1205,6 +1208,12 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl _ -> internalErr msgId "QTEST sent not in duplex connection" AM_EREADY_ -> pure () delMsgKeep (msgType == AM_A_MSG_) msgId + where + setConfirmed = do + withStore' c $ \db -> do + setSndQueueStatus db sq Confirmed + when (isJust rq_) $ removeConfirmations db connId + unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO where delMsg :: InternalId -> m () delMsg = delMsgKeep False @@ -2305,10 +2314,15 @@ connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo (qInfo :| _) = dbQueueId <- withStore c $ \db -> upgradeRcvConnToDuplex db connId sq enqueueConfirmation c cData sq {dbQueueId} ownConnInfo Nothing +confirmQueueAsync :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m () +confirmQueueAsync v c cData sq srv connInfo e2eEncryption_ = do + resumeMsgDelivery c cData sq + msgId <- storeConfirmation c cData sq e2eEncryption_ =<< mkAgentConfirmation v c cData sq srv connInfo + queuePendingMsgs c sq [msgId] + confirmQueue :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m () -confirmQueue (Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo e2eEncryption_ = do - aMessage <- mkAgentMessage agentVersion - msg <- mkConfirmation aMessage +confirmQueue v@(Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo e2eEncryption_ = do + msg <- mkConfirmation =<< mkAgentConfirmation v c cData sq srv connInfo sendConfirmation c sq msg withStore' c $ \db -> setSndQueueStatus db sq Confirmed where @@ -2317,32 +2331,33 @@ confirmQueue (Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo void . liftIO $ updateSndIds db connId encConnInfo <- agentRatchetEncrypt db connId (smpEncode aMessage) e2eEncConnInfoLength pure . smpEncode $ AgentConfirmation {agentVersion, e2eEncryption_, encConnInfo} - mkAgentMessage :: Version -> m AgentMessage - mkAgentMessage 1 = pure $ AgentConnInfo connInfo - mkAgentMessage _ = do + +mkAgentConfirmation :: AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> m AgentMessage +mkAgentConfirmation (Compatible agentVersion) c cData sq srv connInfo + | agentVersion == 1 = pure $ AgentConnInfo connInfo + | otherwise = do qInfo <- createReplyQueue c cData sq srv pure $ AgentConnInfoReply (qInfo :| []) connInfo -enqueueConfirmation :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m () -enqueueConfirmation c cData@ConnData {connId, connAgentVersion} sq connInfo e2eEncryption_ = do +enqueueConfirmation :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m () +enqueueConfirmation c cData sq connInfo e2eEncryption_ = do resumeMsgDelivery c cData sq - msgId <- storeConfirmation + msgId <- storeConfirmation c cData sq e2eEncryption_ $ AgentConnInfo connInfo queuePendingMsgs c sq [msgId] - where - storeConfirmation :: m InternalId - storeConfirmation = withStore c $ \db -> runExceptT $ do - internalTs <- liftIO getCurrentTime - (internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId - let agentMsg = AgentConnInfo connInfo - agentMsgStr = smpEncode agentMsg - internalHash = C.sha256Hash agentMsgStr - encConnInfo <- agentRatchetEncrypt db connId agentMsgStr e2eEncConnInfoLength - let msgBody = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption_, encConnInfo} - msgType = agentMessageType agentMsg - msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash} - liftIO $ createSndMsg db connId msgData - liftIO $ createSndMsgDelivery db connId sq internalId - pure internalId + +storeConfirmation :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> Maybe (CR.E2ERatchetParams 'C.X448) -> AgentMessage -> m InternalId +storeConfirmation c ConnData {connId, connAgentVersion} sq e2eEncryption_ agentMsg = withStore c $ \db -> runExceptT $ do + internalTs <- liftIO getCurrentTime + (internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId + let agentMsgStr = smpEncode agentMsg + internalHash = C.sha256Hash agentMsgStr + encConnInfo <- agentRatchetEncrypt db connId agentMsgStr e2eEncConnInfoLength + let msgBody = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption_, encConnInfo} + msgType = agentMessageType agentMsg + msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash} + liftIO $ createSndMsg db connId msgData + liftIO $ createSndMsgDelivery db connId sq internalId + pure internalId enqueueRatchetKeyMsgs :: forall m. AgentMonad m => AgentClient -> ConnData -> NonEmpty SndQueue -> CR.E2ERatchetParams 'C.X448 -> m AgentMsgId enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 046283bec..96ff2f883 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -1728,7 +1728,7 @@ insertSndQueue_ db connId' SndQueue {..} serverKeyHash_ = do DB.execute db [sql| - INSERT INTO snd_queues + INSERT OR REPLACE INTO snd_queues (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, snd_queue_id, snd_primary, replace_snd_queue_id, smp_client_version, server_key_hash) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?); |] ((host server, port server, sndId, connId', sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret) :. (status, qId, primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_)) diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index f9e8b4c71..46243b17b 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -231,6 +231,8 @@ functionalAPITests t = do withSmpServer t testAcceptContactAsync it "should delete connections using async command when server connection fails" $ testDeleteConnectionAsync t + it "join connection when reply queue creation fails" $ + testJoinConnectionAsyncReplyError t describe "Users" $ do it "should create and delete user with connections" $ withSmpServer t testUsers @@ -1307,6 +1309,46 @@ testDeleteConnectionAsync t = do get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False liftIO $ noMessages a "nothing else should be delivered to alice" +testJoinConnectionAsyncReplyError :: HasCallStack => ATransport -> IO () +testJoinConnectionAsyncReplyError t = do + let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]} + a <- getSMPAgentClient' agentCfg initAgentServers testDB + b <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2 + (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do + bId <- createConnectionAsync a 1 "1" True SCMInvitation + ("1", bId', INV (ACR _ qInfo)) <- get a + liftIO $ bId' `shouldBe` bId + aId <- joinConnectionAsync b 1 "2" True qInfo "bob's connInfo" + liftIO $ threadDelay 500000 + ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId + pure (aId, bId) + nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False + withSmpServerOn t testPort2 $ do + ("2", aId', OK) <- get b + liftIO $ aId' `shouldBe` aId + confId <- withSmpServerStoreLogOn t testPort $ \_ -> do + pGet a >>= \case + ("", "", APC _ (UP _ [_])) -> do + ("", _, CONF confId _ "bob's connInfo") <- get a + pure confId + ("", _, APC _ (CONF confId _ "bob's connInfo")) -> do + ("", "", UP _ [_]) <- nGet a + pure confId + r -> error $ "unexpected response " <> show r + nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False + runRight_ $ do + allowConnectionAsync a "3" bId confId "alice's connInfo" + liftIO $ threadDelay 500000 + ConnectionStats {rcvQueuesInfo = [RcvQueueInfo {}], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId + pure () + withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do + pGet a =##> \case ("3", c, APC _ OK) -> c == bId; ("", "", APC _ (UP _ [c])) -> c == bId; _ -> False + pGet a =##> \case ("3", c, APC _ OK) -> c == bId; ("", "", APC _ (UP _ [c])) -> c == bId; _ -> False + get a ##> ("", bId, CON) + get b ##> ("", aId, INFO "alice's connInfo") + get b ##> ("", aId, CON) + exchangeGreetings a bId b aId + testUsers :: IO () testUsers = do a <- getSMPAgentClient' agentCfg initAgentServers testDB diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 74b24bbc3..5c995ce6f 100644 --- a/tests/ServerTests.hs +++ b/tests/ServerTests.hs @@ -969,7 +969,9 @@ testMsgExpireOnInterval t = Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, _SEND "hello (should expire)") threadDelay 2500000 testSMPClient @c $ \rh -> do - Resp "2" _ OK <- signSendRecv rh rKey ("2", rId, SUB) + signSendRecv rh rKey ("2", rId, SUB) >>= \case + Resp "2" _ OK -> pure () + r -> error $ "unexpected response: " <> show r 1000 `timeout` tGet @ErrorType @BrokerMsg rh >>= \case Nothing -> return () Just _ -> error "nothing should be delivered" diff --git a/tests/XFTPAgent.hs b/tests/XFTPAgent.hs index abec05c90..ad7c81f56 100644 --- a/tests/XFTPAgent.hs +++ b/tests/XFTPAgent.hs @@ -15,7 +15,7 @@ import qualified Data.ByteString.Char8 as B import Data.Int (Int64) import Data.List (find, isSuffixOf) import Data.Maybe (fromJust) -import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2) +import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2, testDB3) import Simplex.FileTransfer.Description import Simplex.FileTransfer.Protocol (FileParty (..), XFTPErrorType (AUTH)) import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..)) @@ -372,7 +372,7 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do length <$> listDirectory xftpServerFiles `shouldReturn` 0 -- receive file - should fail with AUTH error - rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB2 + rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB3 runRight $ do xftpStartWorkers rcp2 (Just recipientFiles) rfId <- xftpReceiveFile rcp2 1 rfd2