dandri/simplexmq
1
0 Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-10-23 06:32:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

track subscription responses better

Browse Source
This commit is contained in:
Evgeny Poberezkin committed 2024-05-04 17:12:02 +01:00
1 parent 6ee7082269
commit 4b5a88f407
19 files changed
+211 -165
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

package.yaml
+1
View File
@@ -180,6 +180,7 @@ ghc-options:
- -Wall
- -Wcompat
- -Werror=incomplete-patterns
- -Werror=missing-methods
- -Wredundant-constraints
- -Wincomplete-record-updates
- -Wincomplete-uni-patterns
simplexmq.cabal
+7 -7
View File
@@ -174,7 +174,7 @@ library
src
default-extensions:
StrictData
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
include-dirs:
cbits
c-sources:
@@ -255,7 +255,7 @@ executable ntf-server
apps/ntf-server
default-extensions:
StrictData
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
build-depends:
aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
@@ -330,7 +330,7 @@ executable smp-agent
apps/smp-agent
default-extensions:
StrictData
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
build-depends:
aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
@@ -405,7 +405,7 @@ executable smp-server
apps/smp-server
default-extensions:
StrictData
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
build-depends:
aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
@@ -480,7 +480,7 @@ executable xftp
apps/xftp
default-extensions:
StrictData
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
build-depends:
aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
@@ -555,7 +555,7 @@ executable xftp-server
apps/xftp-server
default-extensions:
StrictData
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
build-depends:
aeson ==2.2.*
, ansi-terminal >=0.10 && <0.12
@@ -662,7 +662,7 @@ test-suite simplexmq-test
tests
default-extensions:
StrictData
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts -with-rtsopts=-A64M -with-rtsopts=-N1
ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Werror=missing-methods -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts -with-rtsopts=-A64M -with-rtsopts=-N1
build-depends:
HUnit ==1.6.*
, QuickCheck ==2.14.*
src/Simplex/FileTransfer/Agent.hs
+6 -5
View File
@@ -262,9 +262,9 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
withStore' c $ \db -> updateRcvFileStatus db rcvFileId RFSDecrypting
chunkPaths <- getChunkPaths chunks
encSize <- liftIO $ foldM (\s path -> (s +) . fromIntegral <$> getFileSize path) 0 chunkPaths
when (FileSize encSize /= size) $ throwError $ XFTP XFTP.SIZE
when (FileSize encSize /= size) $ throwError $ XFTP "" XFTP.SIZE
encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
when (FileDigest encDigest /= digest) $ throwError $ XFTP XFTP.DIGEST
when (FileDigest encDigest /= digest) $ throwError $ XFTP "" XFTP.DIGEST
let destFile = CryptoFile fsSavePath cfArgs
void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure destFile
case redirect of
@@ -281,10 +281,11 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
-- proceed with redirect
yaml <- liftError (INTERNAL . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `agentFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
next@FileDescription {chunks = nextChunks} <- case strDecode (LB.toStrict yaml) of
Left _ -> throwError . XFTP $ XFTP.REDIRECT "decode error"
-- TODO switch to another error constructor
Left _ -> throwError . XFTP "" $ XFTP.REDIRECT "decode error"
Right (ValidFileDescription fd@FileDescription {size = dstSize, digest = dstDigest})
| dstSize /= redirectSize -> throwError . XFTP $ XFTP.REDIRECT "size mismatch"
| dstDigest /= redirectDigest -> throwError . XFTP $ XFTP.REDIRECT "digest mismatch"
| dstSize /= redirectSize -> throwError . XFTP "" $ XFTP.REDIRECT "size mismatch"
| dstDigest /= redirectDigest -> throwError . XFTP "" $ XFTP.REDIRECT "digest mismatch"
| otherwise -> pure fd
-- register and download chunks from the actual file
withStore c $ \db -> updateRcvFileRedirect db redirectDbId next
src/Simplex/FileTransfer/Protocol.hs
+1
View File
@@ -143,6 +143,7 @@ instance Protocol XFTPVersion XFTPErrorType FileResponse where
type ProtoType FileResponse = 'PXFTP
protocolClientHandshake = xftpClientHandshakeStub
protocolPing = FileCmd SFRecipient PING
protocolSub _ = False
protocolError = \case
FRErr e -> Just e
_ -> Nothing
src/Simplex/Messaging/Agent.hs
+32 -23
View File
@@ -157,7 +157,7 @@ import Simplex.Messaging.Agent.Store
import Simplex.Messaging.Agent.Store.SQLite
import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission)
import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission, TransmissionType (..))
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Crypto.File (CryptoFile, CryptoFileArgs)
import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
@@ -1288,13 +1288,13 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
Left e -> do
let err = if msgType == AM_A_MSG_ then MERR mId e else ERR e
case e of
SMP SMP.QUOTA -> case msgType of
SMP _ SMP.QUOTA -> case msgType of
AM_CONN_INFO -> connError msgId NOT_AVAILABLE
AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
_ -> do
expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RISlow
SMP SMP.AUTH -> case msgType of
SMP _ SMP.AUTH -> case msgType of
AM_CONN_INFO -> connError msgId NOT_AVAILABLE
AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
AM_RATCHET_INFO -> connError msgId NOT_AVAILABLE
@@ -1527,7 +1527,7 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
ackQueueMessage :: AgentClient -> RcvQueue -> SMP.MsgId -> AM ()
ackQueueMessage c rq srvMsgId =
sendAck c rq srvMsgId `catchAgentError` \case
SMP SMP.NO_MSG -> pure ()
SMP _ SMP.NO_MSG -> pure ()
e -> throwError e
-- | Suspend SMP agent connection (OFF command) in Reader monad
@@ -1857,7 +1857,7 @@ deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
withStore' c $ \db -> updateNtfToken db tkn ntfTknStatus ntfTknAction
atomically $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
agentNtfDeleteToken c tknId tkn `catchAgentError` \case
NTF AUTH -> pure ()
NTF _ AUTH -> pure ()
e -> throwError e
withStore' c $ \db -> removeNtfToken db tkn
atomically $ nsRemoveNtfToken ns
@@ -1874,7 +1874,7 @@ withToken c tkn@NtfToken {deviceToken, ntfMode} from_ (toStatus, toAction_) f =
let updatedToken = tkn {ntfTknStatus = toStatus, ntfTknAction = toAction_}
atomically $ nsUpdateToken ns updatedToken
pure toStatus
Left e@(NTF AUTH) -> do
Left e@(NTF _ AUTH) -> do
withStore' c $ \db -> removeNtfToken db tkn
atomically $ nsRemoveNtfToken ns
void $ registerNtfToken' c deviceToken ntfMode
@@ -2040,8 +2040,8 @@ data ACKd = ACKd | ACKPending
-- | make sure to ACK or throw in each message processing branch
-- it cannot be finally, unfortunately, as sometimes it needs to be ACK+DEL
processSMPTransmission :: AgentClient -> ServerTransmission SMPVersion BrokerMsg -> AM ()
processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v, sessId, isResponse, rId, cmd) = do
processSMPTransmission :: AgentClient -> ServerTransmission SMPVersion ErrorType BrokerMsg -> AM ()
processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v, sessId, tType, rId, cmd) = do
(rq, SomeConn _ conn) <- withStore c (\db -> getRcvConn db srv rId)
processSMP rq conn $ toConnData conn
where
@@ -2051,10 +2051,10 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
conn
cData@ConnData {userId, connId, connAgentVersion, ratchetSyncState = rss} =
withConnLock c connId "processSMP" $ case cmd of
SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} ->
Right (SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId}) ->
void . handleNotifyAck $ do
isGET <- atomically $ hasGetLock c rq
when (not isGET && isResponse) checkSubscription
unless isGET checkSubscription
msg' <- decryptSMPMessage rq msg
ack' <- handleNotifyAck $ case msg' of
SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} -> processClientMsg srvTs msgFlags msgBody
@@ -2202,7 +2202,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
ackDel aId = enqueueCmd (ICAckDel rId srvMsgId aId) $> ACKd
handleNotifyAck :: AM ACKd -> AM ACKd
handleNotifyAck m = m `catchAgentError` \e -> notify (ERR e) >> ack
SMP.END ->
Right SMP.END ->
atomically (TM.lookup tSess smpClients $>>= (tryReadTMVar . sessionVar) >>= processEND)
>>= logServer "<--" c srv rId
where
@@ -2215,10 +2215,10 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
| otherwise -> ignored
_ -> ignored
ignored = pure "END from disconnected client - ignored"
SMP.SUBOK -> checkSubscription
_ -> do
logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
notify . ERR $ BROKER (B.unpack $ strEncode srv) $ if isResponse then TIMEOUT else UNEXPECTED
Right (SMP.ERR e) -> notify $ ERR $ SMP (B.unpack $ strEncode srv) e
Right SMP.OK -> checkSubscription
Right _ -> unexpected
Left e -> notify $ ERR $ protocolClientError SMP (B.unpack $ strEncode srv) e
where
notify :: forall e m. MonadIO m => AEntityI e => ACommand 'Agent e -> m ()
notify = atomically . notify'
@@ -2232,15 +2232,24 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
enqueueCmd :: InternalCommand -> AM ()
enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand
unexpected :: AM ()
unexpected = do
logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
-- TODO add extended information about transmission type once UNEXPECTED has string
notify . ERR $ BROKER (B.unpack $ strEncode srv) UNEXPECTED
checkSubscription :: AM ()
checkSubscription = do
added <-
atomically $
ifM
((&&) <$> hasPendingSubscription c connId <*> activeClientSession c tSess sessId)
(True <$ addSubscription c rq)
(pure False)
when (added && isResponse) $ notify $ UP srv [connId]
checkSubscription = case tType of
TTEvent -> pure ()
TTUncorrelatedResponse -> unexpected
TTExpiredSubResponse -> do
added <-
atomically $
ifM
((&&) <$> hasPendingSubscription c connId <*> activeClientSession c tSess sessId)
(True <$ addSubscription c rq)
(pure False)
when added $ notify $ UP srv [connId]
decryptClientMessage :: C.DhSecretX25519 -> SMP.ClientMsgEnvelope -> AM (SMP.PrivHeader, AgentMsgEnvelope)
decryptClientMessage e2eDh SMP.ClientMsgEnvelope {cmNonce, cmEncBody} = do
src/Simplex/Messaging/Agent/Client.hs
+7 -6
View File
@@ -85,6 +85,7 @@ module Simplex.Messaging.Agent.Client
agentDRG,
getAgentSubscriptions,
slowNetworkConfig,
protocolClientError,
Worker (..),
SessionVar (..),
SubscriptionsInfo (..),
@@ -263,7 +264,7 @@ data AgentClient = AgentClient
active :: TVar Bool,
rcvQ :: TBQueue (ATransmission 'Client),
subQ :: TBQueue (ATransmission 'Agent),
msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg),
msgQ :: TBQueue (ServerTransmission SMPVersion ErrorType BrokerMsg),
smpServers :: TMap UserId (NonEmpty SMPServerWithAuth),
smpClients :: TMap SMPTransportSession SMPClientVar,
ntfServers :: TVar [NtfServer],
@@ -514,7 +515,7 @@ agentDRG AgentClient {agentEnv = Env {random}} = random
class (Encoding err, Show err) => ProtocolServerClient v err msg | msg -> v, msg -> err where
type Client msg = c | c -> msg
getProtocolServerClient :: AgentClient -> TransportSession msg -> AM (Client msg)
clientProtocolError :: err -> AgentErrorType
clientProtocolError :: HostName -> err -> AgentErrorType
closeProtocolServerClient :: Client msg -> IO ()
clientServer :: Client msg -> String
clientTransportHost :: Client msg -> TransportHost
@@ -940,13 +941,13 @@ withXFTPClient c (userId, srv, entityId) cmdStr action = do
tSess <- liftIO $ mkTransportSession c userId srv entityId
withLogClient c tSess entityId cmdStr action
liftClient :: (Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a
liftClient :: (Show err, Encoding err) => (HostName -> err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a
liftClient protocolError_ = liftError . protocolClientError protocolError_
{-# INLINE liftClient #-}
protocolClientError :: (Show err, Encoding err) => (err -> AgentErrorType) -> HostName -> ProtocolClientError err -> AgentErrorType
protocolClientError :: (Show err, Encoding err) => (HostName -> err -> AgentErrorType) -> HostName -> ProtocolClientError err -> AgentErrorType
protocolClientError protocolError_ host = \case
PCEProtocolError e -> protocolError_ e
PCEProtocolError e -> protocolError_ host e
PCEResponseError e -> BROKER host $ RESPONSE $ B.unpack $ smpEncode e
PCEUnexpectedResponse _ -> BROKER host UNEXPECTED
PCEResponseTimeout -> BROKER host TIMEOUT
@@ -1025,7 +1026,7 @@ runXFTPServerTest c userId (ProtoServerWithAuth srv auth) = do
liftError (testErr TSUploadFile) $ X.uploadXFTPChunk xftp spKey sId chunkSpec
liftError (testErr TSDownloadFile) $ X.downloadXFTPChunk g xftp rpKey rId $ XFTPRcvChunkSpec rcvPath chSize digest
rcvDigest <- liftIO $ C.sha256Hash <$> B.readFile rcvPath
unless (digest == rcvDigest) $ throwError $ ProtocolTestFailure TSCompareFile $ XFTP DIGEST
unless (digest == rcvDigest) $ throwError $ ProtocolTestFailure TSCompareFile $ XFTP (B.unpack $ strEncode srv) DIGEST
liftError (testErr TSDeleteFile) $ X.deleteXFTPChunk xftp spKey sId
ok <- tcpTimeout xftpNetworkConfig `timeout` X.closeXFTPClient xftp
incClientStat c userId xftp "XFTP_TEST" "OK"
src/Simplex/Messaging/Agent/Protocol.hs
+19 -19
View File
@@ -193,13 +193,13 @@ import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Crypto.Ratchet
( InitialKeys (..),
PQEncryption (..),
pattern PQEncOff,
PQSupport,
pattern PQSupportOn,
pattern PQSupportOff,
RcvE2ERatchetParams,
RcvE2ERatchetParamsUri,
SndE2ERatchetParams
SndE2ERatchetParams,
pattern PQEncOff,
pattern PQSupportOff,
pattern PQSupportOn,
)
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
@@ -213,14 +213,14 @@ import Simplex.Messaging.Protocol
MsgId,
NMsgMeta,
ProtocolServer (..),
SMPClientVersion,
SMPMsgMeta,
SMPServer,
SMPServerWithAuth,
SndPublicAuthKey,
SubscriptionMode,
SMPClientVersion,
VersionSMPC,
VersionRangeSMPC,
VersionSMPC,
initialSMPClientVersion,
legacyEncodeServer,
legacyServerP,
@@ -908,7 +908,7 @@ instance Encoding AgentMsgEnvelope where
-- AgentRatchetInfo is not encrypted with double ratchet, but with per-queue E2E encryption
data AgentMessage
= -- used by the initiating party when confirming reply queue
AgentConnInfo ConnInfo
AgentConnInfo ConnInfo
| -- 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
@@ -1382,9 +1382,9 @@ deriving instance Show (ConnectionRequestUri m)
data AConnectionRequestUri = forall m. ConnectionModeI m => ACR (SConnectionMode m) (ConnectionRequestUri m)
instance Eq AConnectionRequestUri where
ACR m cr == ACR m' cr' = case testEquality m m' of
Just Refl -> cr == cr'
_ -> False
ACR m cr == ACR m' cr' = case testEquality m m' of
Just Refl -> cr == cr'
_ -> False
deriving instance Show AConnectionRequestUri
@@ -1469,11 +1469,11 @@ data AgentErrorType
| -- | connection errors
CONN {connErr :: ConnectionErrorType}
| -- | SMP protocol errors forwarded to agent clients
SMP {smpErr :: ErrorType}
SMP {serverAddress :: String, smpErr :: ErrorType}
| -- | NTF protocol errors forwarded to agent clients
NTF {ntfErr :: ErrorType}
NTF {serverAddress :: String, ntfErr :: ErrorType}
| -- | XFTP protocol errors forwarded to agent clients
XFTP {xftpErr :: XFTPErrorType}
XFTP {serverAddress :: String, xftpErr :: XFTPErrorType}
| -- | XRCP protocol errors forwarded to agent clients
RCP {rcpErr :: RCErrorType}
| -- | SMP server errors
@@ -1584,9 +1584,9 @@ instance StrEncoding AgentErrorType where
strP =
"CMD " *> (CMD <$> parseRead1)
<|> "CONN " *> (CONN <$> parseRead1)
<|> "SMP " *> (SMP <$> strP)
<|> "NTF " *> (NTF <$> strP)
<|> "XFTP " *> (XFTP <$> strP)
<|> "SMP " *> (SMP <$> textP <*> _strP)
<|> "NTF " *> (NTF <$> textP <*> _strP)
<|> "XFTP " *> (XFTP <$> textP <*> _strP)
<|> "RCP " *> (RCP <$> strP)
<|> "BROKER " *> (BROKER <$> textP <* " RESPONSE " <*> (RESPONSE <$> textP))
<|> "BROKER " *> (BROKER <$> textP <* " TRANSPORT " <*> (TRANSPORT <$> transportErrorP))
@@ -1602,9 +1602,9 @@ instance StrEncoding AgentErrorType where
strEncode = \case
CMD e -> "CMD " <> bshow e
CONN e -> "CONN " <> bshow e
SMP e -> "SMP " <> strEncode e
NTF e -> "NTF " <> strEncode e
XFTP e -> "XFTP " <> strEncode e
SMP srv e -> "SMP " <> text srv <> " " <> strEncode e
NTF srv e -> "NTF " <> text srv <> " " <> strEncode e
XFTP srv e -> "XFTP " <> text srv <> " " <> strEncode e
RCP e -> "RCP " <> strEncode e
BROKER srv (RESPONSE e) -> "BROKER " <> text srv <> " RESPONSE " <> text e
BROKER srv (TRANSPORT e) -> "BROKER " <> text srv <> " TRANSPORT " <> serializeTransportError e
src/Simplex/Messaging/Client.hs
+63 -36
View File
@@ -70,6 +70,7 @@ module Simplex.Messaging.Client
proxyUsername,
temporaryClientError,
ServerTransmission,
TransmissionType (..),
ClientCommand,
-- * For testing
@@ -83,6 +84,7 @@ where
import Control.Concurrent.Async
import Control.Concurrent.STM
import Control.Exception
import Control.Logger.Simple
import Control.Monad
import Control.Monad.Except
import Control.Monad.IO.Class (liftIO)
@@ -110,7 +112,7 @@ import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Client (SocksProxy, TransportClientConfig (..), TransportHost (..), defaultTcpConnectTimeout, runTransportClient)
import Simplex.Messaging.Transport.KeepAlive
import Simplex.Messaging.Transport.WebSockets (WS)
import Simplex.Messaging.Util (bshow, diffToMicroseconds, raceAny_, threadDelay', whenM)
import Simplex.Messaging.Util (bshow, diffToMicroseconds, ifM, raceAny_, threadDelay', tshow, whenM)
import Simplex.Messaging.Version
import System.Timeout (timeout)
@@ -137,7 +139,7 @@ data PClient v err msg = PClient
sentCommands :: TMap CorrId (Request err msg),
sndQ :: TBQueue (TVar Bool, ByteString),
rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
msgQ :: Maybe (TBQueue (ServerTransmission v msg))
msgQ :: Maybe (TBQueue (ServerTransmission v err msg))
}
smpClientStub :: TVar ChaChaDRG -> ByteString -> VersionSMP -> Maybe (THandleAuth 'TClient) -> STM SMPClient
@@ -188,7 +190,10 @@ type SMPClient = ProtocolClient SMPVersion ErrorType BrokerMsg
type ClientCommand msg = (Maybe C.APrivateAuthKey, EntityId, ProtoCommand msg)
-- | Type synonym for transmission from some SPM server queue.
type ServerTransmission v msg = (TransportSession msg, Version v, SessionId, Bool, EntityId, msg)
type ServerTransmission v err msg = (TransportSession msg, Version v, SessionId, TransmissionType, EntityId, Either (ProtocolClientError err) msg)
data TransmissionType = TTEvent | TTUncorrelatedResponse | TTExpiredSubResponse
deriving (Show)
data HostMode
= -- | prefer (or require) onion hosts when connecting via SOCKS proxy
@@ -285,6 +290,8 @@ defaultSMPClientConfig = defaultClientConfig supportedClientSMPRelayVRange
data Request err msg = Request
{ corrId :: CorrId,
entityId :: EntityId,
subscription :: Bool,
pending :: TVar Bool,
responseVar :: TMVar (Either (ProtocolClientError err) msg)
}
@@ -331,7 +338,7 @@ type TransportSession msg = (UserId, ProtoServer msg, Maybe EntityId)
--
-- A single queue can be used for multiple 'SMPClient' instances,
-- as 'SMPServerTransmission' includes server information.
getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmission v msg)) -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmission v err msg)) -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, serverVRange, agreeSecret} msgQ disconnected = do
case chooseTransportHost networkConfig (host srv) of
Right useHost ->
@@ -439,27 +446,40 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
processMsg :: ProtocolClient v err msg -> SignedTransmission err msg -> IO ()
processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr))
| isResponse =
| not $ B.null $ bs corrId =
atomically (TM.lookup corrId sentCommands) >>= \case
Nothing -> sendMsg respOrErr
Just Request {entityId, responseVar} -> atomically $ do
TM.delete corrId sentCommands
putTMVar responseVar $ response entityId
| otherwise = sendMsg respOrErr
Nothing -> sendMsg TTUncorrelatedResponse
Just Request {entityId, subscription, pending, responseVar} -> do
wasPending <-
atomically $ do
TM.delete corrId sentCommands
ifM
(readTVar pending)
(True <$ tryPutTMVar responseVar (response entityId))
(pure False)
unless wasPending $ mapM_ sendMsg $ transmissionType entityId subscription
| otherwise = sendMsg TTEvent
where
isResponse = not $ B.null $ bs corrId
-- we don't send expired events that are not subscription responses to avoid double error reporting,
-- as the expired command already reported them.
transmissionType entityId subscription
| entityId /= entId = Just TTUncorrelatedResponse
| subscription = Just TTExpiredSubResponse
| otherwise = Nothing
response entityId
| entityId == entId =
case respOrErr of
Left e -> Left $ PCEResponseError e
Right r -> case protocolError r of
Just e -> Left $ PCEProtocolError e
_ -> Right r
| entityId == entId = clientResp
| otherwise = Left . PCEUnexpectedResponse $ bshow respOrErr
sendMsg :: Either err msg -> IO ()
sendMsg = \case
Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c isResponse entId msg) msgQ
Left e -> putStrLn $ "SMP client error: " <> show e
clientResp = case respOrErr of
Left e -> Left $ PCEResponseError e
Right r -> case protocolError r of
Just e -> Left $ PCEProtocolError e
_ -> Right r
sendMsg :: TransmissionType -> IO ()
sendMsg tType = case msgQ of
Just q -> atomically $ writeTBQueue q $ serverTransmission c tType entId clientResp
Nothing -> case clientResp of
Left e -> logError $ "SMP client error: " <> tshow e
Right _ -> logWarn $ "SMP client unprocessed event: " <> tshow tType
proxyUsername :: TransportSession msg -> ByteString
proxyUsername (userId, _, entityId_) = C.sha256Hash $ bshow userId <> maybe "" (":" <>) entityId_
@@ -533,7 +553,6 @@ subscribeSMPQueue c@ProtocolClient {client_ = PClient {sendPings}} rpKey rId = d
liftIO . atomically $ writeTVar sendPings True
sendSMPCommand c (Just rpKey) rId SUB >>= \case
OK -> pure ()
SUBOK -> pure ()
cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd
r -> throwE . PCEUnexpectedResponse $ bshow r
@@ -554,17 +573,16 @@ streamSubscribeSMPQueues c qs cb = streamProtocolCommands c cs $ mapM process >=
processSUBResponse :: SMPClient -> Response ErrorType BrokerMsg -> IO (Either SMPClientError ())
processSUBResponse c (Response rId r) = case r of
Right OK -> pure $ Right ()
Right SUBOK -> pure $ Right () -- from v8 of SMP protocol
Right cmd@MSG {} -> writeSMPMessage c rId cmd $> Right ()
Right r' -> pure . Left . PCEUnexpectedResponse $ bshow r'
Left e -> pure $ Left e
writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c False rId msg) (msgQ $ client_ c)
writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c TTEvent rId (Right msg)) (msgQ $ client_ c)
serverTransmission :: ProtocolClient v err msg -> Bool -> RecipientId -> msg -> ServerTransmission v msg
serverTransmission ProtocolClient {thParams = THandleParams {thVersion, sessionId}, client_ = PClient {transportSession}} isResponse entityId message =
(transportSession, thVersion, sessionId, isResponse, entityId, message)
serverTransmission :: ProtocolClient v err msg -> TransmissionType -> RecipientId -> Either (ProtocolClientError err) msg -> ServerTransmission v err msg
serverTransmission ProtocolClient {thParams = THandleParams {thVersion, sessionId}, client_ = PClient {transportSession}} tType entityId msgOrErr =
(transportSession, thVersion, sessionId, tType, entityId, msgOrErr)
-- | Get message from SMP queue. The server returns ERR PROHIBITED if a client uses SUB and GET via the same transport connection for the same queue
--
@@ -689,7 +707,7 @@ sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd)
type PCTransmission err msg = (Either TransportError SentRawTransmission, Request err msg)
-- | Send multiple commands with batching and collect responses
sendProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
sendProtocolCommands :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs = do
bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
validate . concat =<< mapM (sendBatch c) bs
@@ -706,7 +724,7 @@ sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSiz
where
diff = L.length cs - length rs
streamProtocolCommands :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> ([Response err msg] -> IO ()) -> IO ()
streamProtocolCommands :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> NonEmpty (ClientCommand msg) -> ([Response err msg] -> IO ()) -> IO ()
streamProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSize}} cs cb = do
bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
mapM_ (cb <=< sendBatch c) bs
@@ -729,7 +747,7 @@ sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
(: []) <$> getResponse c active r
-- | Send Protocol command
sendProtocolCommand :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
sendProtocolCommand :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> Maybe C.APrivateAuthKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THandleParams {batch, blockSize}} pKey entId cmd =
ExceptT $ uncurry sendRecv =<< mkTransmission c (pKey, entId, cmd)
where
@@ -750,20 +768,20 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
-- TODO switch to timeout or TimeManager that supports Int64
getResponse :: ProtocolClient v err msg -> TVar Bool -> Request err msg -> IO (Response err msg)
getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount, sentCommands}} active Request {corrId, entityId, responseVar} = do
getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount}} active Request {entityId, pending, responseVar} = do
response <-
timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case
Just r -> atomically (writeTVar timeoutErrorCount 0) $> r
Nothing -> do
atomically (writeTVar active False >> TM.delete corrId sentCommands)
atomically (writeTVar active False >> writeTVar pending False)
atomically $ modifyTVar' timeoutErrorCount (+ 1)
pure $ Left PCEResponseTimeout
pure Response {entityId, response}
mkTransmission :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> ClientCommand msg -> IO (PCTransmission err msg)
mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} (pKey_, entId, cmd) = do
mkTransmission :: forall v err msg. Protocol v err msg => ProtocolClient v err msg -> ClientCommand msg -> IO (PCTransmission err msg)
mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} (pKey_, entityId, cmd) = do
corrId <- atomically getNextCorrId
let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entId, cmd)
let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entityId, cmd)
auth = authTransmission (thAuth thParams) pKey_ corrId tForAuth
r <- atomically $ mkRequest corrId
pure ((,tToSend) <$> auth, r)
@@ -772,7 +790,16 @@ mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCo
getNextCorrId = CorrId <$> C.randomBytes 24 clientCorrId -- also used as nonce
mkRequest :: CorrId -> STM (Request err msg)
mkRequest corrId = do
r <- Request corrId entId <$> newEmptyTMVar
pending <- newTVar True
responseVar <- newEmptyTMVar
let r =
Request
{ corrId,
entityId,
subscription = protocolSub cmd,
pending,
responseVar
}
TM.insert corrId r sentCommands
pure r
src/Simplex/Messaging/Client/Agent.hs
+2 -2
View File
@@ -39,7 +39,7 @@ import Simplex.Messaging.Agent.RetryInterval
import Simplex.Messaging.Client
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (BrokerMsg, NotifierId, NtfPrivateAuthKey, ProtocolServer (..), QueueId, RcvPrivateAuthKey, RecipientId, SMPServer)
import Simplex.Messaging.Protocol (BrokerMsg, ErrorType, NotifierId, NtfPrivateAuthKey, ProtocolServer (..), QueueId, RcvPrivateAuthKey, RecipientId, SMPServer)
import Simplex.Messaging.Session
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
@@ -94,7 +94,7 @@ defaultSMPClientAgentConfig =
data SMPClientAgent = SMPClientAgent
{ agentCfg :: SMPClientAgentConfig,
msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg),
msgQ :: TBQueue (ServerTransmission SMPVersion ErrorType BrokerMsg),
agentQ :: TBQueue SMPClientAgentEvent,
randomDrg :: TVar ChaChaDRG,
smpClients :: TMap SMPServer SMPClientVar,
src/Simplex/Messaging/Notifications/Protocol.hs
+1
View File
@@ -154,6 +154,7 @@ instance Protocol NTFVersion ErrorType NtfResponse where
type ProtoType NtfResponse = 'PNTF
protocolClientHandshake c _ks = ntfClientHandshake c
protocolPing = NtfCmd SSubscription PING
protocolSub _ = False
protocolError = \case
NRErr e -> Just e
_ -> Nothing
src/Simplex/Messaging/Notifications/Server.hs
+7 -5
View File
@@ -218,10 +218,10 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
receiveSMP :: M ()
receiveSMP = forever $ do
((_, srv, _), _, _, _, ntfId, msg) <- atomically $ readTBQueue msgQ
((_, srv, _), _, _, tType, ntfId, msgOrErr) <- atomically $ readTBQueue msgQ
let smpQueue = SMPQueueNtf srv ntfId
case msg of
SMP.NMSG nmsgNonce encNMsgMeta -> do
case msgOrErr of
Right (SMP.NMSG nmsgNonce encNMsgMeta) -> do
ntfTs <- liftIO getSystemTime
st <- asks store
NtfPushServer {pushQ} <- asks pushServer
@@ -231,8 +231,10 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
findNtfSubscriptionToken st smpQueue
>>= mapM_ (\tkn -> writeTBQueue pushQ (tkn, PNMessage PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}))
incNtfStat ntfReceived
SMP.END -> updateSubStatus smpQueue NSEnd
_ -> pure ()
Right SMP.END -> updateSubStatus smpQueue NSEnd
Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e
Right _ -> logError $ "SMP server unexpected response: " <> tshow tType
Left e -> logError $ "SMP client error: " <> tshow e
receiveAgent =
forever $
src/Simplex/Messaging/Protocol.hs
+6 -9
View File
@@ -395,7 +395,6 @@ data BrokerMsg where
NMSG :: C.CbNonce -> EncNMsgMeta -> BrokerMsg
END :: BrokerMsg
OK :: BrokerMsg
SUBOK :: BrokerMsg
ERR :: ErrorType -> BrokerMsg
PONG :: BrokerMsg
deriving (Eq, Show)
@@ -583,7 +582,6 @@ data BrokerMsgTag
| NMSG_
| END_
| OK_
| SUBOK_
| ERR_
| PONG_
deriving (Show)
@@ -643,7 +641,6 @@ instance Encoding BrokerMsgTag where
NMSG_ -> "NMSG"
END_ -> "END"
OK_ -> "OK"
SUBOK_ -> "SUBOK"
ERR_ -> "ERR"
PONG_ -> "PONG"
smpP = messageTagP
@@ -656,7 +653,6 @@ instance ProtocolMsgTag BrokerMsgTag where
"NMSG" -> Just NMSG_
"END" -> Just END_
"OK" -> Just OK_
"SUBOK" -> Just SUBOK_
"ERR" -> Just ERR_
"PONG" -> Just PONG_
_ -> Nothing
@@ -1098,6 +1094,7 @@ class (ProtocolEncoding v err msg, ProtocolEncoding v err (ProtoCommand msg), Sh
type ProtoType msg = (sch :: ProtocolType) | sch -> msg
protocolClientHandshake :: forall c. Transport c => c -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRange v -> ExceptT TransportError IO (THandle v c 'TClient)
protocolPing :: ProtoCommand msg
protocolSub :: ProtoCommand msg -> Bool
protocolError :: msg -> Maybe err
type ProtoServer msg = ProtocolServer (ProtoType msg)
@@ -1107,6 +1104,10 @@ instance Protocol SMPVersion ErrorType BrokerMsg where
type ProtoType BrokerMsg = 'PSMP
protocolClientHandshake = smpClientHandshake
protocolPing = Cmd SSender PING
protocolSub = \case
Cmd SRecipient SUB -> True
Cmd SNotifier NSUB -> True
_ -> False
protocolError = \case
ERR e -> Just e
_ -> Nothing
@@ -1202,7 +1203,7 @@ instance ProtocolEncoding SMPVersion ErrorType Cmd where
instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
type Tag BrokerMsg = BrokerMsgTag
encodeProtocol v = \case
encodeProtocol _v = \case
IDS (QIK rcvId sndId srvDh) -> e (IDS_, ' ', rcvId, sndId, srvDh)
MSG RcvMessage {msgId, msgBody = EncRcvMsgBody body} ->
e (MSG_, ' ', msgId, Tail body)
@@ -1210,9 +1211,6 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
NMSG nmsgNonce encNMsgMeta -> e (NMSG_, ' ', nmsgNonce, encNMsgMeta)
END -> e END_
OK -> e OK_
SUBOK
| v >= sendingProxySMPVersion -> e SUBOK_
| otherwise -> e OK_
ERR err -> e (ERR_, ' ', err)
PONG -> e PONG_
where
@@ -1230,7 +1228,6 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
NMSG_ -> NMSG <$> _smpP <*> smpP
END_ -> pure END
OK_ -> pure OK
SUBOK_ -> pure SUBOK
ERR_ -> ERR <$> _smpP
PONG_ -> pure PONG
src/Simplex/Messaging/Server.hs
+5 -5
View File
@@ -724,7 +724,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
deliver sub = do
q <- getStoreMsgQueue "SUB" rId
msg_ <- atomically $ tryPeekMsg q
deliverMessage "SUB" qr rId sub q msg_ SUBOK
deliverMessage "SUB" qr rId sub q msg_
getMessage :: QueueRec -> M (Transmission BrokerMsg)
getMessage qr = time "GET" $ do
@@ -781,7 +781,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
_ -> do
(deletedMsg_, msg_) <- atomically $ tryDelPeekMsg q msgId
mapM_ updateStats deletedMsg_
deliverMessage "ACK" qr queueId sub q msg_ OK
deliverMessage "ACK" qr queueId sub q msg_
_ -> pure $ err NO_MSG
where
getDelivered :: TVar Sub -> STM (Maybe Sub)
@@ -863,8 +863,8 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
encNMsgMeta = C.cbEncrypt rcvNtfDhSecret cbNonce (smpEncode msgMeta) 128
pure . (cbNonce,) $ fromRight "" encNMsgMeta
deliverMessage :: T.Text -> QueueRec -> RecipientId -> TVar Sub -> MsgQueue -> Maybe Message -> BrokerMsg -> M (Transmission BrokerMsg)
deliverMessage name qr rId sub q msg_ respCmd = time (name <> " deliver") $ do
deliverMessage :: T.Text -> QueueRec -> RecipientId -> TVar Sub -> MsgQueue -> Maybe Message -> M (Transmission BrokerMsg)
deliverMessage name qr rId sub q msg_ = time (name <> " deliver") $ do
readTVarIO sub >>= \case
s@Sub {subThread = NoSub} ->
case msg_ of
@@ -874,7 +874,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
_ -> forkSub $> resp
_ -> pure resp
where
resp = (corrId, rId, respCmd)
resp = (corrId, rId, OK)
forkSub :: M ()
forkSub = do
atomically . modifyTVar' sub $ \s -> s {subThread = SubPending}
src/Simplex/Messaging/Transport.hs
-4
View File
@@ -41,7 +41,6 @@ module Simplex.Messaging.Transport
basicAuthSMPVersion,
subModeSMPVersion,
authCmdsSMPVersion,
sendingProxySMPVersion,
simplexMQVersion,
smpBlockSize,
TransportConfig (..),
@@ -150,9 +149,6 @@ subModeSMPVersion = VersionSMP 6
authCmdsSMPVersion :: VersionSMP
authCmdsSMPVersion = VersionSMP 7
sendingProxySMPVersion :: VersionSMP
sendingProxySMPVersion = VersionSMP 8
currentClientSMPRelayVersion :: VersionSMP
currentClientSMPRelayVersion = VersionSMP 6
tests/AgentTests.hs
+11 -11
View File
@@ -27,9 +27,9 @@ import GHC.Stack (withFrozenCallStack)
import Network.HTTP.Types (urlEncode)
import SMPAgentClient
import SMPClient (testKeyHash, testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServerStoreLogOn)
import Simplex.Messaging.Agent.Protocol hiding (MID, CONF, INFO, REQ)
import Simplex.Messaging.Agent.Protocol hiding (CONF, INFO, MID, REQ)
import qualified Simplex.Messaging.Agent.Protocol as A
import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOn, pattern IKPQOff, pattern PQEncOn, pattern PQSupportOn, pattern PQSupportOff)
import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOff, pattern IKPQOn, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
import qualified Simplex.Messaging.Crypto.Ratchet as CR
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (ErrorType (..))
@@ -242,7 +242,7 @@ testDuplexConnection' (alice, aPQ) (bob, bPQ) = do
alice #: ("4a", "bob", "ACK 7") #> ("4a", "bob", OK)
alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)
bob #: ("17", "alice", "SEND F 9\nmessage 3") #> ("17", "alice", A.MID 8 pq)
bob <# ("", "alice", MERR 8 (SMP AUTH))
bob <#= \case ("", "alice", MERR 8 (SMP _ AUTH)) -> True; _ -> False
alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
alice #:# "nothing else should be delivered to alice"
@@ -280,7 +280,7 @@ testDuplexConnRandomIds' (alice, aPQ) (bob, bPQ) = do
alice #: ("4a", bobConn, "ACK 7") #> ("4a", bobConn, OK)
alice #: ("5", bobConn, "OFF") #> ("5", bobConn, OK)
bob #: ("17", aliceConn, "SEND F 9\nmessage 3") #> ("17", aliceConn, A.MID 8 pq)
bob <# ("", aliceConn, MERR 8 (SMP AUTH))
bob <#= \case ("", cId, MERR 8 (SMP _ AUTH)) -> cId == aliceConn; _ -> False
alice #: ("6", bobConn, "DEL") #> ("6", bobConn, OK)
alice #:# "nothing else should be delivered to alice"
@@ -383,7 +383,7 @@ testSubscrNotification t (server, _) client = do
killThread server
client <#. ("", "", DOWN testSMPServer ["conn1"])
withSmpServer (ATransport t) $
client <# ("", "conn1", ERR (SMP AUTH)) -- this new server does not have the queue
client <#= \case ("", "conn1", ERR (SMP _ AUTH)) -> True; _ -> False -- this new server does not have the queue
testMsgDeliveryServerRestart :: forall c. Transport c => (c, InitialKeys) -> (c, PQSupport) -> IO ()
testMsgDeliveryServerRestart (alice, aPQ) (bob, bPQ) = do
@@ -547,10 +547,10 @@ testResumeDeliveryQuotaExceeded _ alice bob = do
bob <#= \case ("", "alice", Msg "message 4") -> True; _ -> False
bob #: ("4", "alice", "ACK 7") #> ("4", "alice", OK)
inAnyOrder
(tGetAgent alice)
[ \case ("", c, Right (SENT 8)) -> c == "bob"; _ -> False,
\case ("", c, Right QCONT) -> c == "bob"; _ -> False
]
(tGetAgent alice)
[ \case ("", c, Right (SENT 8)) -> c == "bob"; _ -> False,
\case ("", c, Right QCONT) -> c == "bob"; _ -> False
]
bob <#= \case ("", "alice", Msg "over quota") -> True; _ -> False
-- message 8 is skipped because of alice agent sending "QCONT" message
bob #: ("5", "alice", "ACK 9") #> ("5", "alice", OK)
@@ -580,7 +580,7 @@ enableKEMStr _ = ""
pqConnModeStr :: InitialKeys -> ByteString
pqConnModeStr (IKNoPQ PQSupportOff) = ""
pqConnModeStr pq = " " <> strEncode pq
pqConnModeStr pq = " " <> strEncode pq
sendMessage :: Transport c => (c, ConnId) -> (c, ConnId) -> ByteString -> IO ()
sendMessage (h1, name1) (h2, name2) msg = do
@@ -630,7 +630,7 @@ syntaxTests t = do
<> " subscribe "
<> "14\nbob's connInfo"
)
>#> ("311", "a", "ERR SMP AUTH")
>#> ("311", "a", "ERR SMP smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001 AUTH")
describe "invalid" $ do
it "no parameters" $ ("321", "", "JOIN") >#> ("321", "", "ERR CMD SYNTAX")
where
tests/AgentTests/FunctionalAPITests.hs
+16 -16
View File
@@ -415,7 +415,7 @@ functionalAPITests t = do
describe "server with password" $ do
let auth = Just "abcd"
srv = ProtoServerWithAuth testSMPServer2
authErr = Just (ProtocolTestFailure TSCreateQueue $ SMP AUTH)
authErr = Just (ProtocolTestFailure TSCreateQueue $ SMP (B.unpack $ strEncode testSMPServer2) AUTH)
it "should pass with correct password" $ testSMPServerConnectionTest t auth (srv auth) `shouldReturn` Nothing
it "should fail without password" $ testSMPServerConnectionTest t auth (srv Nothing) `shouldReturn` authErr
it "should fail with incorrect password" $ testSMPServerConnectionTest t auth (srv $ Just "wrong") `shouldReturn` authErr
@@ -530,7 +530,7 @@ runAgentClientTest pqSupport alice@AgentClient {} bob baseId =
ackMessage alice bobId (baseId + 4) Nothing
suspendConnection alice bobId
5 <- msgId <$> A.sendMessage bob aliceId pqEnc SMP.noMsgFlags "message 2"
get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
deleteConnection alice bobId
liftIO $ noMessages alice "nothing else should be delivered to alice"
where
@@ -660,7 +660,7 @@ runAgentClientContactTest pqSupport alice bob baseId =
ackMessage alice bobId (baseId + 4) Nothing
suspendConnection alice bobId
5 <- msgId <$> A.sendMessage bob aliceId pqEnc SMP.noMsgFlags "message 2"
get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
deleteConnection alice bobId
liftIO $ noMessages alice "nothing else should be delivered to alice"
where
@@ -1106,7 +1106,7 @@ testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testP
5 <- sendMessage a bId SMP.noMsgFlags "2"
liftIO $ threadDelay 1000000
6 <- sendMessage a bId SMP.noMsgFlags "3" -- this won't expire
get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
get a =##> \case ("", c, MERR 5 (SMP _ QUOTA)) -> bId == c; _ -> False
pure (aId, bId)
withAgent 3 agentCfg initAgentServers testDB2 $ \b' -> runRight_ $ do
subscribeConnection b' aId
@@ -1134,15 +1134,15 @@ testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1}
7 <- sendMessage a bId SMP.noMsgFlags "4"
liftIO $ threadDelay 1000000
8 <- sendMessage a bId SMP.noMsgFlags "5" -- this won't expire
get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
get a =##> \case ("", c, MERR 5 (SMP _ QUOTA)) -> bId == c; _ -> False
get a >>= \case
("", c, MERR 6 (SMP QUOTA)) -> do
("", c, MERR 6 (SMP _ QUOTA)) -> do
liftIO $ bId `shouldBe` c
get a =##> \case ("", c', MERR 7 (SMP QUOTA)) -> bId == c'; ("", c', MERRS [7] (SMP QUOTA)) -> bId == c'; _ -> False
("", c, MERRS [6] (SMP QUOTA)) -> do
get a =##> \case ("", c', MERR 7 (SMP _ QUOTA)) -> bId == c'; ("", c', MERRS [7] (SMP _ QUOTA)) -> bId == c'; _ -> False
("", c, MERRS [6] (SMP _ QUOTA)) -> do
liftIO $ bId `shouldBe` c
get a =##> \case ("", c', MERR 7 (SMP QUOTA)) -> bId == c'; _ -> False
("", c, MERRS [6, 7] (SMP QUOTA)) -> liftIO $ bId `shouldBe` c
get a =##> \case ("", c', MERR 7 (SMP _ QUOTA)) -> bId == c'; _ -> False
("", c, MERRS [6, 7] (SMP _ QUOTA)) -> liftIO $ bId `shouldBe` c
r -> error $ show r
pure (aId, bId)
withAgent 3 agentCfg initAgentServers testDB2 $ \b' -> runRight_ $ do
@@ -1679,7 +1679,7 @@ testAcceptContactAsync =
ackMessage alice bobId (baseId + 4) Nothing
suspendConnection alice bobId
5 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 5); _ -> False
deleteConnection alice bobId
liftIO $ noMessages alice "nothing else should be delivered to alice"
where
@@ -1725,7 +1725,7 @@ testWaitDeliveryNoPending t = withAgentClients2 $ \alice bob ->
get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False
3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
get bob ##> ("", aliceId, MERR (baseId + 3) (SMP AUTH))
get bob =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == aliceId && mId == (baseId + 3); _ -> False
liftIO $ noMessages alice "nothing else should be delivered to alice"
liftIO $ noMessages bob "nothing else should be delivered to bob"
@@ -1820,8 +1820,8 @@ testWaitDeliveryAUTHErr t =
liftIO $ noMessages bob "nothing else should be delivered to bob"
withSmpServerStoreLogOn t testPort $ \_ -> do
get alice ##> ("", bobId, MERR (baseId + 3) (SMP AUTH))
get alice ##> ("", bobId, MERR (baseId + 4) (SMP AUTH))
get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 3); _ -> False
get alice =##> \case ("", cId, MERR mId (SMP _ AUTH)) -> cId == bobId && mId == (baseId + 4); _ -> False
get alice =##> \case ("", cId, DEL_CONN) -> cId == bobId; _ -> False
liftIO $ noMessages alice "nothing else should be delivered to alice"
@@ -2392,11 +2392,11 @@ testCreateQueueAuth srvVersion clnt1 clnt2 = do
b <- getClient 2 clnt2 testDB2
r <- runRight $ do
tryError (createConnection a 1 True SCMInvitation Nothing SMSubscribe) >>= \case
Left (SMP AUTH) -> pure 0
Left (SMP _ AUTH) -> pure 0
Left e -> throwError e
Right (bId, qInfo) ->
tryError (joinConnection b 1 True qInfo "bob's connInfo" SMSubscribe) >>= \case
Left (SMP AUTH) -> pure 1
Left (SMP _ AUTH) -> pure 1
Left e -> throwError e
Right aId -> do
("", _, CONF confId _ "bob's connInfo") <- get a
tests/AgentTests/NotificationTests.hs
+11 -10
View File
@@ -6,8 +6,8 @@
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE TypeApplications #-}
{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
module AgentTests.NotificationTests where
@@ -17,10 +17,6 @@ import AgentTests.FunctionalAPITests
createConnection,
exchangeGreetingsMsgId,
get,
withAgent,
withAgentClients2,
withAgentClientsCfgServers2,
withAgentClients3,
joinConnection,
makeConnection,
nGet,
@@ -29,7 +25,11 @@ import AgentTests.FunctionalAPITests
sendMessage,
switchComplete,
testServerMatrix2,
withAgent,
withAgentClients2,
withAgentClients3,
withAgentClientsCfg2,
withAgentClientsCfgServers2,
(##>),
(=##>),
pattern CON,
@@ -59,8 +59,8 @@ import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO)
import Simplex.Messaging.Agent.Store.SQLite (getSavedNtfToken)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
import Simplex.Messaging.Notifications.Protocol
import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
import Simplex.Messaging.Notifications.Server.Push.APNS
import Simplex.Messaging.Notifications.Types (NtfToken (..))
import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
@@ -151,7 +151,8 @@ testNtfMatrix t runTest = do
it "next servers: SMP v7, NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfgV2 agentCfg agentCfg runTest
it "curr servers: SMP v6, NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfg agentCfg agentCfg runTest
skip "this case cannot be supported - see RFC" $
it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $ runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
it "servers: SMP v6, NTF v1; clients: v7/v2 (not supported)" $
runNtfTestCfg t cfg ntfServerCfg agentCfgV7 agentCfgV7 runTest
-- servers can be migrated in any order
it "servers: next SMP v7, curr NTF v1; curr clients: v6/v1" $ runNtfTestCfg t cfgV7 ntfServerCfg agentCfg agentCfg runTest
it "servers: curr SMP v6, next NTF v2; curr clients: v6/v1" $ runNtfTestCfg t cfg ntfServerCfgV2 agentCfg agentCfg runTest
@@ -243,7 +244,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} =
-- now the second token registration is verified
verifyNtfToken a' tkn nonce' verification'
-- the first registration is removed
Left (NTF AUTH) <- tryE $ checkNtfToken a tkn
Left (NTF _ AUTH) <- tryE $ checkNtfToken a tkn
-- and the second is active
NTActive <- checkNtfToken a' tkn
pure ()
@@ -258,7 +259,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
atomically $ readTBQueue apnsQ
liftIO $ sendApnsResponse APNSRespOk
pure ntfData
-- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
-- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
threadDelay 1000000
withAgent 2 agentCfg initAgentServers testDB $ \a' ->
-- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,
@@ -266,7 +267,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
withNtfServer t . runRight_ $ do
verification <- ntfData .-> "verification"
nonce <- C.cbNonce <$> ntfData .-> "nonce"
Left (NTF AUTH) <- tryE $ verifyNtfToken a' tkn nonce verification
Left (NTF _ AUTH) <- tryE $ verifyNtfToken a' tkn nonce verification
APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
atomically $ readTBQueue apnsQ
verification' <- ntfData' .-> "verification"
tests/CoreTests/ProtocolErrorTests.hs
+3
View File
@@ -35,6 +35,9 @@ protocolErrorTests = modifyMaxSuccess (const 1000) $ do
errHasSpaces = \case
BROKER srv (RESPONSE e) -> hasSpaces srv || hasSpaces e
BROKER srv _ -> hasSpaces srv
SMP srv _ -> hasSpaces srv
NTF srv _ -> hasSpaces srv
XFTP srv _ -> hasSpaces srv
_ -> False
hasSpaces s = ' ' `B.elem` encodeUtf8 (T.pack s)
tests/XFTPAgent.hs
+13 -7
View File
@@ -69,7 +69,7 @@ xftpAgentTests = around_ testBracket . describe "agent XFTP API" $ do
describe "server with password" $ do
let auth = Just "abcd"
srv = ProtoServerWithAuth testXFTPServer2
authErr = Just (ProtocolTestFailure TSCreateFile $ XFTP AUTH)
authErr = Just (ProtocolTestFailure TSCreateFile $ XFTP (B.unpack $ strEncode testXFTPServer2) AUTH)
it "should pass with correct password" $ testXFTPServerTest auth (srv auth) `shouldReturn` Nothing
it "should fail without password" $ testXFTPServerTest auth (srv Nothing) `shouldReturn` authErr
it "should fail with incorrect password" $ testXFTPServerTest auth (srv $ Just "wrong") `shouldReturn` authErr
@@ -392,7 +392,8 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
-- receive file - should fail with AUTH error
withAgent 3 agentCfg initAgentServers testDB2 $ \rcp' -> do
runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
rfGet rcp'
rfId' `shouldBe` rfId
-- tmp path should be removed after permanent error
@@ -471,7 +472,8 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
-- send file - should fail with AUTH error
withAgent 2 agentCfg initAgentServers testDB $ \sndr' -> do
runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
("", sfId', SFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- sfGet sndr'
("", sfId', SFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
sfGet sndr'
sfId' `shouldBe` sfId
-- prefix path should be removed after permanent error
@@ -506,7 +508,8 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
withAgent 3 agentCfg initAgentServers testDB2 $ \rcp2 -> runRight $ do
xftpStartWorkers rcp2 (Just recipientFiles)
rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp2
("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
rfGet rcp2
liftIO $ rfId' `shouldBe` rfId
testXFTPAgentDeleteRestore :: HasCallStack => IO ()
@@ -543,7 +546,8 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
withAgent 5 agentCfg initAgentServers testDB3 $ \rcp2 -> runRight $ do
xftpStartWorkers rcp2 (Just recipientFiles)
rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp2
("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
rfGet rcp2
liftIO $ rfId' `shouldBe` rfId
testXFTPAgentDeleteOnServer :: HasCallStack => IO ()
@@ -577,7 +581,8 @@ testXFTPAgentDeleteOnServer = withGlobalLogging logCfgNoLogs $
runRight_ . void $ do
-- receive file 1 again
rfId1 <- xftpReceiveFile rcp 1 rfd1_2 Nothing
("", rfId1', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp
("", rfId1', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
rfGet rcp
liftIO $ rfId1 `shouldBe` rfId1'
-- receive file 2
@@ -609,7 +614,8 @@ testXFTPAgentExpiredOnServer = withGlobalLogging logCfgNoLogs $ do
-- receive file 1 again - should fail with AUTH error
runRight $ do
rfId <- xftpReceiveFile rcp 1 rfd1_2 Nothing
("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp
("", rfId', RFERR (INTERNAL "XFTP {serverAddress = \"xftp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:7000\", xftpErr = AUTH}")) <-
rfGet rcp
liftIO $ rfId' `shouldBe` rfId
-- create and send file 2