secure queue asynchronously (#532)

* secure queue asynchronously

* delete commands with internal logic errors

src/Simplex/Messaging/Agent.hs

@@ -470,13 +470,14 @@ createReplyQueue c ConnData {connId, enableNtfs} SndQueue {smpClientVersion} srv
 allowConnection' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
 allowConnection' c connId confId ownConnInfo =
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (RcvConnection cData rq) -> do
-      AcceptedConfirmation {senderConf} <- withStore c $ \db -> runExceptT $ do
-        conf <- ExceptT $ acceptConfirmation db confId ownConnInfo
-        liftIO $ createRatchet db connId $ ratchetState (conf :: AcceptedConfirmation)
-        pure conf
-      processConfirmation c rq senderConf
-      mapM_ (connectReplyQueues c cData ownConnInfo) (L.nonEmpty $ smpReplyQueues senderConf)
+    SomeConn _ (RcvConnection _ rq@RcvQueue {server, rcvId, e2ePrivKey, smpClientVersion = v}) -> do
+      senderKey <- withStore c $ \db -> runExceptT $ do
+        AcceptedConfirmation {ratchetState, senderConf = SMPConfirmation {senderKey, e2ePubKey, smpClientVersion = v'}} <- ExceptT $ acceptConfirmation db confId ownConnInfo
+        liftIO $ createRatchet db connId ratchetState
+        let dhSecret = C.dh' e2ePubKey e2ePrivKey
+        liftIO $ setRcvQueueConfirmedE2E db rq dhSecret $ min v v'
+        pure senderKey
+      enqueueCommand c "" connId (Just server) . AInternalCommand $ ICAllowSecure rcvId senderKey
     _ -> throwError $ CMD PROHIBITED
 
 -- | Accept contact (ACPT command) in Reader monad
@@ -496,13 +497,6 @@ rejectContact' :: AgentMonad m => AgentClient -> ConnId -> InvitationId -> m ()
 rejectContact' c contactConnId invId =
   withStore c $ \db -> deleteInvitation db contactConnId invId
 
-processConfirmation :: AgentMonad m => AgentClient -> RcvQueue -> SMPConfirmation -> m ()
-processConfirmation c rq@RcvQueue {e2ePrivKey, smpClientVersion = v} SMPConfirmation {senderKey, e2ePubKey, smpClientVersion = v'} = do
-  let dhSecret = C.dh' e2ePubKey e2ePrivKey
-  withStore' c $ \db -> setRcvQueueConfirmedE2E db rq dhSecret $ min v v'
-  secureQueue c rq senderKey
-  withStore' c $ \db -> setRcvQueueStatus db rq Secured
-
 -- | Subscribe to receive connection messages (SUB command) in Reader monad
 subscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
 subscribeConnection' c connId =
@@ -717,25 +711,39 @@ runCommandProcessing c@AgentClient {subQ} server = do
         Just _srv -> case cmd of
           ICAckDel _rId srvMsgId msgId -> tryCommand $ ack _rId srvMsgId >> withStore' c (\db -> deleteMsg db connId msgId)
           ICAck _rId srvMsgId -> tryCommand $ ack _rId srvMsgId
-        _ -> notify $ ERR $ INTERNAL $ "command requires server " <> show (internalCmdTag cmd)
+          ICAllowSecure _rId senderKey -> tryCommand $ do
+            (SomeConn _ conn, AcceptedConfirmation {senderConf, ownConnInfo}) <-
+              withStore c $ \db -> runExceptT $ (,) <$> ExceptT (getConn db connId) <*> ExceptT (getAcceptedConfirmation db connId)
+            case conn of
+              RcvConnection cData rq -> do
+                secure rq senderKey
+                mapM_ (connectReplyQueues c cData ownConnInfo) (L.nonEmpty $ smpReplyQueues senderConf)
+              _ -> throwError $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
+          ICDuplexSecure _rId senderKey -> tryCommand $ do
+            SomeConn _ conn <- withStore c (`getConn` connId)
+            case conn of
+              DuplexConnection cData rq sq -> do
+                secure rq senderKey
+                when (duplexHandshake cData == Just True) . void $
+                  enqueueMessage c cData sq SMP.MsgFlags {notification = True} HELLO
+              _ -> throwError $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
+        _ -> throwError $ INTERNAL $ "command requires server " <> show (internalCmdTag cmd)
         where
           ack _rId srvMsgId = do
             -- TODO get particular queue
             rq <- withStore c (`getRcvQueue` connId)
             ackQueueMessage c rq srvMsgId
+          secure :: RcvQueue -> SMP.SndPublicVerifyKey -> m ()
+          secure rq senderKey = do
+            secureQueue c rq senderKey
+            withStore' c $ \db -> setRcvQueueStatus db rq Secured
       where
         tryCommand action = withRetryInterval ri $ \loop ->
           tryError action >>= \case
             Left e
-              | temporaryAgentError e || e == BROKER HOST -> retryCommand loop
+              | temporaryAgentError e || e == BROKER HOST -> retrySndOp c loop
               | otherwise -> notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
             Right () -> withStore' c (`deleteCommand` cmdId)
-        retryCommand loop = do
-          -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
-          atomically $ endAgentOperation c AOSndNetwork
-          atomically $ throwWhenInactive c
-          atomically $ beginAgentOperation c AOSndNetwork
-          loop
         notify cmd = atomically $ writeTBQueue subQ (corrId, connId, cmd)
     withNextSrv :: TVar [SMPServer] -> [SMPServer] -> (SMPServer -> m ()) -> m ()
     withNextSrv usedSrvs initUsed action = do
@@ -827,7 +835,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                 SMP SMP.QUOTA -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retrySending loop
+                  _ -> retrySndOp c loop
                 SMP SMP.AUTH -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -836,7 +844,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                     -- because the queue must be secured by the time the confirmation or the first HELLO is received
                     | duplexHandshake == Just True -> connErr
                     | otherwise ->
-                      ifM (msgExpired helloTimeout) connErr (retrySending loop)
+                      ifM (msgExpired helloTimeout) connErr (retrySndOp c loop)
                     where
                       connErr = case rq_ of
                         -- party initiating connection
@@ -850,7 +858,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                   -- the message sending would be retried
                   | temporaryAgentError e || e == BROKER HOST -> do
                     let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySending loop)
+                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndOp c loop)
                   | otherwise -> notifyDel msgId err
               where
                 msgExpired timeoutSel = do
@@ -898,12 +906,14 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
     notifyDel :: InternalId -> ACommand 'Agent -> m ()
     notifyDel msgId cmd = notify cmd >> delMsg msgId
     connError msgId = notifyDel msgId . ERR . CONN
-    retrySending loop = do
-      -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
-      atomically $ endAgentOperation c AOSndNetwork
-      atomically $ throwWhenInactive c
-      atomically $ beginAgentOperation c AOSndNetwork
-      loop
+
+retrySndOp :: AgentMonad m => AgentClient -> m () -> m ()
+retrySndOp c loop = do
+  -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
+  atomically $ endAgentOperation c AOSndNetwork
+  atomically $ throwWhenInactive c
+  atomically $ beginAgentOperation c AOSndNetwork
+  loop
 
 ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage' c connId msgId = do
@@ -1305,8 +1315,6 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
             ack = enqueueCmd $ ICAck rId srvMsgId
             ackDel :: InternalId -> m ()
             ackDel = enqueueCmd . ICAckDel rId srvMsgId
-            enqueueCmd :: InternalCommand -> m ()
-            enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand
             handleNotifyAck :: m () -> m ()
             handleNotifyAck m = m `catchError` \e -> notify (ERR e) >> ack
         SMP.END ->
@@ -1332,6 +1340,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
         prohibited :: m ()
         prohibited = notify . ERR $ AGENT A_PROHIBITED
 
+        enqueueCmd :: InternalCommand -> m ()
+        enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand
+
         decryptClientMessage :: C.DhSecretX25519 -> SMP.ClientMsgEnvelope -> m (SMP.PrivHeader, AgentMsgEnvelope)
         decryptClientMessage e2eDh SMP.ClientMsgEnvelope {cmNonce, cmEncBody} = do
           clientMsg <- agentCbDecrypt e2eDh cmNonce cmEncBody
@@ -1382,12 +1393,13 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                       queueServer (SMPQueueInfo _ SMPQueueAddress {smpServer}) = smpServer
                   _ -> prohibited
               -- party accepting connection
-              (DuplexConnection _ _ sq, Nothing) -> do
+              (DuplexConnection _ RcvQueue {smpClientVersion = v'} _, Nothing) -> do
                 withStore c (\db -> runExceptT $ agentRatchetDecrypt db connId encConnInfo) >>= parseMessage >>= \case
                   AgentConnInfo connInfo -> do
                     notify $ INFO connInfo
-                    processConfirmation c rq $ SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = [], smpClientVersion}
-                    when (duplexHandshake == Just True) $ enqueueDuplexHello sq
+                    let dhSecret = C.dh' e2ePubKey e2ePrivKey
+                    withStore' c $ \db -> setRcvQueueConfirmedE2E db rq dhSecret $ min v' smpClientVersion
+                    enqueueCmd $ ICDuplexSecure rId senderKey
                   _ -> prohibited
               _ -> prohibited
             _ -> prohibited

src/Simplex/Messaging/Agent/Store.hs

@@ -207,29 +207,41 @@ instance StrEncoding AgentCommandTag where
 data InternalCommand
   = ICAck SMP.RecipientId MsgId
   | ICAckDel SMP.RecipientId MsgId InternalId
+  | ICAllowSecure SMP.RecipientId SMP.SndPublicVerifyKey
+  | ICDuplexSecure SMP.RecipientId SMP.SndPublicVerifyKey
 
 data InternalCommandTag
   = ICAck_
   | ICAckDel_
+  | ICAllowSecure_
+  | ICDuplexSecure_
   deriving (Show)
 
 instance StrEncoding InternalCommand where
   strEncode = \case
     ICAck rId srvMsgId -> strEncode (ICAck_, rId, srvMsgId)
     ICAckDel rId srvMsgId mId -> strEncode (ICAckDel_, rId, srvMsgId, mId)
+    ICAllowSecure rId sndKey -> strEncode (ICAllowSecure_, rId, sndKey)
+    ICDuplexSecure rId sndKey -> strEncode (ICDuplexSecure_, rId, sndKey)
   strP =
     strP_ >>= \case
       ICAck_ -> ICAck <$> strP_ <*> strP
       ICAckDel_ -> ICAckDel <$> strP_ <*> strP_ <*> strP
+      ICAllowSecure_ -> ICAllowSecure <$> strP_ <*> strP
+      ICDuplexSecure_ -> ICDuplexSecure <$> strP_ <*> strP
 
 instance StrEncoding InternalCommandTag where
   strEncode = \case
     ICAck_ -> "ACK"
     ICAckDel_ -> "ACK_DEL"
+    ICAllowSecure_ -> "ALLOW_SECURE"
+    ICDuplexSecure_ -> "DUPLEX_SECURE"
   strP =
     A.takeTill (== ' ') >>= \case
       "ACK" -> pure ICAck_
       "ACK_DEL" -> pure ICAckDel_
+      "ALLOW_SECURE" -> pure ICAllowSecure_
+      "DUPLEX_SECURE" -> pure ICDuplexSecure_
       _ -> fail "bad InternalCommandTag"
 
 agentCommandTag :: AgentCommand -> AgentCommandTag
@@ -241,6 +253,8 @@ internalCmdTag :: InternalCommand -> InternalCommandTag
 internalCmdTag = \case
   ICAck {} -> ICAck_
   ICAckDel {} -> ICAckDel_
+  ICAllowSecure {} -> ICAllowSecure_
+  ICDuplexSecure {} -> ICDuplexSecure_
 
 -- * Confirmation types