From 88f1b727e0e4bfef01a346d818b7f6d94a181b16 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Thu, 30 May 2024 18:49:43 +0100 Subject: [PATCH 1/4] SMP protocol extension to debug subscribed SMP queues (#1181) * SMP protocol extension to debug subscribed SMP queues * fix, test * corrections Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> * increase delays * increase timeout * delay * delay * enable all tests --------- Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> --- simplexmq.cabal | 1 + src/Simplex/Messaging/Agent.hs | 16 +++ src/Simplex/Messaging/Agent/Client.hs | 7 ++ src/Simplex/Messaging/Client.hs | 8 ++ src/Simplex/Messaging/Protocol.hs | 25 ++++- src/Simplex/Messaging/Server.hs | 24 ++++- .../Messaging/Server/QueueStore/QueueInfo.hs | 55 +++++++++++ src/Simplex/RemoteControl/Invitation.hs | 2 +- tests/AgentTests/FunctionalAPITests.hs | 98 ++++++++++++++++++- tests/CLITests.hs | 4 +- 10 files changed, 232 insertions(+), 8 deletions(-) create mode 100644 src/Simplex/Messaging/Server/QueueStore/QueueInfo.hs diff --git a/simplexmq.cabal b/simplexmq.cabal index 829fd23df..31dfff436 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -172,6 +172,7 @@ library Simplex.Messaging.Server.MsgStore Simplex.Messaging.Server.MsgStore.STM Simplex.Messaging.Server.QueueStore + Simplex.Messaging.Server.QueueStore.QueueInfo Simplex.Messaging.Server.QueueStore.STM Simplex.Messaging.Server.Stats Simplex.Messaging.Server.StoreLog diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 76d7d333d..073660ba0 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -70,6 +70,7 @@ module Simplex.Messaging.Agent sendMessages, sendMessagesB, ackMessage, + getConnectionQueueInfo, switchConnection, abortConnectionSwitch, synchronizeRatchet, @@ -176,6 +177,7 @@ import Simplex.Messaging.Notifications.Types import Simplex.Messaging.Parsers (parse) import Simplex.Messaging.Protocol (BrokerMsg, Cmd (..), EntityId, ErrorType (AUTH), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth, ProtocolTypeI (..), SMPMsgMeta, SParty (..), SProtocolType (..), SndPublicAuthKey, SubscriptionMode (..), UserProtocol, VersionSMPC, XFTPServerWithAuth) import qualified Simplex.Messaging.Protocol as SMP +import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.ServiceScheme (ServiceScheme (..)) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport (SMPVersion, THandleParams (sessionId)) @@ -371,6 +373,10 @@ ackMessage :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AE ackMessage c = withAgentEnv c .:. ackMessage' c {-# INLINE ackMessage #-} +getConnectionQueueInfo :: AgentClient -> ConnId -> AE QueueInfo +getConnectionQueueInfo c = withAgentEnv c . getConnectionQueueInfo' c +{-# INLINE getConnectionQueueInfo #-} + -- | Switch connection to the new receive queue switchConnection :: AgentClient -> ConnId -> AE ConnectionStats switchConnection c = withAgentEnv c . switchConnection' c @@ -1510,6 +1516,16 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do withStore' c $ \db -> deleteDeliveredSndMsg db connId $ InternalId sndMsgId _ -> pure () +getConnectionQueueInfo' :: AgentClient -> ConnId -> AM QueueInfo +getConnectionQueueInfo' c connId = do + SomeConn _ conn <- withStore c (`getConn` connId) + case conn of + DuplexConnection _ (rq :| _) _ -> getQueueInfo c rq + RcvConnection _ rq -> getQueueInfo c rq + ContactConnection _ rq -> getQueueInfo c rq + SndConnection {} -> throwE $ CONN SIMPLEX + NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionQueueInfo': NewConnection" + switchConnection' :: AgentClient -> ConnId -> AM ConnectionStats switchConnection' c connId = withConnLock c connId "switchConnection" $ diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index bc2a41ae2..e63ab5d1d 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -56,6 +56,7 @@ module Simplex.Messaging.Agent.Client disableQueueNotifications, disableQueuesNtfs, sendAgentMessage, + getQueueInfo, agentNtfRegisterToken, agentNtfVerifyToken, agentNtfCheckToken, @@ -238,6 +239,7 @@ import Simplex.Messaging.Protocol sameSrvAddr', ) import qualified Simplex.Messaging.Protocol as SMP +import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.Session import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM @@ -1577,6 +1579,11 @@ sendAgentMessage c sq@SndQueue {userId, server, sndId, sndPrivateKey} msgFlags a msg <- agentCbEncrypt sq Nothing $ smpEncode clientMsg sendOrProxySMPMessage c userId server "" (Just sndPrivateKey) sndId msgFlags msg +getQueueInfo :: AgentClient -> RcvQueue -> AM QueueInfo +getQueueInfo c rq@RcvQueue {rcvId, rcvPrivateKey} = + withSMPClient c rq "QUE" $ \smp -> + getSMPQueueInfo smp rcvPrivateKey rcvId + agentNtfRegisterToken :: AgentClient -> NtfToken -> NtfPublicAuthKey -> C.PublicKeyX25519 -> AM (NtfTokenId, C.PublicKeyX25519) agentNtfRegisterToken c NtfToken {deviceToken, ntfServer, ntfPrivKey} ntfPubKey pubDhKey = withClient c (0, ntfServer, Nothing) "TNEW" $ \ntf -> ntfRegisterToken ntf ntfPrivKey (NewNtfTkn deviceToken ntfPubKey pubDhKey) diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 8c23db822..79838792e 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -59,6 +59,7 @@ module Simplex.Messaging.Client connectSMPProxiedRelay, proxySMPMessage, forwardSMPMessage, + getSMPQueueInfo, sendProtocolCommand, -- * Supporting types and client configuration @@ -128,6 +129,7 @@ import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON) import Simplex.Messaging.Protocol +import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport @@ -920,6 +922,12 @@ forwardSMPMessage c@ProtocolClient {thParams, client_ = PClient {clientCorrId = pure fwdResponse r -> throwE $ unexpectedResponse r +getSMPQueueInfo :: SMPClient -> C.APrivateAuthKey -> QueueId -> ExceptT SMPClientError IO QueueInfo +getSMPQueueInfo c pKey qId = + sendSMPCommand c (Just pKey) qId QUE >>= \case + INFO info -> pure info + r -> throwE $ unexpectedResponse r + okSMPCommand :: PartyI p => Command p -> SMPClient -> C.APrivateAuthKey -> QueueId -> ExceptT SMPClientError IO () okSMPCommand cmd c pKey qId = sendSMPCommand c (Just pKey) qId cmd >>= \case diff --git a/src/Simplex/Messaging/Protocol.hs b/src/Simplex/Messaging/Protocol.hs index 1812dce37..5edea1719 100644 --- a/src/Simplex/Messaging/Protocol.hs +++ b/src/Simplex/Messaging/Protocol.hs @@ -156,6 +156,7 @@ module Simplex.Messaging.Protocol sameSrvAddr, sameSrvAddr', noAuthSrv, + toMsgInfo, -- * TCP transport functions TransportBatch (..), @@ -197,8 +198,8 @@ import qualified Data.List.NonEmpty as L import Data.Maybe (isJust, isNothing) import Data.String import qualified Data.Text as T -import Data.Text.Encoding (encodeUtf8) -import Data.Time.Clock.System (SystemTime (..)) +import Data.Text.Encoding (decodeLatin1, encodeUtf8) +import Data.Time.Clock.System (SystemTime (..), systemToUTCTime) import Data.Type.Equality import Data.Word (Word16) import qualified Data.X509 as X @@ -210,6 +211,7 @@ import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers +import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.ServiceScheme import Simplex.Messaging.Transport import Simplex.Messaging.Transport.Client (TransportHost, TransportHosts (..)) @@ -386,6 +388,7 @@ data Command (p :: Party) where ACK :: MsgId -> Command Recipient OFF :: Command Recipient DEL :: Command Recipient + QUE :: Command Recipient -- SMP sender commands -- SEND v1 has to be supported for encoding/decoding -- SEND :: MsgBody -> Command Sender @@ -463,6 +466,7 @@ data BrokerMsg where RRES :: EncFwdResponse -> BrokerMsg -- relay to proxy PRES :: EncResponse -> BrokerMsg -- proxy to client END :: BrokerMsg + INFO :: QueueInfo -> BrokerMsg OK :: BrokerMsg ERR :: ErrorType -> BrokerMsg PONG :: BrokerMsg @@ -505,6 +509,13 @@ data Message msgTs :: SystemTime } +toMsgInfo :: Message -> MsgInfo +toMsgInfo = \case + Message {msgId, msgTs} -> msgInfo msgId msgTs MTMessage + MessageQuota {msgId, msgTs} -> msgInfo msgId msgTs MTQuota + where + msgInfo msgId msgTs msgType = MsgInfo {msgId = decodeLatin1 $ B64.encode msgId, msgTs = systemToUTCTime msgTs, msgType} + messageId :: Message -> MsgId messageId = \case Message {msgId} -> msgId @@ -652,6 +663,7 @@ data CommandTag (p :: Party) where ACK_ :: CommandTag Recipient OFF_ :: CommandTag Recipient DEL_ :: CommandTag Recipient + QUE_ :: CommandTag Recipient SEND_ :: CommandTag Sender PING_ :: CommandTag Sender PRXY_ :: CommandTag ProxiedClient @@ -674,6 +686,7 @@ data BrokerMsgTag | RRES_ | PRES_ | END_ + | INFO_ | OK_ | ERR_ | PONG_ @@ -698,6 +711,7 @@ instance PartyI p => Encoding (CommandTag p) where ACK_ -> "ACK" OFF_ -> "OFF" DEL_ -> "DEL" + QUE_ -> "QUE" SEND_ -> "SEND" PING_ -> "PING" PRXY_ -> "PRXY" @@ -717,6 +731,7 @@ instance ProtocolMsgTag CmdTag where "ACK" -> Just $ CT SRecipient ACK_ "OFF" -> Just $ CT SRecipient OFF_ "DEL" -> Just $ CT SRecipient DEL_ + "QUE" -> Just $ CT SRecipient QUE_ "SEND" -> Just $ CT SSender SEND_ "PING" -> Just $ CT SSender PING_ "PRXY" -> Just $ CT SProxiedClient PRXY_ @@ -742,6 +757,7 @@ instance Encoding BrokerMsgTag where RRES_ -> "RRES" PRES_ -> "PRES" END_ -> "END" + INFO_ -> "INFO" OK_ -> "OK" ERR_ -> "ERR" PONG_ -> "PONG" @@ -757,6 +773,7 @@ instance ProtocolMsgTag BrokerMsgTag where "RRES" -> Just RRES_ "PRES" -> Just PRES_ "END" -> Just END_ + "INFO" -> Just INFO_ "OK" -> Just OK_ "ERR" -> Just ERR_ "PONG" -> Just PONG_ @@ -1275,6 +1292,7 @@ instance PartyI p => ProtocolEncoding SMPVersion ErrorType (Command p) where ACK msgId -> e (ACK_, ' ', msgId) OFF -> e OFF_ DEL -> e DEL_ + QUE -> e QUE_ SEND flags msg -> e (SEND_, ' ', flags, ' ', Tail msg) PING -> e PING_ NSUB -> e NSUB_ @@ -1340,6 +1358,7 @@ instance ProtocolEncoding SMPVersion ErrorType Cmd where ACK_ -> ACK <$> _smpP OFF_ -> pure OFF DEL_ -> pure DEL + QUE_ -> pure QUE CT SSender tag -> Cmd SSender <$> case tag of SEND_ -> SEND <$> _smpP <*> (unTail <$> _smpP) @@ -1368,6 +1387,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where RRES (EncFwdResponse encBlock) -> e (RRES_, ' ', Tail encBlock) PRES (EncResponse encBlock) -> e (PRES_, ' ', Tail encBlock) END -> e END_ + INFO info -> e (INFO_, ' ', info) OK -> e OK_ ERR err -> e (ERR_, ' ', err) PONG -> e PONG_ @@ -1388,6 +1408,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where RRES_ -> RRES <$> (EncFwdResponse . unTail <$> _smpP) PRES_ -> PRES <$> (EncResponse . unTail <$> _smpP) END_ -> pure END + INFO_ -> INFO <$> _smpP OK_ -> pure OK ERR_ -> ERR <$> _smpP PONG_ -> pure PONG diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 0d209229c..d34c0002c 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -59,7 +59,7 @@ import Data.List (intercalate, mapAccumR) import Data.List.NonEmpty (NonEmpty (..)) import qualified Data.List.NonEmpty as L import qualified Data.Map.Strict as M -import Data.Maybe (catMaybes, fromMaybe, isNothing) +import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing) import qualified Data.Text as T import Data.Text.Encoding (decodeLatin1) import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime) @@ -82,6 +82,7 @@ import Simplex.Messaging.Server.Expiration import Simplex.Messaging.Server.MsgStore import Simplex.Messaging.Server.MsgStore.STM import Simplex.Messaging.Server.QueueStore +import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.Server.QueueStore.STM as QS import Simplex.Messaging.Server.Stats import Simplex.Messaging.Server.StoreLog @@ -791,6 +792,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi NDEL -> deleteQueueNotifier_ st OFF -> suspendQueue_ st DEL -> delQueueAndMsgs st + QUE -> withQueue getQueueInfo where createQueue :: QueueStore -> RcvPublicAuthKey -> RcvPublicDhKey -> SubscriptionMode -> M (Transmission BrokerMsg) createQueue st recipientKey dhKey subMode = time "NEW" $ do @@ -1162,6 +1164,26 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi Right q -> updateDeletedStats q $> ok Left e -> pure $ err e + getQueueInfo :: QueueRec -> M (Transmission BrokerMsg) + getQueueInfo QueueRec {senderKey, notifier} = do + q@MsgQueue {size} <- getStoreMsgQueue "getQueueInfo" queueId + info <- atomically $ do + qiSub <- TM.lookup queueId subscriptions >>= mapM mkQSub + qiSize <- readTVar size + qiMsg <- toMsgInfo <$$> tryPeekMsg q + pure QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg} + pure (corrId, queueId, INFO info) + where + mkQSub sub = do + Sub {subThread, delivered} <- readTVar sub + let qSubThread = case subThread of + NoSub -> QNoSub + SubPending -> QSubPending + SubThread _ -> QSubThread + ProhibitSub -> QProhibitSub + qDelivered <- decodeLatin1 . encode <$$> tryReadTMVar delivered + pure QSub {qSubThread, qDelivered} + ok :: Transmission BrokerMsg ok = (corrId, queueId, OK) diff --git a/src/Simplex/Messaging/Server/QueueStore/QueueInfo.hs b/src/Simplex/Messaging/Server/QueueStore/QueueInfo.hs new file mode 100644 index 000000000..b329a54ff --- /dev/null +++ b/src/Simplex/Messaging/Server/QueueStore/QueueInfo.hs @@ -0,0 +1,55 @@ +{-# LANGUAGE TemplateHaskell #-} + +module Simplex.Messaging.Server.QueueStore.QueueInfo where + +import qualified Data.Aeson as J +import qualified Data.Aeson.TH as JQ +import qualified Data.Attoparsec.ByteString.Char8 as A +import qualified Data.ByteString.Lazy.Char8 as LB +import Data.Text (Text) +import Data.Time.Clock (UTCTime) +import Simplex.Messaging.Encoding +import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON) +import Simplex.Messaging.Util ((<$?>)) + +data QueueInfo = QueueInfo + { qiSnd :: Bool, + qiNtf :: Bool, + qiSub :: Maybe QSub, + qiSize :: Int, + qiMsg :: Maybe MsgInfo + } + deriving (Eq, Show) + +data QSub = QSub + { qSubThread :: QSubThread, + qDelivered :: Maybe Text + } + deriving (Eq, Show) + +data QSubThread = QNoSub | QSubPending | QSubThread | QProhibitSub + deriving (Eq, Show) + +data MsgInfo = MsgInfo + { msgId :: Text, + msgTs :: UTCTime, + msgType :: MsgType + } + deriving (Eq, Show) + +data MsgType = MTMessage | MTQuota + deriving (Eq, Show) + +$(JQ.deriveJSON (enumJSON $ dropPrefix "Q") ''QSubThread) + +$(JQ.deriveJSON defaultJSON ''QSub) + +$(JQ.deriveJSON (enumJSON $ dropPrefix "MT") ''MsgType) + +$(JQ.deriveJSON defaultJSON ''MsgInfo) + +$(JQ.deriveJSON defaultJSON ''QueueInfo) + +instance Encoding QueueInfo where + smpEncode = LB.toStrict . J.encode + smpP = J.eitherDecodeStrict <$?> A.takeByteString diff --git a/src/Simplex/RemoteControl/Invitation.hs b/src/Simplex/RemoteControl/Invitation.hs index 712c41a9d..d606c4ff0 100644 --- a/src/Simplex/RemoteControl/Invitation.hs +++ b/src/Simplex/RemoteControl/Invitation.hs @@ -123,7 +123,7 @@ instance StrEncoding RCSignedInvitation where idsig <- requiredP sigs "idsig" $ parseAll strP pure RCSignedInvitation {invitation, ssig, idsig} -signInvitation :: C.PrivateKey C.Ed25519 -> C.PrivateKey C.Ed25519 -> RCInvitation -> RCSignedInvitation +signInvitation :: C.PrivateKey 'C.Ed25519 -> C.PrivateKey 'C.Ed25519 -> RCInvitation -> RCSignedInvitation signInvitation sKey idKey invitation = RCSignedInvitation {invitation, ssig, idsig} where uri = strEncode invitation diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 2b21ff3f7..c2badea63 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -57,6 +57,7 @@ import Control.Monad import Control.Monad.Except import Control.Monad.Reader import Data.Bifunctor (first) +import qualified Data.ByteString.Base64 as B64 import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import Data.Either (isRight) @@ -66,6 +67,7 @@ import Data.List.NonEmpty (NonEmpty) import qualified Data.Map as M import Data.Maybe (isJust, isNothing) import qualified Data.Set as S +import Data.Text.Encoding (decodeLatin1) import Data.Time.Clock (diffUTCTime, getCurrentTime) import Data.Time.Clock.System (SystemTime (..), getSystemTime) import Data.Type.Equality (testEquality, (:~:) (Refl)) @@ -92,6 +94,7 @@ import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolS import qualified Simplex.Messaging.Protocol as SMP import Simplex.Messaging.Server.Env.STM (ServerConfig (..)) import Simplex.Messaging.Server.Expiration +import Simplex.Messaging.Server.QueueStore.QueueInfo import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion, supportedSMPHandshakes) import Simplex.Messaging.Util (diffToMicroseconds) import Simplex.Messaging.Version (VersionRange (..)) @@ -107,8 +110,11 @@ type AEntityTransmission e = (ACorrId, ConnId, ACommand 'Agent e) -- deriving instance Eq (ValidFileDescription p) +shouldRespond :: (HasCallStack, MonadUnliftIO m, Eq a, Show a) => m a -> a -> m () +a `shouldRespond` r = withFrozenCallStack $ withTimeout a (`shouldBe` r) + (##>) :: (HasCallStack, MonadUnliftIO m) => m (AEntityTransmission e) -> AEntityTransmission e -> m () -a ##> t = withTimeout a (`shouldBe` t) +a ##> t = a `shouldRespond` t (=##>) :: (Show a, HasCallStack, MonadUnliftIO m) => m a -> (HasCallStack => a -> Bool) -> m () a =##> p = @@ -228,7 +234,7 @@ mkVersionRange :: Word16 -> Word16 -> VersionRange v mkVersionRange v1 v2 = V.mkVersionRange (Version v1) (Version v2) runRight_ :: (Eq e, Show e, HasCallStack) => ExceptT e IO () -> Expectation -runRight_ action = runExceptT action `shouldReturn` Right () +runRight_ action = withFrozenCallStack $ runExceptT action `shouldReturn` Right () runRight :: (Show e, HasCallStack) => ExceptT e IO a -> IO a runRight action = @@ -444,6 +450,9 @@ functionalAPITests t = do it "should wait for user network" testWaitForUserNetwork it "should not reset online to offline if happens too quickly" testDoNotResetOnlineToOffline it "should resume multiple threads" testResumeMultipleThreads + describe "SMP queue info" $ do + it "server should respond with queue and subscription information" $ + withSmpServer t testServerQueueInfo testBasicAuth :: ATransport -> Bool -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> IO Int testBasicAuth t allowNewQueues srv@(srvAuth, srvVersion) clnt1 clnt2 = do @@ -2755,6 +2764,91 @@ testResumeMultipleThreads = do where aCfg = agentCfg {userOfflineDelay = 0} +testServerQueueInfo :: IO () +testServerQueueInfo = do + withAgentClients2 $ \alice bob -> runRight_ $ do + (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe + liftIO $ threadDelay 200000 + checkEmptyQ alice bobId False + aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe + ("", _, CONF confId _ "bob's connInfo") <- get alice + liftIO $ threadDelay 200000 + checkEmptyQ alice bobId False + allowConnection alice bobId confId "alice's connInfo" + get alice ##> ("", bobId, CON) + get bob ##> ("", aliceId, INFO "alice's connInfo") + get bob ##> ("", aliceId, CON) + liftIO $ threadDelay 200000 + checkEmptyQ alice bobId True + checkEmptyQ bob aliceId True + let msgId = 4 + (msgId', PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello" + liftIO $ msgId' `shouldBe` msgId + get alice ##> ("", bobId, SENT msgId) + liftIO $ threadDelay 200000 + Just srvMsgId <- checkMsgQ bob aliceId 1 + get bob =##> \case + ("", c, MSG MsgMeta {integrity = MsgOk, broker = (smId, _), recipient = (mId, _), pqEncryption = PQEncOn} _ "hello") -> + c == aliceId && decodeLatin1 (B64.encode smId) == srvMsgId && mId == msgId + _ -> False + ackMessage bob aliceId msgId Nothing + liftIO $ threadDelay 200000 + checkEmptyQ bob aliceId True + (msgId1, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 1" + get alice ##> ("", bobId, SENT msgId1) + Just _ <- checkMsgQ bob aliceId 1 + (msgId2, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 2" + get alice ##> ("", bobId, SENT msgId2) + (msgId3, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 3" + get alice ##> ("", bobId, SENT msgId3) + (msgId4, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello 4" + get alice ##> ("", bobId, SENT msgId4) + Just _ <- checkMsgQ bob aliceId 4 + (msgId5, PQEncOn) <- A.sendMessage alice bobId PQEncOn SMP.noMsgFlags "hello: quota exceeded" + liftIO $ threadDelay 200000 + Just _ <- checkMsgQ bob aliceId 5 + get bob =##> \case ("", c, Msg' mId PQEncOn "hello 1") -> c == aliceId && mId == msgId1; _ -> False + ackMessage bob aliceId msgId1 Nothing + liftIO $ threadDelay 200000 + Just _ <- checkMsgQ bob aliceId 4 + get bob =##> \case ("", c, Msg' mId PQEncOn "hello 2") -> c == aliceId && mId == msgId2; _ -> False + ackMessage bob aliceId msgId2 Nothing + get bob =##> \case ("", c, Msg' mId PQEncOn "hello 3") -> c == aliceId && mId == msgId3; _ -> False + ackMessage bob aliceId msgId3 Nothing + liftIO $ threadDelay 200000 + Just _ <- checkMsgQ bob aliceId 2 + get bob =##> \case ("", c, Msg' mId PQEncOn "hello 4") -> c == aliceId && mId == msgId4; _ -> False + ackMessage bob aliceId msgId4 Nothing + liftIO $ threadDelay 200000 + Just _ <- checkMsgQ bob aliceId 1 -- the one that did not fit now accepted + get alice ##> ("", bobId, QCONT) + get alice ##> ("", bobId, SENT msgId5) + liftIO $ threadDelay 200000 + Just _srvMsgId <- checkQ bob aliceId True (Just QNoSub) 1 (Just MTMessage) + get bob =##> \case ("", c, Msg' mId PQEncOn "hello: quota exceeded") -> c == aliceId && mId == msgId5 + 1; _ -> False + ackMessage bob aliceId (msgId5 + 1) Nothing + liftIO $ threadDelay 200000 + checkEmptyQ bob aliceId True + pure () + where + checkEmptyQ c cId qiSnd' = do + r <- checkQ c cId qiSnd' (Just QSubThread) 0 Nothing + liftIO $ r `shouldBe` Nothing + checkMsgQ c cId qiSize' = do + r <- checkQ c cId True (Just QNoSub) qiSize' (Just MTMessage) + liftIO $ isJust r `shouldBe` True + pure r + checkQ c cId qiSnd' qiSubThread_ qiSize' msgType_ = do + QueueInfo {qiSnd, qiNtf, qiSub, qiSize, qiMsg} <- getConnectionQueueInfo c cId + liftIO $ do + qiSnd `shouldBe` qiSnd' + qiNtf `shouldBe` False + qSubThread <$> qiSub `shouldBe` qiSubThread_ + qiSize `shouldBe` qiSize' + msgId_ <- forM qiMsg $ \MsgInfo {msgId, msgType} -> msgId <$ (Just msgType `shouldBe` msgType_) + qDelivered <$> qiSub `shouldBe` Just msgId_ + pure msgId_ + noNetworkDelay :: AgentClient -> IO () noNetworkDelay a = do d <- waitNetwork a diff --git a/tests/CLITests.hs b/tests/CLITests.hs index 2193de5e9..1310665ee 100644 --- a/tests/CLITests.hs +++ b/tests/CLITests.hs @@ -77,13 +77,13 @@ smpServerTest storeLog basicAuth = do let certPath = cfgPath "server.crt" oldCrt@X.Certificate {} <- XF.readSignedObject certPath >>= \case - [cert] -> pure . X.signedObject $ X.getSigned cert + [cert'] -> pure . X.signedObject $ X.getSigned cert' _ -> error "bad crt format" r' <- lines <$> capture_ (withArgs ["cert"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ())) r' `shouldContain` ["Generated new server credentials"] newCrt <- XF.readSignedObject certPath >>= \case - [cert] -> pure . X.signedObject $ X.getSigned cert + [cert'] -> pure . X.signedObject $ X.getSigned cert' _ -> error "bad crt format after cert" X.certSignatureAlg oldCrt `shouldBe` X.certSignatureAlg newCrt X.certSubjectDN oldCrt `shouldBe` X.certSubjectDN newCrt From 6e4067dc0cfb726a86b83e2641531a2ef8da71eb Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Fri, 31 May 2024 09:16:00 +0100 Subject: [PATCH 2/4] add string encodings for SMPProxyMode and SMPProxyFallback --- src/Simplex/Messaging/Client.hs | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 79838792e..56ebf7a3f 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -291,6 +291,32 @@ data SMPProxyFallback | SPFProhibit -- prohibit direct connection to destination relay. deriving (Eq, Show) +instance StrEncoding SMPProxyMode where + strEncode = \case + SPMAlways -> "always" + SPMUnknown -> "unknown" + SPMUnprotected -> "unprotected" + SPMNever -> "never" + strP = + A.takeTill (== ' ') >>= \case + "always" -> pure SPMAlways + "unknown" -> pure SPMUnknown + "unprotected" -> pure SPMUnprotected + "never" -> pure SPMNever + _ -> fail "Invalid SMP proxy mode" + +instance StrEncoding SMPProxyFallback where + strEncode = \case + SPFAllow -> "yes" + SPFAllowProtected -> "protected" + SPFProhibit -> "no" + strP = + A.takeTill (== ' ') >>= \case + "yes" -> pure SPFAllow + "protected" -> pure SPFAllowProtected + "no" -> pure SPFProhibit + _ -> fail "Invalid SMP proxy fallback mode" + defaultNetworkConfig :: NetworkConfig defaultNetworkConfig = NetworkConfig From d12ea9205537a570f369fc493ce3e7891511f623 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Fri, 31 May 2024 09:47:47 +0100 Subject: [PATCH 3/4] agent: report correct errors from xftp handshake so they are treated as temporary (#1184) * agent: report correct errors from xftp handshake so they are treated as temporary * disable slow servers test * remove comments * all tests * remove duplicate functions --- package.yaml | 3 +++ src/Simplex/FileTransfer/Client.hs | 27 +++++++++++++++---------- src/Simplex/FileTransfer/Server.hs | 22 ++++++++++++++++++++ src/Simplex/FileTransfer/Server/Env.hs | 3 ++- src/Simplex/FileTransfer/Server/Main.hs | 3 ++- src/Simplex/Messaging/Agent.hs | 7 ------- tests/XFTPAgent.hs | 8 ++++++-- tests/XFTPClient.hs | 3 ++- 8 files changed, 53 insertions(+), 23 deletions(-) diff --git a/package.yaml b/package.yaml index 6bdb03e4d..7b9486ff4 100644 --- a/package.yaml +++ b/package.yaml @@ -87,6 +87,9 @@ flags: manual: True default: True +# cpp-options: +# - -Dslow_servers + when: - condition: flag(swift) cpp-options: diff --git a/src/Simplex/FileTransfer/Client.hs b/src/Simplex/FileTransfer/Client.hs index ff7742e67..2454a1d57 100644 --- a/src/Simplex/FileTransfer/Client.hs +++ b/src/Simplex/FileTransfer/Client.hs @@ -52,7 +52,7 @@ import Simplex.Messaging.Protocol RecipientId, SenderId, ) -import Simplex.Messaging.Transport (ALPN, THandleAuth (..), THandleParams (..), TransportError (..), TransportPeer (..), supportedParameters) +import Simplex.Messaging.Transport (ALPN, HandshakeError (..), THandleAuth (..), THandleParams (..), TransportError (..), TransportPeer (..), supportedParameters) import Simplex.Messaging.Transport.Client (TransportClientConfig, TransportHost, alpn) import Simplex.Messaging.Transport.HTTP2 import Simplex.Messaging.Transport.HTTP2.Client @@ -98,6 +98,12 @@ defaultXFTPClientConfig = clientALPN = Just supportedXFTPhandshakes } +http2XFTPClientError :: HTTP2ClientError -> XFTPClientError +http2XFTPClientError = \case + HCResponseTimeout -> PCEResponseTimeout + HCNetworkError -> PCENetworkError + HCIOError e -> PCEIOError e + getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient) getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, xftpNetworkConfig, serverVRange} disconnected = runExceptT $ do let username = proxyUsername transportSession @@ -116,8 +122,7 @@ getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, logDebug $ "Client negotiated handshake protocol: " <> tshow sessionALPN thParams@THandleParams {thVersion} <- case sessionALPN of Just "xftp/1" -> xftpClientHandshakeV1 serverVRange keyHash http2Client thParams0 - Nothing -> pure thParams0 - _ -> throwError $ PCETransportError TEVersion + _ -> pure thParams0 logDebug $ "Client negotiated protocol: " <> tshow thVersion let c = XFTPClient {http2Client, thParams, transportSession, config} atomically $ writeTVar clientVar $ Just c @@ -135,15 +140,15 @@ xftpClientHandshakeV1 serverVRange keyHash@(C.KeyHash kh) c@HTTP2Client {session getServerHandshake = do let helloReq = H.requestNoBody "POST" "/" [] HTTP2Response {respBody = HTTP2Body {bodyHead = shsBody}} <- - liftError' (const $ PCEResponseError HANDSHAKE) $ sendRequest c helloReq Nothing - liftHS . smpDecode =<< liftHS (C.unPad shsBody) + liftError' http2XFTPClientError $ sendRequest c helloReq Nothing + liftTransportErr (TEHandshake PARSE) . smpDecode =<< liftTransportErr TEBadBlock (C.unPad shsBody) processServerHandshake :: XFTPServerHandshake -> ExceptT XFTPClientError IO (VersionRangeXFTP, C.PublicKeyX25519) processServerHandshake XFTPServerHandshake {xftpVersionRange, sessionId = serverSessId, authPubKey = serverAuth} = do - unless (sessionId == serverSessId) $ throwError $ PCEResponseError SESSION + unless (sessionId == serverSessId) $ throwError $ PCETransportError TEBadSession case xftpVersionRange `compatibleVRange` serverVRange of Nothing -> throwError $ PCETransportError TEVersion Just (Compatible vr) -> - fmap (vr,) . liftHS $ do + fmap (vr,) . liftTransportErr (TEHandshake BAD_AUTH) $ do let (X.CertificateChain cert, exact) = serverAuth case cert of [_leaf, ca] | XV.Fingerprint kh == XV.getFingerprint ca X.HashSHA256 -> pure () @@ -152,11 +157,11 @@ xftpClientHandshakeV1 serverVRange keyHash@(C.KeyHash kh) c@HTTP2Client {session C.x509ToPublic (pubKey, []) >>= C.pubKey sendClientHandshake :: XFTPClientHandshake -> ExceptT XFTPClientError IO () sendClientHandshake chs = do - chs' <- liftHS $ C.pad (smpEncode chs) xftpBlockSize + chs' <- liftTransportErr TELargeMsg $ C.pad (smpEncode chs) xftpBlockSize let chsReq = H.requestBuilder "POST" "/" [] $ byteString chs' - HTTP2Response {respBody = HTTP2Body {bodyHead}} <- liftError' (const $ PCEResponseError HANDSHAKE) $ sendRequest c chsReq Nothing - unless (B.null bodyHead) $ throwError $ PCEResponseError HANDSHAKE - liftHS = liftEitherWith (const $ PCEResponseError HANDSHAKE) + HTTP2Response {respBody = HTTP2Body {bodyHead}} <- liftError' http2XFTPClientError $ sendRequest c chsReq Nothing + unless (B.null bodyHead) $ throwError $ PCETransportError TEBadBlock + liftTransportErr e = liftEitherWith (const $ PCETransportError e) closeXFTPClient :: XFTPClient -> IO () closeXFTPClient XFTPClient {http2Client} = closeHTTP2Client http2Client diff --git a/src/Simplex/FileTransfer/Server.hs b/src/Simplex/FileTransfer/Server.hs index ea18b4fdc..7eb65ced2 100644 --- a/src/Simplex/FileTransfer/Server.hs +++ b/src/Simplex/FileTransfer/Server.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} @@ -67,6 +68,9 @@ import Simplex.Messaging.Version import System.Exit (exitFailure) import System.FilePath (()) import System.IO (hPrint, hPutStrLn, universalNewlineMode) +#ifdef slow_servers +import System.Random (getStdRandom, randomR) +#endif import UnliftIO import UnliftIO.Concurrent (threadDelay) import UnliftIO.Directory (doesFileExist, removeFile, renameFile) @@ -138,6 +142,9 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira let authPubKey = (chain, C.signX509 serverSignKey $ C.publicToX509 k) let hs = XFTPServerHandshake {xftpVersionRange = xftpServerVRange, sessionId, authPubKey} shs <- encodeXftp hs +#ifdef slow_servers + lift randomDelay +#endif liftIO . sendResponse $ H.responseBuilder N.ok200 [] shs pure Nothing processClientHandshake pk = do @@ -151,6 +158,9 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira let auth = THAuthServer {serverPrivKey = pk, sessSecret' = Nothing} thParams = thParams0 {thAuth = Just auth, thVersion = v, thServerVRange = vr} atomically $ TM.insert sessionId (HandshakeAccepted thParams) sessions +#ifdef slow_servers + lift randomDelay +#endif liftIO . sendResponse $ H.responseNoBody N.ok200 [] pure Nothing Nothing -> throwError HANDSHAKE @@ -315,6 +325,9 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea where sendXFTPResponse (corrId, fId, resp) serverFile_ = do let t_ = xftpEncodeTransmission thParams (corrId, fId, resp) +#ifdef slow_servers + randomDelay +#endif liftIO $ sendResponse $ H.responseStreaming N.ok200 [] $ streamBody t_ where streamBody t_ send done = do @@ -329,6 +342,15 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea withFile filePath ReadMode $ \h -> sendEncFile h send sbState fileSize done +#ifdef slow_servers +randomDelay :: M () +randomDelay = do + d <- asks $ responseDelay . config + when (d > 0) $ do + pc <- getStdRandom (randomR (-200, 200)) + threadDelay $ (d * (1000 + pc)) `div` 1000 +#endif + data VerificationResult = VRVerified XFTPRequest | VRFailed verifyXFTPTransmission :: Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> XFTPFileId -> FileCmd -> M VerificationResult diff --git a/src/Simplex/FileTransfer/Server/Env.hs b/src/Simplex/FileTransfer/Server/Env.hs index 58c1393f3..f8a6bc996 100644 --- a/src/Simplex/FileTransfer/Server/Env.hs +++ b/src/Simplex/FileTransfer/Server/Env.hs @@ -69,7 +69,8 @@ data XFTPServerConfig = XFTPServerConfig logStatsStartTime :: Int64, serverStatsLogFile :: FilePath, serverStatsBackupFile :: Maybe FilePath, - transportConfig :: TransportServerConfig + transportConfig :: TransportServerConfig, + responseDelay :: Int } defaultInactiveClientExpiration :: ExpirationConfig diff --git a/src/Simplex/FileTransfer/Server/Main.hs b/src/Simplex/FileTransfer/Server/Main.hs index ee9a52d34..d31295cd2 100644 --- a/src/Simplex/FileTransfer/Server/Main.hs +++ b/src/Simplex/FileTransfer/Server/Main.hs @@ -185,7 +185,8 @@ xftpServerCLI cfgPath logPath = do defaultTransportServerConfig { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini, alpn = Just supportedXFTPhandshakes - } + }, + responseDelay = 0 } data CliCommand diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 073660ba0..8e9020b7d 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -83,8 +83,6 @@ module Simplex.Messaging.Agent testProtocolServer, setNtfServers, setNetworkConfig, - getNetworkConfig, - getNetworkConfig', setUserNetworkInfo, reconnectAllServers, registerNtfToken, @@ -434,11 +432,6 @@ setNetworkConfig c@AgentClient {useNetworkConfig} cfg' = do else True <$ (writeTVar useNetworkConfig $! (slowNetworkConfig cfg', cfg')) when changed $ reconnectAllServers c --- returns fast network config -getNetworkConfig :: AgentClient -> IO NetworkConfig -getNetworkConfig = getNetworkConfig' -{-# INLINE getNetworkConfig #-} - setUserNetworkInfo :: AgentClient -> UserNetworkInfo -> IO () setUserNetworkInfo c@AgentClient {userNetworkInfo, userNetworkUpdated} ni = withAgentEnv' c $ do ts' <- liftIO getCurrentTime diff --git a/tests/XFTPAgent.hs b/tests/XFTPAgent.hs index 4685d4815..e4cf3d704 100644 --- a/tests/XFTPAgent.hs +++ b/tests/XFTPAgent.hs @@ -46,7 +46,11 @@ import XFTPClient xftpAgentTests :: Spec xftpAgentTests = around_ testBracket . describe "agent XFTP API" $ do - it "should send and receive file" testXFTPAgentSendReceive + it "should send and receive file" $ withXFTPServer testXFTPAgentSendReceive + -- uncomment CPP option slow_servers and run hpack to run this test + xit "should send and receive file with slow server responses" $ + withXFTPServerCfg testXFTPServerConfig {responseDelay = 500000} $ + \_ -> testXFTPAgentSendReceive it "should send and receive with encrypted local files" testXFTPAgentSendReceiveEncrypted it "should send and receive large file with a redirect" testXFTPAgentSendReceiveRedirect it "should send and receive small file without a redirect" testXFTPAgentSendReceiveNoRedirect @@ -100,7 +104,7 @@ checkProgress (prev, expected) (progress, total) loop | otherwise = pure () testXFTPAgentSendReceive :: HasCallStack => IO () -testXFTPAgentSendReceive = withXFTPServer $ do +testXFTPAgentSendReceive = do filePath <- createRandomFile -- send file, delete snd file internally (rfd1, rfd2) <- withAgent 1 agentCfg initAgentServers testDB $ \sndr -> runRight $ do diff --git a/tests/XFTPClient.hs b/tests/XFTPClient.hs index 0152625a8..208b54dc5 100644 --- a/tests/XFTPClient.hs +++ b/tests/XFTPClient.hs @@ -124,7 +124,8 @@ testXFTPServerConfig_ alpn = logStatsStartTime = 0, serverStatsLogFile = "tests/tmp/xftp-server-stats.daily.log", serverStatsBackupFile = Nothing, - transportConfig = defaultTransportServerConfig {alpn} + transportConfig = defaultTransportServerConfig {alpn}, + responseDelay = 0 } testXFTPClientConfig :: XFTPClientConfig From 3a3a84c58c53313906339973adb3caa3fb602b10 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Fri, 31 May 2024 12:20:29 +0100 Subject: [PATCH 4/4] server: log proxy connection errors --- src/Simplex/Messaging/Server.hs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index d34c0002c..88e3bf4a3 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -714,6 +714,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi let own = isOwnServer a srv inc own pRequests inc own $ if temporaryClientError e then pErrorsConnect else pErrorsOther + logError $ "Error connecting: " <> decodeLatin1 (strEncode $ host srv) <> " " <> tshow e pure . ERR $ smpProxyError e where proxyResp smp =