simplexmq 1.0.2 → 1.1.0
raw patch · 36 files changed
+1525/−826 lines, 36 filesdep ~aesondep ~attoparsecdep ~network
Dependency ranges changed: aeson, attoparsec, network
Files
- CHANGELOG.md +37/−0
- README.md +22/−2
- apps/smp-agent/Main.hs +2/−2
- apps/smp-server/Main.hs +12/−8
- simplexmq.cabal +32/−13
- src/Simplex/Messaging/Agent.hs +125/−143
- src/Simplex/Messaging/Agent/Client.hs +141/−83
- src/Simplex/Messaging/Agent/Env/SQLite.hs +9/−7
- src/Simplex/Messaging/Agent/Protocol.hs +126/−82
- src/Simplex/Messaging/Agent/Server.hs +81/−0
- src/Simplex/Messaging/Agent/Store.hs +7/−4
- src/Simplex/Messaging/Agent/Store/SQLite.hs +38/−42
- src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs +3/−1
- src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220301_snd_queue_keys.hs +13/−0
- src/Simplex/Messaging/Client.hs +32/−28
- src/Simplex/Messaging/Parsers.hs +46/−1
- src/Simplex/Messaging/Protocol.hs +34/−5
- src/Simplex/Messaging/Server.hs +62/−39
- src/Simplex/Messaging/Server/Env/STM.hs +25/−18
- src/Simplex/Messaging/Server/MsgStore.hs +2/−0
- src/Simplex/Messaging/Server/MsgStore/STM.hs +20/−13
- src/Simplex/Messaging/Server/QueueStore.hs +1/−0
- src/Simplex/Messaging/Server/QueueStore/STM.hs +57/−80
- src/Simplex/Messaging/TMap.hs +70/−0
- src/Simplex/Messaging/Transport.hs +23/−150
- src/Simplex/Messaging/Transport/Client.hs +84/−0
- src/Simplex/Messaging/Transport/KeepAlive.hs +62/−0
- src/Simplex/Messaging/Transport/Server.hs +107/−0
- src/Simplex/Messaging/Util.hs +4/−0
- tests/AgentTests.hs +105/−66
- tests/AgentTests/FunctionalAPITests.hs +46/−20
- tests/AgentTests/SQLiteTests.hs +7/−3
- tests/CoreTests/ProtocolErrorTests.hs +4/−3
- tests/SMPAgentClient.hs +6/−4
- tests/SMPClient.hs +12/−7
- tests/ServerTests.hs +68/−2
CHANGELOG.md view
@@ -1,24 +1,58 @@+# 1.1.0++SMP server:++- message TTL and periodic deletion of old messages+- configuration to prevent creation of the new queues++SMP agent:++- asynchronous connection handshake+- configurable SMP servers at run-time+- use TCP keep-alive for connection stability+- improve stability of connection subscriptions+- auto-vacuum DB to remove deleted records++# 1.0.3++SMP server:++- Reduce server message queue quota to 128 messages.++SMP agent:++- Add "yes to migrations" option.+- Make new SMP client attempt to reconnect on network error.+- Reduce connection handshake expiration to 2 days.++JSON encoding of types used in simplex-chat, some other minor adjustments.+ # 1.0.2 General:+ - Enable TLS 1.3 parameters for TLS handshake (server and client). - Switch from hs-tls fork to original repo now that it supports getFinished and getPeerFinished APIs for both TLS 1.2 and TLS 1.3. SMP server:+ - Perform TLS handshake in a separate thread per-connection. SMP agent:+ - Cease attempts to send HELLO after one week timeout. - Coalesce requests to connect to SMP servers, to have 1 connection per server. # 1.0.1 SMP server:+ - Explicitly set line buffering in stdout/stderr to log each line when output is redirected to files. # 1.0.0 Security and privacy improvements:+ - Faster and more secure 2-layer E2E encryption with additional encryption layer between servers and recipients: - application messages in each duplex connection (managed by SMP agents - see [overview](https://github.com/simplex-chat/simplexmq/blob/master/protocol/overview-tjr.md)) are encrypted using [double-ratchet algorithm](https://www.signal.org/docs/specifications/doubleratchet/), providing forward secrecy and break-in recovery. This layer uses two Curve448 keys per client for [X3DH key agreement](https://www.signal.org/docs/specifications/x3dh/), SHA512 based HKDFs and AES-GCM AEAD encryption. - SMP client messages are additionally E2E encrypted in each SMP queue to avoid cipher-text correlation of messages sent via multiple redundant queues (that will be supported soon). This and the next layer use [NaCl crypto_box algorithm](https://nacl.cr.yp.to/index.html) with XSalsa20Poly1305 cipher and Curve25519 keys for DH key agreement.@@ -31,13 +65,16 @@ - Server identity verification via server offline certificate fingerprints included in SMP server addresses. New functionality:+ - Support for notification servers with new SMP commands: `NKEY`/`NID`, `NSUB`/`NMSG`. Efficiency improvements:+ - Binary protocol encodings to reduce overhead from circa 15% to approximately 3.7% of transmitted application message size, with only 2.2% overhead for SMP protocol messages. - More performant cryptographic algorithms. For more information about SimpleX:+ - [SimpleX overview](https://github.com/simplex-chat/simplexmq/blob/master/protocol/overview-tjr.md). - [SimpleX chat v1 announcement](https://github.com/simplex-chat/simplex-chat/blob/master/blog/20220112-simplex-chat-v1-released.md).
README.md view
@@ -27,9 +27,9 @@ ### SMP server -[SMP server](https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-server/Main.hs) can be run on any Linux distribution without any dependencies, including low power/low memory devices.+[SMP server](https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-server/Main.hs) can be run on any Linux distribution, including low power/low memory devices. OpenSSL library is required for initialization. -To initialize the server use `smp-server init` command - it will generate keys and certificates for TLS transport. The fingerprint of offline certificate is used as part of the server address to protect client/server connection against man-in-the-middle attacks: `smp://<fingerprint>@<hostname>[:5223]`.+To initialize the server use `smp-server init -n <fqdn>` (or `smp-server init --ip <ip>` for IP based address) command - it will generate keys and certificates for TLS transport. The fingerprint of offline certificate is used as part of the server address to protect client/server connection against man-in-the-middle attacks: `smp://<fingerprint>@<hostname>[:5223]`. SMP server uses in-memory persistence with an optional append-only log of created queues that allows to re-start the server without losing the connections. This log is compacted on every server restart, permanently removing suspended and removed queues. @@ -86,6 +86,26 @@ `smp://PQUV2eL0t7OStZOoAsPEV2QYWt4-xilbakvGUGOItUo=@smp6.simplex.im` It's the easiest to try SMP agent via a prototype [simplex-chat](https://github.com/simplex-chat/simplex-chat) terminal UI.++## Deploy SMP server on Linux++You can run your SMP server as a Linux process, optionally using a service manager for booting and restarts.++- For Ubuntu you can download a binary from [the latest release](https://github.com/simplex-chat/simplexmq/releases).++ If you're using other Linux distribution and the binary is incompatible with it, you can build from source using [Haskell stack](https://docs.haskellstack.org/en/stable/README/):++ ```shell+ curl -sSL https://get.haskellstack.org/ | sh+ ...+ stack install+ ```++- Initialize SMP server with `smp-server init [-l] -n <fqdn>` or `smp-server init [-l] --ip <ip>` - depending on how you initialize it, either FQDN or IP will be used for server's address.++- Run `smp-server start` to start SMP server, or you can configure a service manager to run it as a service.++See [this section](#smp-server) for more information. Run `smp-server -h` and `smp-server init -h` for explanation of commands and options. [<img alt="Linode" src="https://raw.githubusercontent.com/simplex-chat/simplexmq/master/img/linode.svg" align="right" width="200">](https://cloud.linode.com/stackscripts/748014)
apps/smp-agent/Main.hs view
@@ -6,12 +6,12 @@ import Control.Logger.Simple import qualified Data.List.NonEmpty as L-import Simplex.Messaging.Agent (runSMPAgent) import Simplex.Messaging.Agent.Env.SQLite+import Simplex.Messaging.Agent.Server (runSMPAgent) import Simplex.Messaging.Transport (TLS, Transport (..)) cfg :: AgentConfig-cfg = defaultAgentConfig {smpServers = L.fromList ["smp://bU0K-bRg24xWW__lS0umO1Zdw_SXqpJNtm1_RrPLViE=@localhost:5223"]}+cfg = defaultAgentConfig {initialSMPServers = L.fromList ["smp://bU0K-bRg24xWW__lS0umO1Zdw_SXqpJNtm1_RrPLViE=@localhost:5223"]} logCfg :: LogConfig logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
apps/smp-server/Main.hs view
@@ -3,6 +3,7 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeApplications #-} @@ -22,12 +23,13 @@ import Simplex.Messaging.Server (runSMPServer) import Simplex.Messaging.Server.Env.STM import Simplex.Messaging.Server.StoreLog (StoreLog, openReadStoreLog, storeLogFilePath)-import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), loadFingerprint, simplexMQVersion)+import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), simplexMQVersion)+import Simplex.Messaging.Transport.Server (loadFingerprint) import Simplex.Messaging.Transport.WebSockets (WS) import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, removeDirectoryRecursive) import System.Exit (exitFailure) import System.FilePath (combine)-import System.IO (BufferMode (..), IOMode (..), hGetLine, withFile, hSetBuffering, stderr, stdout)+import System.IO (BufferMode (..), IOMode (..), hGetLine, hSetBuffering, stderr, stdout, withFile) import System.Process (readCreateProcess, shell) import Text.Read (readMaybe) @@ -130,8 +132,7 @@ <*> strOption ( long "ip" <> help- "Server IP address used as Subject Alternative Name for TLS online certificate, \- \also used as Common Name if FQDN is not supplied"+ "Server IP address, used as Common Name for TLS online certificate if FQDN is not supplied" <> value "127.0.0.1" <> showDefault <> metavar "IP"@@ -139,7 +140,7 @@ <*> (optional . strOption) ( long "fqdn" <> short 'n'- <> help "Server FQDN used as Common Name and Subject Alternative Name for TLS online certificate"+ <> help "Server FQDN used as Common Name for TLS online certificate" <> showDefault <> metavar "FQDN" )@@ -276,14 +277,17 @@ ServerConfig { transports = (port, transport @TLS) : [("80", transport @WS) | enableWebsockets], tbqSize = 16,- serverTbqSize = 128,- msgQueueQuota = 256,+ serverTbqSize = 64,+ msgQueueQuota = 128, queueIdBytes = 24, msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20 caCertificateFile = caCrtFile, privateKeyFile = serverKeyFile, certificateFile = serverCrtFile,- storeLog+ storeLog,+ allowNewQueues = True,+ messageTTL = Just $ 7 * 86400, -- 7 days+ expireMessagesInterval = Just 21600_000000 -- microseconds, 6 hours } openStoreLog :: IO (Maybe (StoreLog 'ReadMode))
simplexmq.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: simplexmq-version: 1.0.2+version: 1.1.0 synopsis: SimpleXMQ message broker description: This package includes <./docs/Simplex-Messaging-Server.html server>, <./docs/Simplex-Messaging-Client.html client> and@@ -27,6 +27,11 @@ README.md CHANGELOG.md +flag swift+ description: Enable swift JSON format+ manual: True+ default: False+ library exposed-modules: Simplex.Messaging.Agent@@ -35,10 +40,12 @@ Simplex.Messaging.Agent.Protocol Simplex.Messaging.Agent.QueryString Simplex.Messaging.Agent.RetryInterval+ Simplex.Messaging.Agent.Server Simplex.Messaging.Agent.Store Simplex.Messaging.Agent.Store.SQLite Simplex.Messaging.Agent.Store.SQLite.Migrations Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial+ Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys Simplex.Messaging.Client Simplex.Messaging.Crypto Simplex.Messaging.Crypto.Ratchet@@ -53,7 +60,11 @@ Simplex.Messaging.Server.QueueStore Simplex.Messaging.Server.QueueStore.STM Simplex.Messaging.Server.StoreLog+ Simplex.Messaging.TMap Simplex.Messaging.Transport+ Simplex.Messaging.Transport.Client+ Simplex.Messaging.Transport.KeepAlive+ Simplex.Messaging.Transport.Server Simplex.Messaging.Transport.WebSockets Simplex.Messaging.Util Simplex.Messaging.Version@@ -64,12 +75,12 @@ ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns build-depends: QuickCheck ==2.14.*- , aeson ==1.5.*+ , aeson ==2.0.* , ansi-terminal >=0.10 && <0.12 , asn1-encoding ==0.9.* , asn1-types ==0.3.* , async ==2.2.*- , attoparsec ==0.13.*+ , attoparsec ==0.14.* , base >=4.7 && <5 , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.*@@ -87,7 +98,7 @@ , iso8601-time ==0.1.* , memory ==0.15.* , mtl ==2.2.*- , network ==3.1.*+ , network ==3.1.2.* , network-transport ==0.5.* , random >=1.1 && <1.3 , simple-logger ==0.1.*@@ -104,6 +115,8 @@ , x509 ==1.7.* , x509-store ==1.6.* , x509-validation ==1.6.*+ if flag(swift)+ cpp-options: -DswiftJSON default-language: Haskell2010 executable smp-agent@@ -115,12 +128,12 @@ ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded build-depends: QuickCheck ==2.14.*- , aeson ==1.5.*+ , aeson ==2.0.* , ansi-terminal >=0.10 && <0.12 , asn1-encoding ==0.9.* , asn1-types ==0.3.* , async ==2.2.*- , attoparsec ==0.13.*+ , attoparsec ==0.14.* , base >=4.7 && <5 , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.*@@ -138,7 +151,7 @@ , iso8601-time ==0.1.* , memory ==0.15.* , mtl ==2.2.*- , network ==3.1.*+ , network ==3.1.2.* , network-transport ==0.5.* , random >=1.1 && <1.3 , simple-logger ==0.1.*@@ -156,6 +169,8 @@ , x509 ==1.7.* , x509-store ==1.6.* , x509-validation ==1.6.*+ if flag(swift)+ cpp-options: -DswiftJSON default-language: Haskell2010 executable smp-server@@ -167,12 +182,12 @@ ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded build-depends: QuickCheck ==2.14.*- , aeson ==1.5.*+ , aeson ==2.0.* , ansi-terminal >=0.10 && <0.12 , asn1-encoding ==0.9.* , asn1-types ==0.3.* , async ==2.2.*- , attoparsec ==0.13.*+ , attoparsec ==0.14.* , base >=4.7 && <5 , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.*@@ -191,7 +206,7 @@ , iso8601-time ==0.1.* , memory ==0.15.* , mtl ==2.2.*- , network ==3.1.*+ , network ==3.1.2.* , network-transport ==0.5.* , optparse-applicative >=0.15 && <0.17 , process ==1.6.*@@ -211,6 +226,8 @@ , x509 ==1.7.* , x509-store ==1.6.* , x509-validation ==1.6.*+ if flag(swift)+ cpp-options: -DswiftJSON default-language: Haskell2010 test-suite smp-server-test@@ -235,12 +252,12 @@ build-depends: HUnit ==1.6.* , QuickCheck ==2.14.*- , aeson ==1.5.*+ , aeson ==2.0.* , ansi-terminal >=0.10 && <0.12 , asn1-encoding ==0.9.* , asn1-types ==0.3.* , async ==2.2.*- , attoparsec ==0.13.*+ , attoparsec ==0.14.* , base >=4.7 && <5 , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.*@@ -260,7 +277,7 @@ , iso8601-time ==0.1.* , memory ==0.15.* , mtl ==2.2.*- , network ==3.1.*+ , network ==3.1.2.* , network-transport ==0.5.* , random >=1.1 && <1.3 , simple-logger ==0.1.*@@ -279,4 +296,6 @@ , x509 ==1.7.* , x509-store ==1.6.* , x509-validation ==1.6.*+ if flag(swift)+ cpp-options: -DswiftJSON default-language: Haskell2010
src/Simplex/Messaging/Agent.hs view
@@ -26,11 +26,7 @@ -- -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md module Simplex.Messaging.Agent- ( -- * SMP agent over TCP- runSMPAgent,- runSMPAgentBlocking,-- -- * queue-based SMP agent+ ( -- * queue-based SMP agent getAgentClient, runAgentClient, @@ -51,6 +47,8 @@ ackMessage, suspendConnection, deleteConnection,+ setSMPServers,+ logConnection, ) where @@ -62,7 +60,6 @@ import Crypto.Random (MonadRandom) import Data.Bifunctor (first, second) import Data.ByteString.Char8 (ByteString)-import qualified Data.ByteString.Char8 as B import Data.Composition ((.:), (.:.)) import Data.Functor (($>)) import Data.List.NonEmpty (NonEmpty (..))@@ -70,7 +67,6 @@ import qualified Data.Map.Strict as M import Data.Maybe (isJust) import qualified Data.Text as T-import Data.Text.Encoding (decodeUtf8) import Data.Time.Clock import Data.Time.Clock.System (systemToUTCTime) import Database.SQLite.Simple (SQLError)@@ -80,14 +76,14 @@ import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Agent.Store import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)-import Simplex.Messaging.Client (SMPServerTransmission)+import Simplex.Messaging.Client (SMPClient (..), SMPServerTransmission) import qualified Simplex.Messaging.Crypto as C import qualified Simplex.Messaging.Crypto.Ratchet as CR import Simplex.Messaging.Encoding import Simplex.Messaging.Parsers (parse) import Simplex.Messaging.Protocol (MsgBody) import qualified Simplex.Messaging.Protocol as SMP-import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), loadTLSServerParams, runTransportServer, simplexMQVersion)+import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Util (bshow, liftError, tryError, unlessM) import Simplex.Messaging.Version import System.Random (randomR)@@ -95,33 +91,6 @@ import qualified UnliftIO.Exception as E import UnliftIO.STM --- | Runs an SMP agent as a TCP service using passed configuration.------ See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs-runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> m ()-runSMPAgent t cfg = do- started <- newEmptyTMVarIO- runSMPAgentBlocking t started cfg---- | Runs an SMP agent as a TCP service using passed configuration with signalling.------ This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)--- and when it is disconnected from the TCP socket once the server thread is killed (False).-runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()-runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do- runReaderT (smpAgent t) =<< newSMPAgentEnv cfg- where- smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()- smpAgent _ = do- -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation- tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile- runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do- liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion- c <- getAgentClient- logConnection c True- race_ (connectClient h c) (runAgentClient c)- `E.finally` disconnectAgentClient c- -- | Creates an SMP agent client instance getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> m AgentClient getSMPAgentClient cfg = newSMPAgentEnv cfg >>= runReaderT runAgent@@ -176,6 +145,10 @@ deleteConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m () deleteConnection c = withAgentEnv c . deleteConnection' c +-- | Change servers to be used for creating new queues+setSMPServers :: AgentErrorMonad m => AgentClient -> NonEmpty SMPServer -> m ()+setSMPServers c = withAgentEnv c . setSMPServers' c+ withAgentEnv :: AgentClient -> ReaderT Env m a -> m a withAgentEnv c = (`runReaderT` agentEnv c) @@ -186,9 +159,6 @@ getAgentClient :: (MonadUnliftIO m, MonadReader Env m) => m AgentClient getAgentClient = ask >>= atomically . newAgentClient -connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()-connectClient h c = race_ (send h c) (receive h c)- logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m () logConnection c connected = let event = if connected then "connected to" else "disconnected from"@@ -198,28 +168,6 @@ runAgentClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m () runAgentClient c = race_ (subscriber c) (client c) -receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()-receive h c@AgentClient {rcvQ, subQ} = forever $ do- (corrId, connId, cmdOrErr) <- tGet SClient h- case cmdOrErr of- Right cmd -> write rcvQ (corrId, connId, cmd)- Left e -> write subQ (corrId, connId, ERR e)- where- write :: TBQueue (ATransmission p) -> ATransmission p -> m ()- write q t = do- logClient c "-->" t- atomically $ writeTBQueue q t--send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()-send h c@AgentClient {subQ} = forever $ do- t <- atomically $ readTBQueue subQ- tPut h t- logClient c "<--" t--logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()-logClient AgentClient {clientId} dir (corrId, connId, cmd) = do- logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]- client :: forall m. (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m () client c@AgentClient {rcvQ, subQ} = forever $ do (corrId, connId, cmd) <- atomically $ readTBQueue rcvQ@@ -267,7 +215,7 @@ newConn :: AgentMonad m => AgentClient -> ConnId -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c) newConn c connId cMode = do- srv <- getSMPServer+ srv <- getSMPServer c (rq, qUri) <- newRcvQueue c srv g <- asks idsDrg let cData = ConnData {connId}@@ -292,16 +240,21 @@ (pk1, pk2, e2eSndParams) <- liftIO . CR.generateE2EParams $ version e2eRcvParams (_, rcDHRs) <- liftIO C.generateKeyPair' let rc = CR.initSndRatchet rcDHRr rcDHRs $ CR.x3dhSnd pk1 pk2 e2eRcvParams- (sq, smpConf) <- newSndQueue qInfo cInfo+ sq <- newSndQueue qInfo g <- asks idsDrg let cData = ConnData {connId} connId' <- withStore $ \st -> do connId' <- createSndConn st g cData sq createRatchet st connId' rc pure connId'- confirmQueue c connId' sq smpConf $ Just e2eSndParams- void $ enqueueMessage c connId' sq HELLO- pure connId'+ tryError (confirmQueue c connId' sq cInfo $ Just e2eSndParams) >>= \case+ Right _ -> do+ void $ enqueueMessage c connId' sq HELLO+ pure connId'+ Left e -> do+ -- TODO recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md+ withStore (`deleteConn` connId')+ throwError e _ -> throwError $ AGENT A_VERSION joinConn c connId (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo = case ( qUri `compatibleVersion` SMP.smpClientVRange,@@ -316,7 +269,7 @@ createReplyQueue :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> m () createReplyQueue c connId sq = do- srv <- getSMPServer+ srv <- getSMPServer c (rq, qUri) <- newRcvQueue c srv -- TODO reply queue version should be the same as send queue, ignoring it in v1 let qInfo = toVersionT qUri SMP.smpClientVersion@@ -363,16 +316,6 @@ SomeConn _ (DuplexConnection _ rq sq) -> do resumeMsgDelivery c connId sq subscribeQueue c rq connId- case status (sq :: SndQueue) of- Confirmed -> do- -- TODO if there is no confirmation saved, just update the status without securing the queue- AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <-- withStore (`getAcceptedConfirmation` connId)- secureQueue c rq senderKey- withStore $ \st -> setRcvQueueStatus st rq Secured- Secured -> pure ()- Active -> pure ()- _ -> throwError $ INTERNAL "unexpected queue status" SomeConn _ (SndConnection _ sq) -> do resumeMsgDelivery c connId sq case status (sq :: SndQueue) of@@ -405,12 +348,12 @@ internalTs <- liftIO getCurrentTime (internalId, internalSndId, prevMsgHash) <- withStore (`updateSndIds` connId) let privHeader = APrivHeader (unSndId internalSndId) prevMsgHash- agentMessage = smpEncode $ AgentMessage privHeader aMessage- internalHash = C.sha256Hash agentMessage-- encAgentMessage <- agentRatchetEncrypt connId agentMessage e2eEncUserMsgLength+ agentMsg = AgentMessage privHeader aMessage+ agentMsgStr = smpEncode agentMsg+ internalHash = C.sha256Hash agentMsgStr+ encAgentMessage <- agentRatchetEncrypt connId agentMsgStr e2eEncUserMsgLength let msgBody = smpEncode $ AgentMsgEnvelope {agentVersion = smpAgentVersion, encAgentMessage}- msgType = aMessageType aMessage+ msgType = agentMessageType agentMsg msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, internalHash, prevMsgHash} withStore $ \st -> createSndMsg st connId msgData pure internalId@@ -420,18 +363,13 @@ let qKey = (connId, server, sndId) unlessM (queueDelivering qKey) $ async (runSmpQueueMsgDelivery c connId sq)- >>= atomically . modifyTVar (smpQueueMsgDeliveries c) . M.insert qKey+ >>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c) unlessM connQueued $ withStore (`getPendingMsgs` connId) >>= queuePendingMsgs c connId sq where- queueDelivering qKey = isJust . M.lookup qKey <$> readTVarIO (smpQueueMsgDeliveries c)- connQueued =- atomically $- isJust- <$> stateTVar- (connMsgsQueued c)- (\m -> (M.lookup connId m, M.insert connId True m))+ queueDelivering qKey = atomically $ isJust <$> TM.lookup qKey (smpQueueMsgDeliveries c)+ connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connMsgsQueued c) queuePendingMsgs :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> [InternalId] -> m () queuePendingMsgs c connId sq msgIds = atomically $ do@@ -441,11 +379,11 @@ getPendingMsgQ :: AgentClient -> ConnId -> SndQueue -> STM (TQueue InternalId) getPendingMsgQ c connId SndQueue {server, sndId} = do let qKey = (connId, server, sndId)- maybe (newMsgQueue qKey) pure . M.lookup qKey =<< readTVar (smpQueueMsgQueues c)+ maybe (newMsgQueue qKey) pure =<< TM.lookup qKey (smpQueueMsgQueues c) where newMsgQueue qKey = do mq <- newTQueue- modifyTVar (smpQueueMsgQueues c) $ M.insert qKey mq+ TM.insert qKey mq $ smpQueueMsgQueues c pure mq runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> m ()@@ -460,30 +398,38 @@ notify $ MERR mId (INTERNAL $ show e) Right (rq_, (msgType, msgBody, internalTs)) -> withRetryInterval ri $ \loop ->- tryError (sendAgentMessage c sq msgBody) >>= \case+ tryError (send msgType c sq msgBody) >>= \case Left e -> do+ let err = if msgType == AM_CONN_INFO then ERR e else MERR mId e case e of- SMP SMP.QUOTA -> loop+ SMP SMP.QUOTA -> case msgType of+ AM_CONN_INFO -> connError msgId NOT_AVAILABLE+ _ -> loop SMP SMP.AUTH -> case msgType of- HELLO_ -> do+ AM_CONN_INFO -> connError msgId NOT_AVAILABLE+ AM_HELLO_ -> do helloTimeout <- asks $ helloTimeout . config currentTime <- liftIO getCurrentTime if diffUTCTime currentTime internalTs > helloTimeout then case rq_ of -- party initiating connection- Just _ -> notifyDel msgId . ERR $ CONN NOT_AVAILABLE+ Just _ -> connError msgId NOT_AVAILABLE -- party joining connection- _ -> notifyDel msgId . ERR $ CONN NOT_ACCEPTED+ _ -> connError msgId NOT_ACCEPTED else loop- REPLY_ -> notifyDel msgId $ ERR e- A_MSG_ -> notifyDel msgId $ MERR mId e- SMP (SMP.CMD _) -> notifyDel msgId $ MERR mId e- SMP SMP.LARGE_MSG -> notifyDel msgId $ MERR mId e- SMP {} -> notify (MERR mId e) >> loop+ AM_REPLY_ -> notifyDel msgId $ ERR e+ AM_A_MSG_ -> notifyDel msgId $ MERR mId e+ SMP (SMP.CMD _) -> notifyDel msgId err+ SMP SMP.LARGE_MSG -> notifyDel msgId err+ SMP {} -> notify err >> loop _ -> loop Right () -> do case msgType of- HELLO_ -> do+ AM_CONN_INFO -> do+ withStore $ \st -> setSndQueueStatus st sq Confirmed+ when (isJust rq_) $ withStore (`removeConfirmations` connId)+ void $ enqueueMessage c connId sq HELLO+ AM_HELLO_ -> do withStore $ \st -> setSndQueueStatus st sq Active case rq_ of -- party initiating connection@@ -492,16 +438,20 @@ notify CON -- party joining connection _ -> createReplyQueue c connId sq- A_MSG_ -> notify $ SENT mId+ AM_A_MSG_ -> notify $ SENT mId _ -> pure () delMsg msgId where+ send = \case+ AM_CONN_INFO -> sendConfirmation+ _ -> sendAgentMessage delMsg :: InternalId -> m () delMsg msgId = withStore $ \st -> deleteMsg st connId msgId notify :: ACommand 'Agent -> m () notify cmd = atomically $ writeTBQueue subQ ("", connId, cmd) notifyDel :: InternalId -> ACommand 'Agent -> m () notifyDel msgId cmd = notify cmd >> delMsg msgId+ connError msgId = notifyDel msgId . ERR . CONN ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m () ackMessage' c connId msgId = do@@ -537,12 +487,18 @@ delete :: RcvQueue -> m () delete rq = do deleteQueue c rq- removeSubscription c connId+ atomically $ removeSubscription c connId withStore (`deleteConn` connId) -getSMPServer :: AgentMonad m => m SMPServer-getSMPServer =- asks (smpServers . config) >>= \case+-- | Change servers to be used for creating new queues, in Reader monad+setSMPServers' :: forall m. AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()+setSMPServers' c servers = do+ atomically $ writeTVar (smpServers c) servers++getSMPServer :: AgentMonad m => AgentClient -> m SMPServer+getSMPServer c = do+ smpServers <- readTVarIO $ smpServers c+ case smpServers of srv :| [] -> pure srv servers -> do gen <- asks randomServer@@ -557,7 +513,7 @@ Right _ -> return () processSMPTransmission :: forall m. AgentMonad m => AgentClient -> SMPServerTransmission -> m ()-processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do+processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd) = do withStore (\st -> getRcvConn st srv rId) >>= \case SomeConn SCDuplex (DuplexConnection cData rq _) -> processSMP SCDuplex cData rq SomeConn SCRcv (RcvConnection cData rq) -> processSMP SCRcv cData rq@@ -587,8 +543,9 @@ (SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) -> do agentMsgBody <- agentRatchetDecrypt connId encAgentMsg parseMessage agentMsgBody >>= \case- AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage -> do- (msgId, msgMeta) <- agentClientMsg prevMsgHash sndMsgId (srvMsgId, systemToUTCTime srvTs) agentMsgBody aMessage+ agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do+ let msgType = agentMessageType agentMsg+ (msgId, msgMeta) <- agentClientMsg prevMsgHash sndMsgId (srvMsgId, systemToUTCTime srvTs) agentMsgBody msgType case aMessage of HELLO -> helloMsg >> ack >> withStore (\st -> deleteMsg st connId msgId) REPLY cReq -> replyMsg cReq >> ack >> withStore (\st -> deleteMsg st connId msgId)@@ -597,10 +554,19 @@ _ -> prohibited >> ack _ -> prohibited >> ack _ -> prohibited >> ack- SMP.END -> do- removeSubscription c connId- logServer "<--" c srv rId "END"- notify END+ SMP.END ->+ atomically (TM.lookup srv smpClients >>= fmap join . mapM tryReadTMVar >>= processEND)+ >>= logServer "<--" c srv rId+ where+ processEND = \case+ Just (Right clnt)+ | sessId == sessionId clnt -> do+ removeSubscription c connId+ writeTBQueue subQ ("", connId, END)+ pure "END"+ | otherwise -> ignored+ _ -> ignored+ ignored = pure "END from disconnected client - ignored" _ -> do logServer "<--" c srv rId $ "unexpected: " <> bshow cmd notify . ERR $ BROKER UNEXPECTED@@ -678,15 +644,13 @@ case qInfo `proveCompatible` SMP.smpClientVRange of Nothing -> notify . ERR $ AGENT A_VERSION Just qInfo' -> do- (sq, smpConf) <- newSndQueue qInfo' ownConnInfo+ sq <- newSndQueue qInfo' withStore $ \st -> upgradeRcvConnToDuplex st connId sq- confirmQueue c connId sq smpConf Nothing- withStore (`removeConfirmations` connId)- void $ enqueueMessage c connId sq HELLO+ enqueueConfirmation c connId sq ownConnInfo Nothing _ -> prohibited - agentClientMsg :: PrevRcvMsgHash -> ExternalSndId -> (BrokerId, BrokerTs) -> MsgBody -> AMessage -> m (InternalId, MsgMeta)- agentClientMsg externalPrevSndHash sndMsgId broker msgBody aMessage = do+ agentClientMsg :: PrevRcvMsgHash -> ExternalSndId -> (BrokerId, BrokerTs) -> MsgBody -> AgentMessageType -> m (InternalId, MsgMeta)+ agentClientMsg externalPrevSndHash sndMsgId broker msgBody msgType = do logServer "<--" c srv rId "MSG <MSG>" let internalHash = C.sha256Hash msgBody internalTs <- liftIO getCurrentTime@@ -694,7 +658,6 @@ let integrity = checkMsgIntegrity prevExtSndId sndMsgId prevRcvMsgHash externalPrevSndHash recipient = (unId internalId, internalTs) msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}- msgType = aMessageType aMessage rcvMsg = RcvMsgData {msgMeta, msgType, msgBody, internalRcvId, internalHash, externalPrevSndHash} withStore $ \st -> createRcvMsg st connId rcvMsg pure (internalId, msgMeta)@@ -719,8 +682,9 @@ | internalPrevMsgHash /= receivedPrevMsgHash = MsgError MsgBadHash | otherwise = MsgError MsgDuplicate -- this case is not possible -confirmQueue :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> SMPConfirmation -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()-confirmQueue c connId sq SMPConfirmation {senderKey, e2ePubKey, connInfo} e2eEncryption = do+confirmQueue :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()+confirmQueue c connId sq connInfo e2eEncryption = do+ _ <- withStore (`updateSndIds` connId) msg <- mkConfirmation sendConfirmation c sq msg withStore $ \st -> setSndQueueStatus st sq Confirmed@@ -728,10 +692,28 @@ mkConfirmation :: m MsgBody mkConfirmation = do encConnInfo <- agentRatchetEncrypt connId (smpEncode $ AgentConnInfo connInfo) e2eEncConnInfoLength- let agentEnvelope = AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}- agentCbEncrypt sq (Just e2ePubKey) . smpEncode $- SMP.ClientMessage (SMP.PHConfirmation senderKey) $ smpEncode agentEnvelope+ pure . smpEncode $ AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo} +enqueueConfirmation :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()+enqueueConfirmation c connId sq connInfo e2eEncryption = do+ resumeMsgDelivery c connId sq+ msgId <- storeConfirmation+ queuePendingMsgs c connId sq [msgId]+ where+ storeConfirmation :: m InternalId+ storeConfirmation = do+ internalTs <- liftIO getCurrentTime+ (internalId, internalSndId, prevMsgHash) <- withStore (`updateSndIds` connId)+ let agentMsg = AgentConnInfo connInfo+ agentMsgStr = smpEncode agentMsg+ internalHash = C.sha256Hash agentMsgStr+ encConnInfo <- agentRatchetEncrypt connId agentMsgStr e2eEncConnInfoLength+ let msgBody = smpEncode $ AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}+ msgType = agentMessageType agentMsg+ msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, internalHash, prevMsgHash}+ withStore $ \st -> createSndMsg st connId msgData+ pure internalId+ -- encoded AgentMessage -> encoded EncAgentMessage agentRatchetEncrypt :: AgentMonad m => ConnId -> ByteString -> Int -> m ByteString agentRatchetEncrypt connId msg paddedLen = do@@ -752,27 +734,27 @@ notifyConnected :: AgentMonad m => AgentClient -> ConnId -> m () notifyConnected c connId = atomically $ writeTBQueue (subQ c) ("", connId, CON) -newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> ConnInfo -> m (SndQueue, SMPConfirmation)-newSndQueue qInfo cInfo =+newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> m SndQueue+newSndQueue qInfo = asks (cmdSignAlg . config) >>= \case- C.SignAlg a -> newSndQueue_ a qInfo cInfo+ C.SignAlg a -> newSndQueue_ a qInfo newSndQueue_ :: (C.SignatureAlgorithm a, C.AlgorithmI a, MonadUnliftIO m) => C.SAlgorithm a -> Compatible SMPQueueInfo ->- ConnInfo ->- m (SndQueue, SMPConfirmation)-newSndQueue_ a (Compatible (SMPQueueInfo _clientVersion smpServer senderId rcvE2ePubDhKey)) cInfo = do+ m SndQueue+newSndQueue_ a (Compatible (SMPQueueInfo _clientVersion smpServer senderId rcvE2ePubDhKey)) = do -- this function assumes clientVersion is compatible - it was tested before- (senderKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair a+ (sndPublicKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair a (e2ePubKey, e2ePrivKey) <- liftIO C.generateKeyPair'- let sndQueue =- SndQueue- { server = smpServer,- sndId = senderId,- sndPrivateKey,- e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,- status = New- }- pure (sndQueue, SMPConfirmation senderKey e2ePubKey cInfo)+ pure+ SndQueue+ { server = smpServer,+ sndId = senderId,+ sndPublicKey = Just sndPublicKey,+ sndPrivateKey,+ e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,+ e2ePubKey = Just e2ePubKey,+ status = New+ }
src/Simplex/Messaging/Agent/Client.hs view
@@ -37,7 +37,7 @@ where import Control.Concurrent (forkIO)-import Control.Concurrent.Async (Async, async, uninterruptibleCancel)+import Control.Concurrent.Async (Async, uninterruptibleCancel) import Control.Concurrent.STM (stateTVar) import Control.Logger.Simple import Control.Monad.Except@@ -47,11 +47,10 @@ import Data.ByteString.Base64 import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B+import Data.List.NonEmpty (NonEmpty) import Data.Map.Strict (Map) import qualified Data.Map.Strict as M import Data.Maybe (isNothing)-import Data.Set (Set)-import qualified Data.Set as S import Data.Text.Encoding import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol@@ -62,8 +61,12 @@ import Simplex.Messaging.Encoding import Simplex.Messaging.Protocol (QueueId, QueueIdsKeys (..), SndPublicVerifyKey) import qualified Simplex.Messaging.Protocol as SMP-import Simplex.Messaging.Util (bshow, liftEitherError, liftError, liftIOEither, tryError)+import Simplex.Messaging.TMap (TMap)+import qualified Simplex.Messaging.TMap as TM+import Simplex.Messaging.Util (bshow, liftEitherError, liftError, tryError, whenM) import Simplex.Messaging.Version+import System.Timeout (timeout)+import UnliftIO (async, forConcurrently_) import UnliftIO.Exception (Exception, IOException) import qualified UnliftIO.Exception as E import UnliftIO.STM@@ -74,13 +77,16 @@ { rcvQ :: TBQueue (ATransmission 'Client), subQ :: TBQueue (ATransmission 'Agent), msgQ :: TBQueue SMPServerTransmission,- smpClients :: TVar (Map SMPServer SMPClientVar),- subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),- subscrConns :: TVar (Map ConnId SMPServer),- connMsgsQueued :: TVar (Map ConnId Bool),- smpQueueMsgQueues :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId)),- smpQueueMsgDeliveries :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (Async ())),+ smpServers :: TVar (NonEmpty SMPServer),+ smpClients :: TMap SMPServer SMPClientVar,+ subscrSrvrs :: TMap SMPServer (TMap ConnId RcvQueue),+ pendingSubscrSrvrs :: TMap SMPServer (TMap ConnId RcvQueue),+ subscrConns :: TMap ConnId SMPServer,+ connMsgsQueued :: TMap ConnId Bool,+ smpQueueMsgQueues :: TMap (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId),+ smpQueueMsgDeliveries :: TMap (ConnId, SMPServer, SMP.SenderId) (Async ()), reconnections :: TVar [Async ()],+ asyncClients :: TVar [Async ()], clientId :: Int, agentEnv :: Env, smpSubscriber :: Async (),@@ -93,16 +99,19 @@ rcvQ <- newTBQueue qSize subQ <- newTBQueue qSize msgQ <- newTBQueue qSize- smpClients <- newTVar M.empty- subscrSrvrs <- newTVar M.empty- subscrConns <- newTVar M.empty- connMsgsQueued <- newTVar M.empty- smpQueueMsgQueues <- newTVar M.empty- smpQueueMsgDeliveries <- newTVar M.empty+ smpServers <- newTVar $ initialSMPServers (config agentEnv)+ smpClients <- TM.empty+ subscrSrvrs <- TM.empty+ pendingSubscrSrvrs <- TM.empty+ subscrConns <- TM.empty+ connMsgsQueued <- TM.empty+ smpQueueMsgQueues <- TM.empty+ smpQueueMsgDeliveries <- TM.empty reconnections <- newTVar []+ asyncClients <- newTVar [] clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1) lock <- newTMVar ()- return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}+ return AgentClient {rcvQ, subQ, msgQ, smpServers, smpClients, subscrSrvrs, pendingSubscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, asyncClients, clientId, agentEnv, smpSubscriber = undefined, lock} -- | Agent monad with MonadReader Env and MonadError AgentErrorType type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)@@ -124,29 +133,48 @@ atomically getClientVar >>= either newSMPClient waitForSMPClient where getClientVar :: STM (Either SMPClientVar SMPClientVar)- getClientVar = maybe (Left <$> newClientVar) (pure . Right) . M.lookup srv =<< readTVar smpClients+ getClientVar = maybe (Left <$> newClientVar) (pure . Right) =<< TM.lookup srv smpClients newClientVar :: STM SMPClientVar newClientVar = do smpVar <- newEmptyTMVar- modifyTVar smpClients $ M.insert srv smpVar+ TM.insert srv smpVar smpClients pure smpVar waitForSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient- waitForSMPClient = liftIOEither . atomically . readTMVar+ waitForSMPClient smpVar = do+ SMPClientConfig {tcpTimeout} <- asks $ smpCfg . config+ smpClient_ <- liftIO $ tcpTimeout `timeout` atomically (readTMVar smpVar)+ liftEither $ case smpClient_ of+ Just (Right smpClient) -> Right smpClient+ Just (Left e) -> Left e+ Nothing -> Left $ BROKER TIMEOUT newSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient- newSMPClient smpVar =- tryError connectClient >>= \r -> case r of- Right smp -> do- logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv- atomically $ putTMVar smpVar r- pure smp- Left e -> do- atomically $ do- putTMVar smpVar r- modifyTVar smpClients $ M.delete srv- throwError e+ newSMPClient smpVar = tryConnectClient pure tryConnectAsync+ where+ tryConnectClient :: (SMPClient -> m a) -> m () -> m a+ tryConnectClient successAction retryAction =+ tryError connectClient >>= \r -> case r of+ Right smp -> do+ logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv+ atomically $ putTMVar smpVar r+ successAction smp+ Left e -> do+ if e == BROKER NETWORK || e == BROKER TIMEOUT+ then retryAction+ else atomically $ do+ putTMVar smpVar (Left e)+ TM.delete srv smpClients+ throwError e+ tryConnectAsync :: m ()+ tryConnectAsync = do+ a <- async connectAsync+ atomically $ modifyTVar' (asyncClients c) (a :)+ connectAsync :: m ()+ connectAsync = do+ ri <- asks $ reconnectInterval . config+ withRetryInterval ri $ \loop -> void $ tryConnectClient (const reconnectClient) loop connectClient :: m SMPClient connectClient = do@@ -160,45 +188,63 @@ clientDisconnected :: UnliftIO m -> IO () clientDisconnected u = do- removeClientSubs >>= (`forM_` serverDown u)+ removeClientAndSubs >>= (`forM_` serverDown u) logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv - removeClientSubs :: IO (Maybe (Map ConnId RcvQueue))- removeClientSubs = atomically $ do- modifyTVar smpClients $ M.delete srv- cs <- M.lookup srv <$> readTVar (subscrSrvrs c)- modifyTVar (subscrSrvrs c) $ M.delete srv- modifyTVar (subscrConns c) $ maybe id (deleteKeys . M.keysSet) cs- return cs+ removeClientAndSubs :: IO (Maybe (Map ConnId RcvQueue))+ removeClientAndSubs = atomically $ do+ TM.delete srv smpClients+ cVar_ <- TM.lookupDelete srv $ subscrSrvrs c+ forM cVar_ $ \cVar -> do+ cs <- readTVar cVar+ modifyTVar' (subscrConns c) (`M.withoutKeys` M.keysSet cs)+ addPendingSubs cVar cs+ pure cs where- deleteKeys :: Ord k => Set k -> Map k a -> Map k a- deleteKeys ks m = S.foldr' M.delete m ks+ addPendingSubs cVar cs = do+ let ps = pendingSubscrSrvrs c+ TM.lookup srv ps >>= \case+ Just v -> TM.union cs v+ _ -> TM.insert srv cVar ps serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO () serverDown u cs = unless (M.null cs) $ do mapM_ (notifySub DOWN) $ M.keysSet cs- a <- async . unliftIO u $ tryReconnectClient cs- atomically $ modifyTVar (reconnections c) (a :)+ unliftIO u reconnectServer - tryReconnectClient :: Map ConnId RcvQueue -> m ()- tryReconnectClient cs = do+ reconnectServer :: m ()+ reconnectServer = do+ a <- async tryReconnectClient+ atomically $ modifyTVar' (reconnections c) (a :)++ tryReconnectClient :: m ()+ tryReconnectClient = do ri <- asks $ reconnectInterval . config withRetryInterval ri $ \loop ->- reconnectClient cs `catchError` const loop+ reconnectClient `catchError` const loop - reconnectClient :: Map ConnId RcvQueue -> m ()- reconnectClient cs = do+ reconnectClient :: m ()+ reconnectClient = withAgentLock c . withSMP c srv $ \smp -> do- subs <- readTVarIO $ subscrConns c- forM_ (M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->- when (isNothing $ M.lookup connId subs) $ do- subscribeSMPQueue smp rcvPrivateKey rcvId- `catchError` \case- SMPServerError e -> liftIO $ notifySub (ERR $ SMP e) connId- e -> throwError e- addSubscription c rq connId- liftIO $ notifySub UP connId+ cs <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSubscrSrvrs c)+ forConcurrently_ (maybe [] M.toList cs) $ \sub@(connId, _) ->+ whenM (atomically $ isNothing <$> TM.lookup connId (subscrConns c)) $+ subscribe_ smp sub `catchError` handleError connId+ where+ subscribe_ :: SMPClient -> (ConnId, RcvQueue) -> ExceptT SMPClientError IO ()+ subscribe_ smp (connId, rq@RcvQueue {rcvPrivateKey, rcvId}) = do+ subscribeSMPQueue smp rcvPrivateKey rcvId+ addSubscription c rq connId+ liftIO $ notifySub UP connId + handleError :: ConnId -> SMPClientError -> ExceptT SMPClientError IO ()+ handleError connId = \case+ e@SMPResponseTimeout -> throwError e+ e@SMPNetworkError -> throwError e+ e -> do+ liftIO $ notifySub (ERR $ smpClientError e) connId+ atomically $ removePendingSubscription c srv connId+ notifySub :: ACommand 'Agent -> ConnId -> IO () notifySub cmd connId = atomically $ writeTBQueue (subQ c) ("", connId, cmd) @@ -206,6 +252,7 @@ closeAgentClient c = liftIO $ do closeSMPServerClients c cancelActions $ reconnections c+ cancelActions $ asyncClients c cancelActions $ smpQueueMsgDeliveries c closeSMPServerClients :: AgentClient -> IO ()@@ -294,32 +341,42 @@ subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m () subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do- withLogSMP c server rcvId "SUB" $ \smp ->- subscribeSMPQueue smp rcvPrivateKey rcvId- addSubscription c rq connId+ atomically $ addPendingSubscription c rq connId+ withLogSMP c server rcvId "SUB" $ \smp -> do+ liftIO (runExceptT $ subscribeSMPQueue smp rcvPrivateKey rcvId) >>= \case+ Left e -> do+ atomically . when (e /= SMPNetworkError && e /= SMPResponseTimeout) $+ removePendingSubscription c server connId+ throwError e+ Right _ -> addSubscription c rq connId addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m () addSubscription c rq@RcvQueue {server} connId = atomically $ do- modifyTVar (subscrConns c) $ M.insert connId server- modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server- where- addSub :: Maybe (Map ConnId RcvQueue) -> Map ConnId RcvQueue- addSub (Just cs) = M.insert connId rq cs- addSub _ = M.singleton connId rq+ TM.insert connId server $ subscrConns c+ addSubs_ (subscrSrvrs c) rq connId+ removePendingSubscription c server connId -removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()-removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do- cs <- readTVar subscrConns- writeTVar subscrConns $ M.delete connId cs- mapM_- (modifyTVar subscrSrvrs . M.alter (>>= delSub))- (M.lookup connId cs)- where- delSub :: Map ConnId RcvQueue -> Maybe (Map ConnId RcvQueue)- delSub cs =- let cs' = M.delete connId cs- in if M.null cs' then Nothing else Just cs'+addPendingSubscription :: AgentClient -> RcvQueue -> ConnId -> STM ()+addPendingSubscription = addSubs_ . pendingSubscrSrvrs +addSubs_ :: TMap SMPServer (TMap ConnId RcvQueue) -> RcvQueue -> ConnId -> STM ()+addSubs_ ss rq@RcvQueue {server} connId =+ TM.lookup server ss >>= \case+ Just m -> TM.insert connId rq m+ _ -> TM.singleton connId rq >>= \m -> TM.insert server m ss++removeSubscription :: AgentClient -> ConnId -> STM ()+removeSubscription c@AgentClient {subscrConns} connId = do+ server_ <- TM.lookupDelete connId subscrConns+ mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_++removePendingSubscription :: AgentClient -> SMPServer -> ConnId -> STM ()+removePendingSubscription = removeSubs_ . pendingSubscrSrvrs++removeSubs_ :: TMap SMPServer (TMap ConnId RcvQueue) -> SMPServer -> ConnId -> STM ()+removeSubs_ ss server connId =+ TM.lookup server ss >>= mapM_ (TM.delete connId)+ logServer :: AgentMonad m => ByteString -> AgentClient -> SMPServer -> QueueId -> ByteString -> m () logServer dir AgentClient {clientId} srv qId cmdStr = logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret qId, cmdStr]@@ -331,11 +388,13 @@ logSecret :: ByteString -> ByteString logSecret bs = encode $ B.take 3 bs --- TODO maybe package E2ERatchetParams into SMPConfirmation sendConfirmation :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()-sendConfirmation c SndQueue {server, sndId} encConfirmation =- withLogSMP_ c server sndId "SEND <CONF>" $ \smp ->- liftSMP $ sendSMPMessage smp Nothing sndId encConfirmation+sendConfirmation c sq@SndQueue {server, sndId, sndPublicKey = Just sndPublicKey, e2ePubKey = e2ePubKey@Just {}} agentConfirmation =+ withLogSMP_ c server sndId "SEND <CONF>" $ \smp -> do+ let clientMsg = SMP.ClientMessage (SMP.PHConfirmation sndPublicKey) agentConfirmation+ msg <- agentCbEncrypt sq e2ePubKey $ smpEncode clientMsg+ liftSMP $ sendSMPMessage smp Nothing sndId msg+sendConfirmation _ _ _ = throwError $ INTERNAL "sendConfirmation called without snd_queue public key(s) in the database" sendInvitation :: forall m. AgentMonad m => AgentClient -> Compatible SMPQueueInfo -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> m () sendInvitation c (Compatible SMPQueueInfo {smpServer, senderId, dhPublicKey}) connReq connInfo =@@ -370,7 +429,6 @@ withLogSMP c server rcvId "DEL" $ \smp -> deleteSMPQueue smp rcvPrivateKey rcvId --- TODO this is just wrong sendAgentMessage :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m () sendAgentMessage c sq@SndQueue {server, sndId, sndPrivateKey} agentMsg = withLogSMP_ c server sndId "SEND <MSG>" $ \smp -> do
src/Simplex/Messaging/Agent/Env/SQLite.hs view
@@ -29,12 +29,13 @@ data AgentConfig = AgentConfig { tcpPort :: ServiceName,- smpServers :: NonEmpty SMPServer,+ initialSMPServers :: NonEmpty SMPServer, cmdSignAlg :: C.SignAlg, connIdBytes :: Int, tbqSize :: Natural, dbFile :: FilePath, dbPoolSize :: Int,+ yesToMigrations :: Bool, smpCfg :: SMPClientConfig, reconnectInterval :: RetryInterval, helloTimeout :: NominalDiffTime,@@ -47,12 +48,13 @@ defaultAgentConfig = AgentConfig { tcpPort = "5224",- smpServers = undefined, -- TODO move it elsewhere?+ initialSMPServers = undefined, -- TODO move it elsewhere? cmdSignAlg = C.SignAlg C.SEd448, connIdBytes = 12,- tbqSize = 16,+ tbqSize = 64, dbFile = "smp-agent.db", dbPoolSize = 4,+ yesToMigrations = False, smpCfg = smpDefaultConfig, reconnectInterval = RetryInterval@@ -60,7 +62,7 @@ increaseAfter = 10 * second, maxInterval = 10 * second },- helloTimeout = 7 * nominalDay,+ helloTimeout = 2 * nominalDay, -- CA certificate private key is not needed for initialization -- ! we do not generate these caCertificateFile = "/etc/opt/simplex-agent/ca.crt",@@ -79,9 +81,9 @@ } newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env-newSMPAgentEnv cfg = do+newSMPAgentEnv config@AgentConfig {dbFile, dbPoolSize, yesToMigrations} = do idsDrg <- newTVarIO =<< drgNew- store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app+ store <- liftIO $ createSQLiteStore dbFile dbPoolSize Migrations.app yesToMigrations clientCounter <- newTVarIO 0 randomServer <- newTVarIO =<< liftIO newStdGen- return Env {config = cfg, store, idsDrg, clientCounter, randomServer}+ return Env {config, store, idsDrg, clientCounter, randomServer}
src/Simplex/Messaging/Agent/Protocol.hs view
@@ -46,9 +46,9 @@ SMPConfirmation (..), AgentMsgEnvelope (..), AgentMessage (..),+ AgentMessageType (..), APrivHeader (..), AMessage (..),- AMsgType (..), SMPServer (..), SrvLoc (..), SMPQueueUri (..),@@ -83,19 +83,13 @@ -- * Encode/decode serializeCommand,- serializeMsgIntegrity, connMode, connMode',- serializeAgentError,- serializeSmpErrorType, commandP, connModeT,- msgIntegrityP,- agentErrorTypeP,- smpErrorTypeP, serializeQueueStatus, queueStatusT,- aMessageType,+ agentMessageType, -- * TCP transport functions tPut,@@ -108,6 +102,7 @@ import Control.Applicative (optional, (<|>)) import Control.Monad.IO.Class import Data.Aeson (FromJSON (..), ToJSON (..))+import qualified Data.Aeson as J import Data.Attoparsec.ByteString.Char8 (Parser) import qualified Data.Attoparsec.ByteString.Char8 as A import Data.ByteString.Base64@@ -131,7 +126,7 @@ import Simplex.Messaging.Crypto.Ratchet (E2ERatchetParams, E2ERatchetParamsUri) import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String-import Simplex.Messaging.Parsers (base64P, parse, parseRead, parseRead1, parseRead2, tsISO8601P)+import Simplex.Messaging.Parsers import Simplex.Messaging.Protocol ( ErrorType, MsgBody,@@ -305,7 +300,7 @@ } | AgentInvitation -- the connInfo in contactInvite is only encrypted with per-queue E2E, not with double ratchet, { agentVersion :: Version,- connReq :: (ConnectionRequestUri 'CMInvitation),+ connReq :: ConnectionRequestUri 'CMInvitation, connInfo :: ByteString -- this message is only encrypted with per-queue E2E, not with double ratchet, } deriving (Show)@@ -348,6 +343,31 @@ 'M' -> AgentMessage <$> smpP <*> smpP _ -> fail "bad AgentMessage" +data AgentMessageType = AM_CONN_INFO | AM_HELLO_ | AM_REPLY_ | AM_A_MSG_+ deriving (Eq, Show)++instance Encoding AgentMessageType where+ smpEncode = \case+ AM_CONN_INFO -> "C"+ AM_HELLO_ -> "H"+ AM_REPLY_ -> "R"+ AM_A_MSG_ -> "M"+ smpP =+ A.anyChar >>= \case+ 'C' -> pure AM_CONN_INFO+ 'H' -> pure AM_HELLO_+ 'R' -> pure AM_REPLY_+ 'M' -> pure AM_A_MSG_+ _ -> fail "bad AgentMessageType"++agentMessageType :: AgentMessage -> AgentMessageType+agentMessageType = \case+ AgentConnInfo _ -> AM_CONN_INFO+ AgentMessage _ aMsg -> case aMsg of+ HELLO -> AM_HELLO_+ REPLY _ -> AM_REPLY_+ A_MSG _ -> AM_A_MSG_+ data APrivHeader = APrivHeader { -- | sequential ID assigned by the sending agent sndMsgId :: AgentMsgId,@@ -376,12 +396,6 @@ 'M' -> pure A_MSG_ _ -> fail "bad AMsgType" -aMessageType :: AMessage -> AMsgType-aMessageType = \case- HELLO -> HELLO_- REPLY _ -> REPLY_- A_MSG _ -> A_MSG_- -- | Messages sent between SMP agents once SMP queue is secured. -- -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md#messages-between-smp-agents@@ -618,29 +632,70 @@ type AgentMsgId = Int64 -- | Result of received message integrity validation.-data MsgIntegrity = MsgOk | MsgError MsgErrorType- deriving (Eq, Show)+data MsgIntegrity = MsgOk | MsgError {errorInfo :: MsgErrorType}+ deriving (Eq, Show, Generic) +instance StrEncoding MsgIntegrity where+ strP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> strP)+ strEncode = \case+ MsgOk -> "OK"+ MsgError e -> "ERR" <> strEncode e++instance ToJSON MsgIntegrity where+ toJSON = J.genericToJSON $ sumTypeJSON fstToLower+ toEncoding = J.genericToEncoding $ sumTypeJSON fstToLower++instance FromJSON MsgIntegrity where+ parseJSON = J.genericParseJSON $ sumTypeJSON fstToLower+ -- | Error of message integrity validation.-data MsgErrorType = MsgSkipped AgentMsgId AgentMsgId | MsgBadId AgentMsgId | MsgBadHash | MsgDuplicate- deriving (Eq, Show)+data MsgErrorType+ = MsgSkipped {fromMsgId :: AgentMsgId, toMsgId :: AgentMsgId}+ | MsgBadId {msgId :: AgentMsgId}+ | MsgBadHash+ | MsgDuplicate+ deriving (Eq, Show, Generic) +instance StrEncoding MsgErrorType where+ strP =+ "ID " *> (MsgBadId <$> A.decimal)+ <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)+ <|> "HASH" $> MsgBadHash+ <|> "DUPLICATE" $> MsgDuplicate+ strEncode = \case+ MsgSkipped fromMsgId toMsgId ->+ B.unwords ["NO_ID", bshow fromMsgId, bshow toMsgId]+ MsgBadId aMsgId -> "ID " <> bshow aMsgId+ MsgBadHash -> "HASH"+ MsgDuplicate -> "DUPLICATE"++instance ToJSON MsgErrorType where+ toJSON = J.genericToJSON $ sumTypeJSON fstToLower+ toEncoding = J.genericToEncoding $ sumTypeJSON fstToLower++instance FromJSON MsgErrorType where+ parseJSON = J.genericParseJSON $ sumTypeJSON fstToLower+ -- | Error type used in errors sent to agent clients. data AgentErrorType = -- | command or response error- CMD CommandErrorType+ CMD {cmdErr :: CommandErrorType} | -- | connection errors- CONN ConnectionErrorType+ CONN {connErr :: ConnectionErrorType} | -- | SMP protocol errors forwarded to agent clients- SMP ErrorType+ SMP {smpErr :: ErrorType} | -- | SMP server errors- BROKER BrokerErrorType+ BROKER {brokerErr :: BrokerErrorType} | -- | errors of other agents- AGENT SMPAgentError+ AGENT {agentErr :: SMPAgentError} | -- | agent implementation or dependency errors- INTERNAL String+ INTERNAL {internalErr :: String} deriving (Eq, Generic, Read, Show, Exception) +instance ToJSON AgentErrorType where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id+ -- | SMP agent protocol command or response error. data CommandErrorType = -- | command is prohibited in this context@@ -655,6 +710,10 @@ LARGE deriving (Eq, Generic, Read, Show, Exception) +instance ToJSON CommandErrorType where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id+ -- | Connection error. data ConnectionErrorType = -- | connection is not in the database@@ -665,24 +724,32 @@ SIMPLEX | -- | connection not accepted on join HELLO after timeout NOT_ACCEPTED- | -- | connection not available on reply HELLO after timeout+ | -- | connection not available on reply confirmation/HELLO after timeout NOT_AVAILABLE deriving (Eq, Generic, Read, Show, Exception) +instance ToJSON ConnectionErrorType where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id+ -- | SMP server errors. data BrokerErrorType = -- | invalid server response (failed to parse)- RESPONSE ErrorType+ RESPONSE {smpErr :: ErrorType} | -- | unexpected response UNEXPECTED | -- | network error NETWORK | -- | handshake or other transport error- TRANSPORT TransportError+ TRANSPORT {transportErr :: TransportError} | -- | command response timeout TIMEOUT deriving (Eq, Generic, Read, Show, Exception) +instance ToJSON BrokerErrorType where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id+ -- | Errors of another SMP agent. -- TODO encode/decode without A prefix data SMPAgentError@@ -696,6 +763,30 @@ A_ENCRYPTION deriving (Eq, Generic, Read, Show, Exception) +instance ToJSON SMPAgentError where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id++instance StrEncoding AgentErrorType where+ strP =+ "CMD " *> (CMD <$> parseRead1)+ <|> "CONN " *> (CONN <$> parseRead1)+ <|> "SMP " *> (SMP <$> strP)+ <|> "BROKER RESPONSE " *> (BROKER . RESPONSE <$> strP)+ <|> "BROKER TRANSPORT " *> (BROKER . TRANSPORT <$> transportErrorP)+ <|> "BROKER " *> (BROKER <$> parseRead1)+ <|> "AGENT " *> (AGENT <$> parseRead1)+ <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)+ strEncode = \case+ CMD e -> "CMD " <> bshow e+ CONN e -> "CONN " <> bshow e+ SMP e -> "SMP " <> strEncode e+ BROKER (RESPONSE e) -> "BROKER RESPONSE " <> strEncode e+ BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e+ BROKER e -> "BROKER " <> bshow e+ AGENT e -> "AGENT " <> bshow e+ INTERNAL e -> "INTERNAL " <> bshow e+ instance Arbitrary AgentErrorType where arbitrary = genericArbitraryU instance Arbitrary CommandErrorType where arbitrary = genericArbitraryU@@ -746,27 +837,17 @@ sendCmd = ACmd SClient . SEND <$> A.takeByteString msgIdResp = ACmd SAgent . MID <$> A.decimal sentResp = ACmd SAgent . SENT <$> A.decimal- msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> agentErrorTypeP+ msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP message = ACmd SAgent .: MSG <$> msgMetaP <* A.space <*> A.takeByteString ackCmd = ACmd SClient . ACK <$> A.decimal msgMetaP = do- integrity <- msgIntegrityP+ integrity <- strP recipient <- " R=" *> partyMeta A.decimal broker <- " B=" *> partyMeta base64P sndMsgId <- " S=" *> A.decimal pure MsgMeta {integrity, recipient, broker, sndMsgId} partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P- agentError = ACmd SAgent . ERR <$> agentErrorTypeP---- | Message integrity validation result parser.-msgIntegrityP :: Parser MsgIntegrity-msgIntegrityP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> msgErrorType)- where- msgErrorType =- "ID " *> (MsgBadId <$> A.decimal)- <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)- <|> "HASH" $> MsgBadHash- <|> "DUPLICATE" $> MsgDuplicate+ agentError = ACmd SAgent . ERR <$> strP parseCommand :: ByteString -> Either AgentErrorType ACmd parseCommand = parse commandP $ CMD SYNTAX@@ -790,13 +871,13 @@ SEND msgBody -> "SEND " <> serializeBinary msgBody MID mId -> "MID " <> bshow mId SENT mId -> "SENT " <> bshow mId- MERR mId e -> B.unwords ["MERR", bshow mId, serializeAgentError e]+ MERR mId e -> B.unwords ["MERR", bshow mId, strEncode e] MSG msgMeta msgBody -> B.unwords ["MSG", serializeMsgMeta msgMeta, serializeBinary msgBody] ACK mId -> "ACK " <> bshow mId OFF -> "OFF" DEL -> "DEL" CON -> "CON"- ERR e -> "ERR " <> serializeAgentError e+ ERR e -> "ERR " <> strEncode e OK -> "OK" where showTs :: UTCTime -> ByteString@@ -804,48 +885,11 @@ serializeMsgMeta :: MsgMeta -> ByteString serializeMsgMeta MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} = B.unwords- [ serializeMsgIntegrity integrity,+ [ strEncode integrity, "R=" <> bshow rmId <> "," <> showTs rTs, "B=" <> encode bmId <> "," <> showTs bTs, "S=" <> bshow sndMsgId ]---- | Serialize message integrity validation result.-serializeMsgIntegrity :: MsgIntegrity -> ByteString-serializeMsgIntegrity = \case- MsgOk -> "OK"- MsgError e ->- "ERR " <> case e of- MsgSkipped fromMsgId toMsgId ->- B.unwords ["NO_ID", bshow fromMsgId, bshow toMsgId]- MsgBadId aMsgId -> "ID " <> bshow aMsgId- MsgBadHash -> "HASH"- MsgDuplicate -> "DUPLICATE"---- | SMP agent protocol error parser.-agentErrorTypeP :: Parser AgentErrorType-agentErrorTypeP =- "SMP " *> (SMP <$> smpErrorTypeP)- <|> "BROKER RESPONSE " *> (BROKER . RESPONSE <$> smpErrorTypeP)- <|> "BROKER TRANSPORT " *> (BROKER . TRANSPORT <$> transportErrorP)- <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)- <|> parseRead2---- | Serialize SMP agent protocol error.-serializeAgentError :: AgentErrorType -> ByteString-serializeAgentError = \case- SMP e -> "SMP " <> serializeSmpErrorType e- BROKER (RESPONSE e) -> "BROKER RESPONSE " <> serializeSmpErrorType e- BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e- e -> bshow e---- | SMP error parser.-smpErrorTypeP :: Parser ErrorType-smpErrorTypeP = "CMD " *> (SMP.CMD <$> parseRead1) <|> parseRead1---- | Serialize SMP error.-serializeSmpErrorType :: ErrorType -> ByteString-serializeSmpErrorType = bshow serializeBinary :: ByteString -> ByteString serializeBinary body = bshow (B.length body) <> "\n" <> body
+ src/Simplex/Messaging/Agent/Server.hs view
@@ -0,0 +1,81 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Simplex.Messaging.Agent.Server+ ( -- * SMP agent over TCP+ runSMPAgent,+ runSMPAgentBlocking,+ )+where++import Control.Logger.Simple (logInfo)+import Control.Monad.Except+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Control.Monad.Reader+import Crypto.Random (MonadRandom)+import Data.ByteString.Char8 (ByteString)+import qualified Data.ByteString.Char8 as B+import Data.Text.Encoding (decodeUtf8)+import Simplex.Messaging.Agent+import Simplex.Messaging.Agent.Env.SQLite+import Simplex.Messaging.Agent.Protocol+import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)+import Simplex.Messaging.Transport.Server (loadTLSServerParams, runTransportServer)+import Simplex.Messaging.Util (bshow)+import UnliftIO.Async (race_)+import qualified UnliftIO.Exception as E+import UnliftIO.STM++-- | Runs an SMP agent as a TCP service using passed configuration.+--+-- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs+runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> m ()+runSMPAgent t cfg = do+ started <- newEmptyTMVarIO+ runSMPAgentBlocking t started cfg++-- | Runs an SMP agent as a TCP service using passed configuration with signalling.+--+-- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)+-- and when it is disconnected from the TCP socket once the server thread is killed (False).+runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()+runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do+ runReaderT (smpAgent t) =<< newSMPAgentEnv cfg+ where+ smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()+ smpAgent _ = do+ -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation+ tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile+ runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do+ liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion+ c <- getAgentClient+ logConnection c True+ race_ (connectClient h c) (runAgentClient c)+ `E.finally` disconnectAgentClient c++connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()+connectClient h c = race_ (send h c) (receive h c)++receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()+receive h c@AgentClient {rcvQ, subQ} = forever $ do+ (corrId, connId, cmdOrErr) <- tGet SClient h+ case cmdOrErr of+ Right cmd -> write rcvQ (corrId, connId, cmd)+ Left e -> write subQ (corrId, connId, ERR e)+ where+ write :: TBQueue (ATransmission p) -> ATransmission p -> m ()+ write q t = do+ logClient c "-->" t+ atomically $ writeTBQueue q t++send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()+send h c@AgentClient {subQ} = forever $ do+ t <- atomically $ readTBQueue subQ+ tPut h t+ logClient c "<--" t++logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()+logClient AgentClient {clientId} dir (corrId, connId, cmd) = do+ logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]
src/Simplex/Messaging/Agent/Store.hs view
@@ -62,7 +62,7 @@ createRcvMsg :: s -> ConnId -> RcvMsgData -> m () updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash) createSndMsg :: s -> ConnId -> SndMsgData -> m ()- getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))+ getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AgentMessageType, MsgBody, InternalTs)) getPendingMsgs :: s -> ConnId -> m [InternalId] checkRcvMsg :: s -> ConnId -> InternalId -> m () deleteMsg :: s -> ConnId -> InternalId -> m ()@@ -102,8 +102,11 @@ { server :: SMPServer, -- | sender queue ID sndId :: SMP.SenderId,- -- | key used by the sender to sign transmissions+ -- | key pair used by the sender to sign transmissions+ sndPublicKey :: Maybe C.APublicVerifyKey, sndPrivateKey :: SndPrivateSignKey,+ -- | DH public key used to negotiate per-queue e2e encryption+ e2ePubKey :: Maybe C.PublicKeyX25519, -- | shared DH secret agreed for simple per-queue e2e encryption e2eDhSecret :: C.DhSecretX25519, -- | queue status@@ -221,7 +224,7 @@ data RcvMsgData = RcvMsgData { msgMeta :: MsgMeta,- msgType :: AMsgType,+ msgType :: AgentMessageType, msgBody :: MsgBody, internalRcvId :: InternalRcvId, internalHash :: MsgHash,@@ -232,7 +235,7 @@ { internalId :: InternalId, internalSndId :: InternalSndId, internalTs :: InternalTs,- msgType :: AMsgType,+ msgType :: AgentMessageType, msgBody :: MsgBody, internalHash :: MsgHash, prevMsgHash :: MsgHash
src/Simplex/Messaging/Agent/Store/SQLite.hs view
@@ -76,13 +76,13 @@ dbNew :: Bool } -createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore-createSQLiteStore dbFilePath poolSize migrations = do+createSQLiteStore :: FilePath -> Int -> [Migration] -> Bool -> IO SQLiteStore+createSQLiteStore dbFilePath poolSize migrations yesToMigrations = do let dbDir = takeDirectory dbFilePath createDirectoryIfMissing False dbDir st <- connectSQLiteStore dbFilePath poolSize checkThreadsafe st- migrateSchema st migrations+ migrateSchema st migrations yesToMigrations pure st checkThreadsafe :: SQLiteStore -> IO ()@@ -94,15 +94,16 @@ Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found" _ -> return () -migrateSchema :: SQLiteStore -> [Migration] -> IO ()-migrateSchema st migrations = withConnection st $ \db -> do+migrateSchema :: SQLiteStore -> [Migration] -> Bool -> IO ()+migrateSchema st migrations yesToMigrations = withConnection st $ \db -> do Migrations.initialize db Migrations.get db migrations >>= \case Left e -> confirmOrExit $ "Database error: " <> e Right [] -> pure () Right ms -> do unless (dbNew st) $ do- confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."+ unless yesToMigrations $+ confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded." let f = dbFilePath st copyFile f (f <> ".bak") Migrations.run db ms@@ -126,9 +127,25 @@ connectDB :: FilePath -> IO DB.Connection connectDB path = do dbConn <- DB.open path- DB.execute_ dbConn "PRAGMA foreign_keys = ON; PRAGMA journal_mode = WAL;"+ DB.execute_ dbConn "PRAGMA foreign_keys = ON;"+ -- DB.execute_ dbConn "PRAGMA trusted_schema = OFF;"+ DB.execute_ dbConn "PRAGMA secure_delete = ON;"+ DB.execute_ dbConn "PRAGMA auto_vacuum = FULL;"+ -- _printPragmas dbConn path pure dbConn +_printPragmas :: DB.Connection -> FilePath -> IO ()+_printPragmas db path = do+ foreign_keys <- DB.query_ db "PRAGMA foreign_keys;" :: IO [[Int]]+ print $ path <> " foreign_keys: " <> show foreign_keys+ -- when run via sqlite-simple query for trusted_schema seems to return empty list+ trusted_schema <- DB.query_ db "PRAGMA trusted_schema;" :: IO [[Int]]+ print $ path <> " trusted_schema: " <> show trusted_schema+ secure_delete <- DB.query_ db "PRAGMA secure_delete;" :: IO [[Int]]+ print $ path <> " secure_delete: " <> show secure_delete+ auto_vacuum <- DB.query_ db "PRAGMA auto_vacuum;" :: IO [[Int]]+ print $ path <> " auto_vacuum: " <> show auto_vacuum+ checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a) checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err) @@ -440,7 +457,7 @@ insertSndMsgDetails_ db connId sndMsgData updateHashSnd_ db connId sndMsgData - getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))+ getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AgentMessageType, MsgBody, InternalTs)) getPendingMsgData st connId msgId = liftIOEither . withTransaction st $ \db -> runExceptT $ do rq_ <- liftIO $ getRcvQueueByConnId_ db connId@@ -565,13 +582,13 @@ instance FromField InternalId where fromField x = InternalId <$> fromField x -instance ToField AMsgType where toField = toField . smpEncode+instance ToField AgentMessageType where toField = toField . smpEncode -instance FromField AMsgType where fromField = blobFieldParser smpP+instance FromField AgentMessageType where fromField = blobFieldParser smpP -instance ToField MsgIntegrity where toField = toField . serializeMsgIntegrity+instance ToField MsgIntegrity where toField = toField . strEncode -instance FromField MsgIntegrity where fromField = blobFieldParser msgIntegrityP+instance FromField MsgIntegrity where fromField = blobFieldParser strP instance ToField SMPQueueUri where toField = toField . strEncode @@ -655,46 +672,25 @@ insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO () insertRcvQueue_ dbConn connId RcvQueue {..} = do- DB.executeNamed+ DB.execute dbConn [sql| INSERT INTO rcv_queues- ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status)- VALUES- (:host,:port,:rcv_id,:conn_id,:rcv_private_key,:rcv_dh_secret,:e2e_priv_key,:e2e_dh_secret,:snd_id,:status);+ ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status) VALUES (?,?,?,?,?,?,?,?,?,?); |]- [ ":host" := host server,- ":port" := port server,- ":rcv_id" := rcvId,- ":conn_id" := connId,- ":rcv_private_key" := rcvPrivateKey,- ":rcv_dh_secret" := rcvDhSecret,- ":e2e_priv_key" := e2ePrivKey,- ":e2e_dh_secret" := e2eDhSecret,- ":snd_id" := sndId,- ":status" := status- ]+ (host server, port server, rcvId, connId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) -- * createSndConn helpers insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO () insertSndQueue_ dbConn connId SndQueue {..} = do- DB.executeNamed+ DB.execute dbConn [sql| INSERT INTO snd_queues- ( host, port, snd_id, conn_id, snd_private_key, e2e_dh_secret, status)- VALUES- (:host,:port,:snd_id,:conn_id,:snd_private_key,:e2e_dh_secret,:status);+ (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status) VALUES (?,?,?,?,?, ?,?, ?,?); |]- [ ":host" := host server,- ":port" := port server,- ":snd_id" := sndId,- ":conn_id" := connId,- ":snd_private_key" := sndPrivateKey,- ":e2e_dh_secret" := e2eDhSecret,- ":status" := status- ]+ (host server, port server, sndId, connId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status) -- * getConn helpers @@ -745,16 +741,16 @@ <$> DB.query dbConn [sql|- SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.e2e_dh_secret, q.status+ SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status FROM snd_queues q INNER JOIN servers s ON q.host = s.host AND q.port = s.port WHERE q.conn_id = ?; |] (Only connId) where- sndQueue [(keyHash, host, port, sndId, sndPrivateKey, e2eDhSecret, status)] =+ sndQueue [(keyHash, host, port, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status)] = let server = SMPServer host port keyHash- in Just SndQueue {server, sndId, sndPrivateKey, e2eDhSecret, status}+ in Just SndQueue {server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status} sndQueue _ = Nothing -- * updateRcvIds helpers
src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs view
@@ -25,13 +25,15 @@ import Database.SQLite.Simple.QQ (sql) import qualified Database.SQLite3 as SQLite3 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys data Migration = Migration {name :: String, up :: Text} deriving (Show) schemaMigrations :: [(String, Query)] schemaMigrations =- [ ("20220101_initial", m20220101_initial)+ [ ("20220101_initial", m20220101_initial),+ ("20220301_snd_queue_keys", m20220301_snd_queue_keys) ] -- | The list of migrations in ascending order by date
+ src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220301_snd_queue_keys.hs view
@@ -0,0 +1,13 @@+{-# LANGUAGE QuasiQuotes #-}++module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys where++import Database.SQLite.Simple (Query)+import Database.SQLite.Simple.QQ (sql)++m20220301_snd_queue_keys :: Query+m20220301_snd_queue_keys =+ [sql|+ALTER TABLE snd_queues ADD COLUMN snd_public_key BLOB;+ALTER TABLE snd_queues ADD COLUMN e2e_pub_key BLOB;+|]
src/Simplex/Messaging/Client.hs view
@@ -23,7 +23,7 @@ -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md module Simplex.Messaging.Client ( -- * Connect (disconnect) client to (from) SMP server- SMPClient,+ SMPClient (sessionId), getSMPClient, closeSMPClient, @@ -56,14 +56,16 @@ import Control.Monad.Trans.Except import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B-import Data.Map.Strict (Map)-import qualified Data.Map.Strict as M import Data.Maybe (fromMaybe) import Network.Socket (ServiceName) import Numeric.Natural import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Protocol-import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)+import Simplex.Messaging.TMap (TMap)+import qualified Simplex.Messaging.TMap as TM+import Simplex.Messaging.Transport+import Simplex.Messaging.Transport.Client (runTransportClient)+import Simplex.Messaging.Transport.KeepAlive import Simplex.Messaging.Transport.WebSockets (WS) import Simplex.Messaging.Util (bshow, liftError, raceAny_) import System.Timeout (timeout)@@ -77,18 +79,18 @@ data SMPClient = SMPClient { action :: Async (), connected :: TVar Bool,- sessionId :: ByteString,+ sessionId :: SessionId, smpServer :: SMPServer, tcpTimeout :: Int, clientCorrId :: TVar Natural,- sentCommands :: TVar (Map CorrId Request),+ sentCommands :: TMap CorrId Request, sndQ :: TBQueue SentRawTransmission, rcvQ :: TBQueue (SignedTransmission BrokerMsg), msgQ :: TBQueue SMPServerTransmission } -- | Type synonym for transmission from some SPM server queue.-type SMPServerTransmission = (SMPServer, RecipientId, BrokerMsg)+type SMPServerTransmission = (SMPServer, SessionId, RecipientId, BrokerMsg) -- | SMP client configuration. data SMPClientConfig = SMPClientConfig@@ -98,6 +100,8 @@ defaultTransport :: (ServiceName, ATransport), -- | timeout of TCP commands (microseconds) tcpTimeout :: Int,+ -- | TCP keep-alive options, Nothing to skip enabling keep-alive+ tcpKeepAlive :: Maybe KeepAliveOpts, -- | period for SMP ping commands (microseconds) smpPing :: Int }@@ -106,10 +110,11 @@ smpDefaultConfig :: SMPClientConfig smpDefaultConfig = SMPClientConfig- { qSize = 16,+ { qSize = 64, defaultTransport = ("5223", transport @TLS),- tcpTimeout = 4_000_000,- smpPing = 30_000_000+ tcpTimeout = 5_000_000,+ tcpKeepAlive = Just defaultKeepAliveOpts,+ smpPing = 600_000_000 -- 10min } data Request = Request@@ -125,14 +130,14 @@ -- A single queue can be used for multiple 'SMPClient' instances, -- as 'SMPServerTransmission' includes server information. getSMPClient :: SMPServer -> SMPClientConfig -> TBQueue SMPServerTransmission -> IO () -> IO (Either SMPClientError SMPClient)-getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ disconnected =+getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, tcpKeepAlive, smpPing} msgQ disconnected = atomically mkSMPClient >>= runClient useTransport where mkSMPClient :: STM SMPClient mkSMPClient = do connected <- newTVar False clientCorrId <- newTVar 0- sentCommands <- newTVar M.empty+ sentCommands <- TM.empty sndQ <- newTBQueue qSize rcvQ <- newTBQueue qSize return@@ -154,7 +159,7 @@ thVar <- newEmptyTMVarIO action <- async $- runTransportClient (host smpServer) port' (keyHash smpServer) (client t c thVar)+ runTransportClient (host smpServer) port' (keyHash smpServer) tcpKeepAlive (client t c thVar) `finally` atomically (putTMVar thVar $ Left SMPNetworkError) th_ <- tcpTimeout `timeout` atomically (takeTMVar thVar) pure $ case th_ of@@ -192,16 +197,15 @@ runExceptT $ sendSMPCommand c Nothing "" PING process :: SMPClient -> IO ()- process SMPClient {rcvQ, sentCommands} = forever $ do+ process SMPClient {sessionId, rcvQ, sentCommands} = forever $ do (_, _, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ if B.null $ bs corrId then sendMsg qId respOrErr else do- cs <- readTVarIO sentCommands- case M.lookup corrId cs of+ atomically (TM.lookup corrId sentCommands) >>= \case Nothing -> sendMsg qId respOrErr Just Request {queueId, responseVar} -> atomically $ do- modifyTVar sentCommands $ M.delete corrId+ TM.delete corrId sentCommands putTMVar responseVar $ if queueId == qId then case respOrErr of@@ -209,12 +213,12 @@ Right (ERR e) -> Left $ SMPServerError e Right r -> Right r else Left SMPUnexpectedResponse-- sendMsg :: QueueId -> Either ErrorType BrokerMsg -> IO ()- sendMsg qId = \case- Right cmd -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)- -- TODO send everything else to errQ and log in agent- _ -> return ()+ where+ sendMsg :: QueueId -> Either ErrorType BrokerMsg -> IO ()+ sendMsg qId = \case+ Right cmd -> atomically $ writeTBQueue msgQ (smpServer, sessionId, qId, cmd)+ -- TODO send everything else to errQ and log in agent+ _ -> return () -- | Disconnects SMP client from the server and terminates client threads. closeSMPClient :: SMPClient -> IO ()@@ -264,11 +268,11 @@ -- -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#subscribe-to-queue subscribeSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT SMPClientError IO ()-subscribeSMPQueue c@SMPClient {smpServer, msgQ} rpKey rId =+subscribeSMPQueue c@SMPClient {smpServer, sessionId, msgQ} rpKey rId = sendSMPCommand c (Just rpKey) rId SUB >>= \case OK -> return () cmd@MSG {} ->- lift . atomically $ writeTBQueue msgQ (smpServer, rId, cmd)+ lift . atomically $ writeTBQueue msgQ (smpServer, sessionId, rId, cmd) _ -> throwE SMPUnexpectedResponse -- | Subscribe to the SMP queue notifications.@@ -305,11 +309,11 @@ -- -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#acknowledge-message-delivery ackSMPMessage :: SMPClient -> RcvPrivateSignKey -> QueueId -> ExceptT SMPClientError IO ()-ackSMPMessage c@SMPClient {smpServer, msgQ} rpKey rId =+ackSMPMessage c@SMPClient {smpServer, sessionId, msgQ} rpKey rId = sendSMPCommand c (Just rpKey) rId ACK >>= \case OK -> return () cmd@MSG {} ->- lift . atomically $ writeTBQueue msgQ (smpServer, rId, cmd)+ lift . atomically $ writeTBQueue msgQ (smpServer, sessionId, rId, cmd) _ -> throwE SMPUnexpectedResponse -- | Irreversibly suspend SMP queue.@@ -363,6 +367,6 @@ send :: CorrId -> SentRawTransmission -> STM (TMVar Response) send corrId t = do r <- newEmptyTMVar- modifyTVar sentCommands . M.insert corrId $ Request qId r+ TM.insert corrId (Request qId r) sentCommands writeTBQueue sndQ t return r
src/Simplex/Messaging/Parsers.hs view
@@ -1,16 +1,18 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} module Simplex.Messaging.Parsers where import Control.Monad.Trans.Except+import qualified Data.Aeson as J import Data.Attoparsec.ByteString.Char8 (Parser) import qualified Data.Attoparsec.ByteString.Char8 as A import Data.Bifunctor (first) import Data.ByteString.Base64 import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B-import Data.Char (isAlphaNum)+import Data.Char (isAlphaNum, toLower) import Data.Time.Clock (UTCTime) import Data.Time.ISO8601 (parseISO8601) import Data.Typeable (Typeable)@@ -78,3 +80,46 @@ Right k -> Ok k Left e -> returnError ConversionFailed f ("couldn't parse field: " ++ e) f -> returnError ConversionFailed f "expecting SQLBlob column type"++fstToLower :: String -> String+fstToLower "" = ""+fstToLower (h : t) = toLower h : t++dropPrefix :: String -> String -> String+dropPrefix pfx s =+ let (p, rest) = splitAt (length pfx) s+ in fstToLower $ if p == pfx then rest else s++enumJSON :: (String -> String) -> J.Options+enumJSON tagModifier =+ J.defaultOptions+ { J.constructorTagModifier = tagModifier,+ J.allNullaryToStringTag = True+ }++sumTypeJSON :: (String -> String) -> J.Options+#if defined(darwin_HOST_OS) && defined(swiftJSON)+sumTypeJSON = singleFieldJSON+#else+sumTypeJSON = taggedObjectJSON+#endif++taggedObjectJSON :: (String -> String) -> J.Options+taggedObjectJSON tagModifier =+ J.defaultOptions+ { J.sumEncoding = J.TaggedObject "type" "data",+ J.constructorTagModifier = tagModifier,+ J.allNullaryToStringTag = False,+ J.nullaryToObject = True,+ J.omitNothingFields = True+ }++singleFieldJSON :: (String -> String) -> J.Options+singleFieldJSON tagModifier =+ J.defaultOptions+ { J.sumEncoding = J.ObjectWithSingleField,+ J.constructorTagModifier = tagModifier,+ J.allNullaryToStringTag = False,+ J.nullaryToObject = True,+ J.omitNothingFields = True+ }
src/Simplex/Messaging/Protocol.hs view
@@ -90,6 +90,8 @@ import Control.Applicative (optional, (<|>)) import Control.Monad.Except+import Data.Aeson (ToJSON (..))+import qualified Data.Aeson as J import Data.Attoparsec.ByteString.Char8 (Parser) import qualified Data.Attoparsec.ByteString.Char8 as A import Data.ByteString.Char8 (ByteString)@@ -106,8 +108,8 @@ import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers-import Simplex.Messaging.Transport (THandle (..), Transport, TransportError (..), tGetBlock, tPutBlock)-import Simplex.Messaging.Util ((<$?>))+import Simplex.Messaging.Transport (SessionId, THandle (..), Transport, TransportError (..), tGetBlock, tPutBlock)+import Simplex.Messaging.Util (bshow, (<$?>)) import Simplex.Messaging.Version import Test.QuickCheck (Arbitrary (..)) @@ -170,14 +172,14 @@ data RawTransmission = RawTransmission { signature :: ByteString, signed :: ByteString,- sessId :: ByteString,+ sessId :: SessionId, corrId :: ByteString, queueId :: ByteString, command :: ByteString } -- | unparsed sent SMP transmission with signature, without session ID.-type SignedRawTransmission = (Maybe C.ASignature, ByteString, ByteString, ByteString)+type SignedRawTransmission = (Maybe C.ASignature, SessionId, ByteString, ByteString) -- | unparsed sent SMP transmission with signature. type SentRawTransmission = (Maybe C.ASignature, ByteString)@@ -397,6 +399,10 @@ SrvLoc host port <- strP pure SMPServer {host, port, keyHash} +instance ToJSON SMPServer where+ toJSON = strToJSON+ toEncoding = strToJEncoding+ data SrvLoc = SrvLoc HostName ServiceName deriving (Eq, Ord, Show) @@ -413,6 +419,15 @@ instance IsString CorrId where fromString = CorrId . fromString +instance StrEncoding CorrId where+ strEncode (CorrId cId) = strEncode cId+ strDecode s = CorrId <$> strDecode s+ strP = CorrId <$> strP++instance ToJSON CorrId where+ toJSON = strToJSON+ toEncoding = strToJEncoding+ -- | Queue IDs and keys data QueueIdsKeys = QIK { rcvId :: RecipientId,@@ -462,7 +477,7 @@ | -- | incorrect SMP session ID (TLS Finished message / tls-unique binding RFC5929) SESSION | -- | SMP command is unknown or has invalid syntax- CMD CommandError+ CMD {cmdErr :: CommandError} | -- | command authorization error - bad signature or non-existing SMP queue AUTH | -- | SMP queue capacity is exceeded on the server@@ -477,6 +492,16 @@ DUPLICATE_ -- TODO remove, not part of SMP protocol deriving (Eq, Generic, Read, Show) +instance ToJSON ErrorType where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id++instance StrEncoding ErrorType where+ strEncode = \case+ CMD e -> "CMD " <> bshow e+ e -> bshow e+ strP = "CMD " *> (CMD <$> parseRead1) <|> parseRead1+ -- | SMP command error type. data CommandError = -- | unknown command@@ -490,6 +515,10 @@ | -- | transmission has no required queue ID NO_QUEUE deriving (Eq, Generic, Read, Show)++instance ToJSON CommandError where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id instance Arbitrary ErrorType where arbitrary = genericArbitraryU
src/Simplex/Messaging/Server.hs view
@@ -25,7 +25,6 @@ -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md module Simplex.Messaging.Server (runSMPServer, runSMPServerBlocking) where -import Control.Concurrent.STM (stateTVar) import Control.Monad import Control.Monad.Except import Control.Monad.IO.Unlift@@ -34,9 +33,10 @@ import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import Data.Functor (($>))+import Data.Int (Int64) import qualified Data.Map.Strict as M import Data.Maybe (isNothing)-import Data.Time.Clock.System (getSystemTime)+import Data.Time.Clock.System (SystemTime (..), getSystemTime) import Data.Type.Equality import Network.Socket (ServiceName) import qualified Simplex.Messaging.Crypto as C@@ -47,7 +47,10 @@ import Simplex.Messaging.Server.QueueStore import Simplex.Messaging.Server.QueueStore.STM (QueueStore) import Simplex.Messaging.Server.StoreLog+import Simplex.Messaging.TMap (TMap)+import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport+import Simplex.Messaging.Transport.Server import Simplex.Messaging.Util import UnliftIO.Concurrent import UnliftIO.Exception@@ -67,34 +70,32 @@ -- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True) -- and when it is disconnected from the TCP socket once the server thread is killed (False). runSMPServerBlocking :: (MonadRandom m, MonadUnliftIO m) => TMVar Bool -> ServerConfig -> m ()-runSMPServerBlocking started cfg@ServerConfig {transports} = do- env <- newEnv cfg- runReaderT smpServer env- where- smpServer :: (MonadUnliftIO m', MonadReader Env m') => m' ()- smpServer = do- s <- asks server- raceAny_- ( serverThread s subscribedQ subscribers subscriptions cancelSub :- serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :- map runServer transports- )- `finally` withLog closeStoreLog+runSMPServerBlocking started cfg = newEnv cfg >>= runReaderT (smpServer started) - runServer :: (MonadUnliftIO m', MonadReader Env m') => (ServiceName, ATransport) -> m' ()+smpServer :: forall m. (MonadUnliftIO m, MonadReader Env m) => TMVar Bool -> m ()+smpServer started = do+ s <- asks server+ cfg@ServerConfig {transports} <- asks config+ raceAny_+ ( serverThread s subscribedQ subscribers subscriptions cancelSub :+ serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :+ map runServer transports <> expireMessagesThread_ cfg+ )+ `finally` withLog closeStoreLog+ where+ runServer :: (ServiceName, ATransport) -> m () runServer (tcpPort, ATransport t) = do serverParams <- asks tlsServerParams runTransportServer started tcpPort serverParams (runClient t) serverThread ::- forall m' s.- MonadUnliftIO m' =>+ forall s. Server -> (Server -> TBQueue (QueueId, Client)) ->- (Server -> TVar (M.Map QueueId Client)) ->- (Client -> TVar (M.Map QueueId s)) ->- (s -> m' ()) ->- m' ()+ (Server -> TMap QueueId Client) ->+ (Client -> TMap QueueId s) ->+ (s -> m ()) ->+ m () serverThread s subQ subs clientSubs unsub = forever $ do atomically updateSubscribers >>= fmap join . mapM endPreviousSubscriptions@@ -109,15 +110,33 @@ else do yes <- readTVar $ connected c' pure $ if yes then Just (qId, c') else Nothing- stateTVar (subs s) (\cs -> (M.lookup qId cs, M.insert qId clnt cs))+ TM.lookupInsert qId clnt (subs s) >>= fmap join . mapM clientToBeNotified- endPreviousSubscriptions :: (QueueId, Client) -> m' (Maybe s)+ endPreviousSubscriptions :: (QueueId, Client) -> m (Maybe s) endPreviousSubscriptions (qId, c) = do void . forkIO . atomically $ writeTBQueue (sndQ c) (CorrId "", qId, END)- atomically . stateTVar (clientSubs c) $ \ss -> (M.lookup qId ss, M.delete qId ss)+ atomically $ TM.lookupDelete qId (clientSubs c) - runClient :: (Transport c, MonadUnliftIO m, MonadReader Env m) => TProxy c -> c -> m ()+ expireMessagesThread_ :: ServerConfig -> [m ()]+ expireMessagesThread_ ServerConfig {messageTTL, expireMessagesInterval} =+ case (messageTTL, expireMessagesInterval) of+ (Just ttl, Just int) -> [expireMessages ttl int]+ _ -> []++ expireMessages :: Int64 -> Int -> m ()+ expireMessages ttl interval = do+ ms <- asks msgStore+ quota <- asks $ msgQueueQuota . config+ forever $ do+ threadDelay interval+ old <- subtract ttl . systemSeconds <$> liftIO getSystemTime+ rIds <- M.keysSet <$> readTVarIO ms+ forM_ rIds $ \rId ->+ atomically (getMsgQueue ms rId quota)+ >>= atomically . (`deleteExpiredMsgs` old)++ runClient :: Transport c => TProxy c -> c -> m () runClient _ h = do kh <- asks serverIdentity liftIO (runExceptT $ serverHandshake h kh) >>= \case@@ -138,7 +157,7 @@ subs <- readTVarIO subscriptions mapM_ cancelSub subs cs <- asks $ subscribers . server- atomically . mapM_ (modifyTVar cs . M.update deleteCurrentClient) $ M.keys subs+ atomically . mapM_ (\rId -> TM.update deleteCurrentClient rId cs) $ M.keys subs where deleteCurrentClient :: Client -> Maybe Client deleteCurrentClient c'@@ -229,7 +248,11 @@ Cmd SNotifier NSUB -> subscribeNotifications Cmd SRecipient command -> case command of- NEW rKey dhKey -> createQueue st rKey dhKey+ NEW rKey dhKey ->+ ifM+ (asks $ allowNewQueues . config)+ (createQueue st rKey dhKey)+ (pure (corrId, queueId, ERR AUTH)) SUB -> subscribeQueue queueId ACK -> acknowledgeMsg KEY sKey -> secureQueue_ st sKey@@ -308,21 +331,19 @@ getSubscription :: RecipientId -> STM Sub getSubscription rId = do- subs <- readTVar subscriptions- case M.lookup rId subs of+ TM.lookup rId subscriptions >>= \case Just s -> tryTakeTMVar (delivered s) $> s Nothing -> do writeTBQueue subscribedQ (rId, clnt) s <- newSubscription- writeTVar subscriptions $ M.insert rId s subs+ TM.insert rId s subscriptions return s subscribeNotifications :: m (Transmission BrokerMsg) subscribeNotifications = atomically $ do- subs <- readTVar ntfSubscriptions- when (isNothing $ M.lookup queueId subs) $ do+ whenM (isNothing <$> TM.lookup queueId ntfSubscriptions) $ do writeTBQueue ntfSubscribedQ (queueId, clnt)- writeTVar ntfSubscriptions $ M.insert queueId () subs+ TM.insert queueId () ntfSubscriptions pure ok acknowledgeMsg :: m (Transmission BrokerMsg)@@ -333,7 +354,7 @@ _ -> return $ err NO_MSG withSub :: RecipientId -> (Sub -> STM a) -> STM (Maybe a)- withSub rId f = readTVar subscriptions >>= mapM f . M.lookup rId+ withSub rId f = mapM f =<< TM.lookup rId subscriptions sendMessage :: QueueStore -> MsgBody -> m (Transmission BrokerMsg) sendMessage st msgBody@@ -350,9 +371,11 @@ Left _ -> pure $ err LARGE_MSG Right msg -> do ms <- asks msgStore- quota <- asks $ msgQueueQuota . config+ ServerConfig {messageTTL, msgQueueQuota} <- asks config+ old <- forM messageTTL $ \ttl -> subtract ttl . systemSeconds <$> liftIO getSystemTime atomically $ do- q <- getMsgQueue ms (recipientId qr) quota+ q <- getMsgQueue ms (recipientId qr) msgQueueQuota+ mapM_ (deleteExpiredMsgs q) old ifM (isFull q) (pure $ err QUOTA) $ do trySendNotification writeMsg q msg@@ -368,7 +391,7 @@ trySendNotification :: STM () trySendNotification = forM_ (notifier qr) $ \(nId, _) ->- mapM_ (writeNtf nId) . M.lookup nId =<< readTVar notifiers+ mapM_ (writeNtf nId) =<< TM.lookup nId notifiers writeNtf :: NotifierId -> Client -> STM () writeNtf nId Client {sndQ = q} =@@ -402,7 +425,7 @@ void setDelivered setSub :: (Sub -> Sub) -> STM ()- setSub f = modifyTVar subscriptions $ M.adjust f rId+ setSub f = TM.adjust f rId subscriptions setDelivered :: STM (Maybe Bool) setDelivered = withSub rId $ \s -> tryPutTMVar (delivered s) ()
src/Simplex/Messaging/Server/Env/STM.hs view
@@ -9,6 +9,7 @@ import Control.Monad.IO.Unlift import Crypto.Random import Data.ByteString.Char8 (ByteString)+import Data.Int (Int64) import Data.Map.Strict (Map) import qualified Data.Map.Strict as M import Data.X509.Validation (Fingerprint (..))@@ -21,7 +22,10 @@ import Simplex.Messaging.Server.QueueStore (QueueRec (..)) import Simplex.Messaging.Server.QueueStore.STM import Simplex.Messaging.Server.StoreLog-import Simplex.Messaging.Transport (ATransport, loadFingerprint, loadTLSServerParams)+import Simplex.Messaging.TMap (TMap)+import qualified Simplex.Messaging.TMap as TM+import Simplex.Messaging.Transport (ATransport)+import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams) import System.IO (IOMode (..)) import UnliftIO.STM @@ -33,6 +37,12 @@ queueIdBytes :: Int, msgIdBytes :: Int, storeLog :: Maybe (StoreLog 'ReadMode),+ -- | set to False to prohibit creating new queues+ allowNewQueues :: Bool,+ -- | time after which the messages can be removed from the queues, seconds+ messageTTL :: Maybe Int64,+ -- | interval to periodically remove expired messages (when no messages are sent to the queue), microseconds+ expireMessagesInterval :: Maybe Int, -- CA certificate private key is not needed for initialization caCertificateFile :: FilePath, privateKeyFile :: FilePath,@@ -52,14 +62,14 @@ data Server = Server { subscribedQ :: TBQueue (RecipientId, Client),- subscribers :: TVar (Map RecipientId Client),+ subscribers :: TMap RecipientId Client, ntfSubscribedQ :: TBQueue (NotifierId, Client),- notifiers :: TVar (Map NotifierId Client)+ notifiers :: TMap NotifierId Client } data Client = Client- { subscriptions :: TVar (Map RecipientId Sub),- ntfSubscriptions :: TVar (Map NotifierId ()),+ { subscriptions :: TMap RecipientId Sub,+ ntfSubscriptions :: TMap NotifierId (), rcvQ :: TBQueue (Transmission Cmd), sndQ :: TBQueue (Transmission BrokerMsg), sessionId :: ByteString,@@ -76,15 +86,15 @@ newServer :: Natural -> STM Server newServer qSize = do subscribedQ <- newTBQueue qSize- subscribers <- newTVar M.empty+ subscribers <- TM.empty ntfSubscribedQ <- newTBQueue qSize- notifiers <- newTVar M.empty+ notifiers <- TM.empty return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers} newClient :: Natural -> ByteString -> STM Client newClient qSize sessionId = do- subscriptions <- newTVar M.empty- ntfSubscriptions <- newTVar M.empty+ subscriptions <- TM.empty+ ntfSubscriptions <- TM.empty rcvQ <- newTBQueue qSize sndQ <- newTBQueue qSize connected <- newTVar True@@ -108,15 +118,12 @@ return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog = s', tlsServerParams} where restoreQueues :: QueueStore -> StoreLog 'ReadMode -> m (StoreLog 'WriteMode)- restoreQueues queueStore s = do- (queues, s') <- liftIO $ readWriteStoreLog s- atomically $- modifyTVar queueStore $ \d ->- d- { queues,- senders = M.foldr' addSender M.empty queues,- notifiers = M.foldr' addNotifier M.empty queues- }+ restoreQueues QueueStore {queues, senders, notifiers} s = do+ (qs, s') <- liftIO $ readWriteStoreLog s+ atomically $ do+ writeTVar queues =<< mapM newTVar qs+ writeTVar senders $ M.foldr' addSender M.empty qs+ writeTVar notifiers $ M.foldr' addNotifier M.empty qs pure s' addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId addSender q = M.insert (senderId q) (recipientId q)
src/Simplex/Messaging/Server/MsgStore.hs view
@@ -2,6 +2,7 @@ module Simplex.Messaging.Server.MsgStore where +import Data.Int (Int64) import Data.Time.Clock.System (SystemTime) import Numeric.Natural import Simplex.Messaging.Protocol (MsgBody, MsgId, RecipientId)@@ -22,3 +23,4 @@ tryPeekMsg :: q -> m (Maybe Message) -- non blocking peekMsg :: q -> m Message -- blocking tryDelPeekMsg :: q -> m (Maybe Message) -- atomic delete (== read) last and peek next message, if available+ deleteExpiredMsgs :: q -> Int64 -> m ()
src/Simplex/Messaging/Server/MsgStore/STM.hs view
@@ -3,39 +3,38 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE InstanceSigs #-} {-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-} module Simplex.Messaging.Server.MsgStore.STM where -import Data.Map.Strict (Map)-import qualified Data.Map.Strict as M+import Control.Monad (when)+import Data.Int (Int64)+import Data.Time.Clock.System (SystemTime (systemSeconds)) import Numeric.Natural import Simplex.Messaging.Protocol (RecipientId) import Simplex.Messaging.Server.MsgStore+import Simplex.Messaging.TMap (TMap)+import qualified Simplex.Messaging.TMap as TM import UnliftIO.STM newtype MsgQueue = MsgQueue {msgQueue :: TBQueue Message} -newtype MsgStoreData = MsgStoreData {messages :: Map RecipientId MsgQueue}--type STMMsgStore = TVar MsgStoreData+type STMMsgStore = TMap RecipientId MsgQueue newMsgStore :: STM STMMsgStore-newMsgStore = newTVar $ MsgStoreData M.empty+newMsgStore = TM.empty instance MonadMsgStore STMMsgStore MsgQueue STM where getMsgQueue :: STMMsgStore -> RecipientId -> Natural -> STM MsgQueue- getMsgQueue store rId quota = do- m <- messages <$> readTVar store- maybe (newQ m) return $ M.lookup rId m+ getMsgQueue st rId quota = maybe newQ pure =<< TM.lookup rId st where- newQ m' = do+ newQ = do q <- MsgQueue <$> newTBQueue quota- writeTVar store . MsgStoreData $ M.insert rId q m'+ TM.insert rId q st return q delMsgQueue :: STMMsgStore -> RecipientId -> STM ()- delMsgQueue store rId =- modifyTVar store $ MsgStoreData . M.delete rId . messages+ delMsgQueue st rId = TM.delete rId st instance MonadMsgQueue MsgQueue STM where isFull :: MsgQueue -> STM Bool@@ -53,3 +52,11 @@ -- atomic delete (== read) last and peek next message if available tryDelPeekMsg :: MsgQueue -> STM (Maybe Message) tryDelPeekMsg (MsgQueue q) = tryReadTBQueue q >> tryPeekTBQueue q++ deleteExpiredMsgs :: MsgQueue -> Int64 -> STM ()+ deleteExpiredMsgs (MsgQueue q) old = loop+ where+ loop = tryPeekTBQueue q >>= mapM_ delOldMsg+ delOldMsg Message {ts} =+ when (systemSeconds ts < old) $+ tryReadTBQueue q >> loop
src/Simplex/Messaging/Server/QueueStore.hs view
@@ -15,6 +15,7 @@ notifier :: Maybe (NotifierId, NtfPublicVerifyKey), status :: QueueStatus }+ deriving (Eq, Show) data QueueStatus = QueueActive | QueueOff deriving (Eq, Show)
src/Simplex/Messaging/Server/QueueStore/STM.hs view
@@ -3,6 +3,7 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE InstanceSigs #-} {-# LANGUAGE KindSignatures #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE RankNTypes #-}@@ -11,107 +12,83 @@ module Simplex.Messaging.Server.QueueStore.STM where -import Data.Map.Strict (Map)-import qualified Data.Map.Strict as M+import Control.Monad+import Data.Functor (($>)) import Simplex.Messaging.Protocol import Simplex.Messaging.Server.QueueStore+import Simplex.Messaging.TMap (TMap)+import qualified Simplex.Messaging.TMap as TM+import Simplex.Messaging.Util (ifM) import UnliftIO.STM -data QueueStoreData = QueueStoreData- { queues :: Map RecipientId QueueRec,- senders :: Map SenderId RecipientId,- notifiers :: Map NotifierId RecipientId+data QueueStore = QueueStore+ { queues :: TMap RecipientId (TVar QueueRec),+ senders :: TMap SenderId RecipientId,+ notifiers :: TMap NotifierId RecipientId } -type QueueStore = TVar QueueStoreData- newQueueStore :: STM QueueStore-newQueueStore = newTVar QueueStoreData {queues = M.empty, senders = M.empty, notifiers = M.empty}+newQueueStore = do+ queues <- TM.empty+ senders <- TM.empty+ notifiers <- TM.empty+ pure QueueStore {queues, senders, notifiers} instance MonadQueueStore QueueStore STM where addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ())- addQueue store qRec@QueueRec {recipientId = rId, senderId = sId} = do- cs@QueueStoreData {queues, senders} <- readTVar store- if M.member rId queues || M.member sId senders- then return $ Left DUPLICATE_- else do- writeTVar store $- cs- { queues = M.insert rId qRec queues,- senders = M.insert sId rId senders- }- return $ Right ()+ addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = do+ ifM hasId (pure $ Left DUPLICATE_) $ do+ qVar <- newTVar q+ TM.insert rId qVar queues+ TM.insert sId rId senders+ pure $ Right ()+ where+ hasId = (||) <$> TM.member rId queues <*> TM.member sId senders getQueue :: QueueStore -> SParty p -> QueueId -> STM (Either ErrorType QueueRec)- getQueue st party qId = do- cs <- readTVar st- pure $ case party of- SRecipient -> getRcpQueue cs qId- SSender -> getPartyQueue cs senders- SNotifier -> getPartyQueue cs notifiers+ getQueue QueueStore {queues, senders, notifiers} party qId =+ toResult <$> (mapM readTVar =<< getVar) where- getPartyQueue ::- QueueStoreData ->- (QueueStoreData -> Map QueueId RecipientId) ->- Either ErrorType QueueRec- getPartyQueue cs recipientIds =- case M.lookup qId $ recipientIds cs of- Just rId -> getRcpQueue cs rId- Nothing -> Left AUTH+ getVar = case party of+ SRecipient -> TM.lookup qId queues+ SSender -> TM.lookup qId senders >>= get+ SNotifier -> TM.lookup qId notifiers >>= get+ get = fmap join . mapM (`TM.lookup` queues) secureQueue :: QueueStore -> RecipientId -> SndPublicVerifyKey -> STM (Either ErrorType QueueRec)- secureQueue store rId sKey =- updateQueues store rId $ \cs c ->- case senderKey c of- Just _ -> (Left AUTH, cs)- _ -> (Right c, cs {queues = M.insert rId c {senderKey = Just sKey} (queues cs)})+ secureQueue QueueStore {queues} rId sKey =+ withQueue rId queues $ \qVar ->+ readTVar qVar >>= \q -> case senderKey q of+ Just _ -> pure Nothing+ _ -> writeTVar qVar q {senderKey = Just sKey} $> Just q addQueueNotifier :: QueueStore -> RecipientId -> NotifierId -> NtfPublicVerifyKey -> STM (Either ErrorType QueueRec)- addQueueNotifier store rId nId nKey = do- cs@QueueStoreData {queues, notifiers} <- readTVar store- if M.member nId notifiers- then pure $ Left DUPLICATE_- else case M.lookup rId queues of- Nothing -> pure $ Left AUTH- Just q -> case notifier q of- Just _ -> pure $ Left AUTH+ addQueueNotifier QueueStore {queues, notifiers} rId nId nKey = do+ ifM (TM.member nId notifiers) (pure $ Left DUPLICATE_) $+ withQueue rId queues $ \qVar ->+ readTVar qVar >>= \q -> case notifier q of+ Just _ -> pure Nothing _ -> do- writeTVar store $- cs- { queues = M.insert rId q {notifier = Just (nId, nKey)} queues,- notifiers = M.insert nId rId notifiers- }- pure $ Right q+ writeTVar qVar q {notifier = Just (nId, nKey)}+ TM.insert nId rId notifiers+ pure $ Just q suspendQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())- suspendQueue store rId =- updateQueues store rId $ \cs c ->- (Right (), cs {queues = M.insert rId c {status = QueueOff} (queues cs)})+ suspendQueue QueueStore {queues} rId =+ withQueue rId queues $ \qVar -> modifyTVar' qVar (\q -> q {status = QueueOff}) $> Just () deleteQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())- deleteQueue store rId =- updateQueues store rId $ \cs c ->- ( Right (),- cs- { queues = M.delete rId (queues cs),- senders = M.delete (senderId c) (senders cs)- }- )+ deleteQueue QueueStore {queues, senders, notifiers} rId = do+ TM.lookupDelete rId queues >>= \case+ Just qVar ->+ readTVar qVar >>= \q -> do+ TM.delete (senderId q) senders+ forM_ (notifier q) $ \(nId, _) -> TM.delete nId notifiers+ pure $ Right ()+ _ -> pure $ Left AUTH -updateQueues ::- QueueStore ->- RecipientId ->- (QueueStoreData -> QueueRec -> (Either ErrorType a, QueueStoreData)) ->- STM (Either ErrorType a)-updateQueues store rId update = do- cs <- readTVar store- let conn = getRcpQueue cs rId- either (return . Left) (_update cs) conn- where- _update cs c = do- let (res, cs') = update cs c- writeTVar store cs'- return res+toResult :: Maybe a -> Either ErrorType a+toResult = maybe (Left AUTH) Right -getRcpQueue :: QueueStoreData -> RecipientId -> Either ErrorType QueueRec-getRcpQueue cs rId = maybe (Left AUTH) Right . M.lookup rId $ queues cs+withQueue :: RecipientId -> TMap RecipientId (TVar QueueRec) -> (TVar QueueRec -> STM (Maybe a)) -> STM (Either ErrorType a)+withQueue rId queues f = toResult <$> (TM.lookup rId queues >>= fmap join . mapM f)
+ src/Simplex/Messaging/TMap.hs view
@@ -0,0 +1,70 @@+module Simplex.Messaging.TMap+ ( TMap,+ empty,+ singleton,+ Simplex.Messaging.TMap.lookup,+ member,+ insert,+ delete,+ lookupInsert,+ lookupDelete,+ adjust,+ update,+ alter,+ union,+ )+where++import Control.Concurrent.STM+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as M++type TMap k a = TVar (Map k a)++empty :: STM (TMap k a)+empty = newTVar M.empty+{-# INLINE empty #-}++singleton :: k -> a -> STM (TMap k a)+singleton k v = newTVar $ M.singleton k v+{-# INLINE singleton #-}++lookup :: Ord k => k -> TMap k a -> STM (Maybe a)+lookup k m = M.lookup k <$> readTVar m+{-# INLINE lookup #-}++member :: Ord k => k -> TMap k a -> STM Bool+member k m = M.member k <$> readTVar m+{-# INLINE member #-}++insert :: Ord k => k -> a -> TMap k a -> STM ()+insert k v m = modifyTVar' m $ M.insert k v+{-# INLINE insert #-}++delete :: Ord k => k -> TMap k a -> STM ()+delete k m = modifyTVar' m $ M.delete k+{-# INLINE delete #-}++lookupInsert :: Ord k => k -> a -> TMap k a -> STM (Maybe a)+lookupInsert k v m = stateTVar m $ \mv -> (M.lookup k mv, M.insert k v mv)+{-# INLINE lookupInsert #-}++lookupDelete :: Ord k => k -> TMap k a -> STM (Maybe a)+lookupDelete k m = stateTVar m $ \mv -> (M.lookup k mv, M.delete k mv)+{-# INLINE lookupDelete #-}++adjust :: Ord k => (a -> a) -> k -> TMap k a -> STM ()+adjust f k m = modifyTVar' m $ M.adjust f k+{-# INLINE adjust #-}++update :: Ord k => (a -> Maybe a) -> k -> TMap k a -> STM ()+update f k m = modifyTVar' m $ M.update f k+{-# INLINE update #-}++alter :: Ord k => (Maybe a -> Maybe a) -> k -> TMap k a -> STM ()+alter f k m = modifyTVar' m $ M.alter f k+{-# INLINE alter #-}++union :: Ord k => Map k a -> TMap k a -> STM ()+union m' m = modifyTVar' m $ M.union m'+{-# INLINE union #-}
src/Simplex/Messaging/Transport.hs view
@@ -36,15 +36,12 @@ ATransport (..), TransportPeer (..), - -- * Transport over TLS- runTransportServer,- runTransportClient,- loadTLSServerParams,- loadFingerprint,- -- * TLS Transport TLS (..),+ SessionId,+ connectTLS, closeTLS,+ supportedParameters, withTlsUnique, -- * SMP transport@@ -64,9 +61,9 @@ import Control.Applicative ((<|>)) import Control.Monad.Except-import Control.Monad.IO.Unlift import Control.Monad.Trans.Except (throwE)-import qualified Crypto.Store.X509 as SX+import Data.Aeson (ToJSON)+import qualified Data.Aeson as J import Data.Attoparsec.ByteString.Char8 (Parser) import Data.Bifunctor (first) import Data.Bitraversable (bimapM)@@ -75,14 +72,7 @@ import qualified Data.ByteString.Lazy as BL import Data.Default (def) import Data.Functor (($>))-import Data.Set (Set)-import qualified Data.Set as S-import qualified Data.X509 as X-import qualified Data.X509.CertificateStore as XS-import Data.X509.Validation (Fingerprint (..))-import qualified Data.X509.Validation as XV import GHC.Generics (Generic)-import GHC.IO.Exception (IOErrorType (..)) import GHC.IO.Handle.Internals (ioe_EOF) import Generic.Random (genericArbitraryU) import Network.Socket@@ -90,14 +80,11 @@ import qualified Network.TLS.Extra as TE import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding-import Simplex.Messaging.Parsers (parse, parseRead1)+import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON) import Simplex.Messaging.Util (bshow) import Simplex.Messaging.Version-import System.Exit (exitFailure)-import System.IO.Error import Test.QuickCheck (Arbitrary (..))-import UnliftIO.Concurrent-import UnliftIO.Exception (Exception, IOException)+import UnliftIO.Exception (Exception) import qualified UnliftIO.Exception as E import UnliftIO.STM @@ -110,7 +97,7 @@ supportedSMPVersions = mkVersionRange 1 1 simplexMQVersion :: String-simplexMQVersion = "1.0.2"+simplexMQVersion = "1.1.0" -- * Transport connection class @@ -129,7 +116,7 @@ getClientConnection :: T.Context -> IO c -- | tls-unique channel binding per RFC5929- tlsUnique :: c -> ByteString+ tlsUnique :: c -> SessionId -- | Close connection closeConnection :: c -> IO ()@@ -154,104 +141,6 @@ data ATransport = forall c. Transport c => ATransport (TProxy c) --- * Transport over TLS---- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.------ All accepted connections are passed to the passed function.-runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> (c -> m ()) -> m ()-runTransportServer started port serverParams server = do- u <- askUnliftIO- liftIO $ do- clients <- newTVarIO S.empty- E.bracket- (startTCPServer started port)- (closeServer clients)- $ \sock -> forever $ do- (connSock, _) <- accept sock- tid <- forkIO $ connectClient u connSock `E.catch` \(_ :: E.SomeException) -> pure ()- atomically . modifyTVar clients $ S.insert tid- where- connectClient :: UnliftIO m -> Socket -> IO ()- connectClient u connSock =- E.bracket- (connectTLS serverParams connSock >>= getServerConnection)- closeConnection- (unliftIO u . server)- closeServer :: TVar (Set ThreadId) -> Socket -> IO ()- closeServer clients sock = do- readTVarIO clients >>= mapM_ killThread- close sock- void . atomically $ tryPutTMVar started False--startTCPServer :: TMVar Bool -> ServiceName -> IO Socket-startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted- where- resolve =- let hints = defaultHints {addrFlags = [AI_PASSIVE], addrSocketType = Stream}- in head <$> getAddrInfo (Just hints) Nothing (Just port)- open addr = do- sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)- setSocketOption sock ReuseAddr 1- withFdSocket sock setCloseOnExecIfNeeded- bind sock $ addrAddress addr- listen sock 1024- return sock- setStarted sock = atomically (tryPutTMVar started True) >> pure sock---- | Connect to passed TCP host:port and pass handle to the client.-runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> (c -> m a) -> m a-runTransportClient host port keyHash client = do- let clientParams = mkTLSClientParams host port keyHash- c <- liftIO $ startTCPClient host port clientParams- client c `E.finally` liftIO (closeConnection c)--startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> IO c-startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err- where- err :: IOException- err = mkIOError NoSuchThing "no address" Nothing Nothing-- resolve :: IO [AddrInfo]- resolve =- let hints = defaultHints {addrSocketType = Stream}- in getAddrInfo (Just hints) (Just host) (Just port)-- tryOpen :: IOException -> [AddrInfo] -> IO c- tryOpen e [] = E.throwIO e- tryOpen _ (addr : as) =- E.try (open addr) >>= either (`tryOpen` as) pure-- open :: AddrInfo -> IO c- open addr = do- sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)- connect sock $ addrAddress addr- ctx <- connectTLS clientParams sock- getClientConnection ctx--loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams-loadTLSServerParams caCertificateFile certificateFile privateKeyFile =- fromCredential <$> loadServerCredential- where- loadServerCredential :: IO T.Credential- loadServerCredential =- T.credentialLoadX509Chain certificateFile [caCertificateFile] privateKeyFile >>= \case- Right credential -> pure credential- Left _ -> putStrLn "invalid credential" >> exitFailure- fromCredential :: T.Credential -> T.ServerParams- fromCredential credential =- def- { T.serverWantClientCert = False,- T.serverShared = def {T.sharedCredentials = T.Credentials [credential]},- T.serverHooks = def,- T.serverSupported = supportedParameters- }--loadFingerprint :: FilePath -> IO Fingerprint-loadFingerprint certificateFile = do- (cert : _) <- SX.readSignedObject certificateFile- pure $ XV.getFingerprint (cert :: X.SignedExact X.Certificate) X.HashSHA256- -- * TLS Transport data TLS = TLS@@ -290,33 +179,6 @@ (T.bye ctx >> T.contextClose ctx) -- sometimes socket was closed before 'TLS.bye' `E.catch` (\(_ :: E.SomeException) -> pure ()) -- so we catch the 'Broken pipe' error here -mkTLSClientParams :: HostName -> ServiceName -> C.KeyHash -> T.ClientParams-mkTLSClientParams host port keyHash = do- let p = B.pack port- (T.defaultParamsClient host p)- { T.clientShared = def,- T.clientHooks = def {T.onServerCertificate = \_ _ _ -> validateCertificateChain keyHash host p},- T.clientSupported = supportedParameters- }--validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]-validateCertificateChain _ _ _ (X.CertificateChain []) = pure [XV.EmptyChain]-validateCertificateChain _ _ _ (X.CertificateChain [_]) = pure [XV.EmptyChain]-validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_, caCert]) =- if Fingerprint kh == XV.getFingerprint caCert X.HashSHA256- then x509validate- else pure [XV.UnknownCA]- where- x509validate :: IO [XV.FailedReason]- x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc- where- hooks = XV.defaultHooks- checks = XV.defaultChecks- certStore = XS.makeCertificateStore sc- cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)- serviceID = (host, port)-validateCertificateChain _ _ _ _ = pure [XV.AuthorityTooDeep]- supportedParameters :: T.Supported supportedParameters = def@@ -389,14 +251,17 @@ -- | The handle for SMP encrypted transport connection over Transport . data THandle c = THandle { connection :: c,- sessionId :: ByteString,+ sessionId :: SessionId, -- | agreed SMP server protocol version smpVersion :: Version } +-- | TLS-unique channel binding+type SessionId = ByteString+ data ServerHandshake = ServerHandshake { smpVersionRange :: VersionRange,- sessionId :: ByteString+ sessionId :: SessionId } data ClientHandshake = ClientHandshake@@ -428,9 +293,13 @@ | -- | incorrect session ID TEBadSession | -- | transport handshake error- TEHandshake HandshakeError+ TEHandshake {handshakeErr :: HandshakeError} deriving (Eq, Generic, Read, Show, Exception) +instance ToJSON TransportError where+ toJSON = J.genericToJSON . sumTypeJSON $ dropPrefix "TE"+ toEncoding = J.genericToEncoding . sumTypeJSON $ dropPrefix "TE"+ -- | Transport handshake error. data HandshakeError = -- | parsing error@@ -440,6 +309,10 @@ | -- | incorrect server identity IDENTITY deriving (Eq, Generic, Read, Show, Exception)++instance ToJSON HandshakeError where+ toJSON = J.genericToJSON $ sumTypeJSON id+ toEncoding = J.genericToEncoding $ sumTypeJSON id instance Arbitrary TransportError where arbitrary = genericArbitraryU
+ src/Simplex/Messaging/Transport/Client.hs view
@@ -0,0 +1,84 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Simplex.Messaging.Transport.Client+ ( runTransportClient,+ clientHandshake,+ )+where++import Control.Monad.Except+import Control.Monad.IO.Unlift+import Data.ByteString.Char8 (ByteString)+import qualified Data.ByteString.Char8 as B+import Data.Default (def)+import qualified Data.X509 as X+import qualified Data.X509.CertificateStore as XS+import Data.X509.Validation (Fingerprint (..))+import qualified Data.X509.Validation as XV+import GHC.IO.Exception (IOErrorType (..))+import Network.Socket+import qualified Network.TLS as T+import qualified Simplex.Messaging.Crypto as C+import Simplex.Messaging.Transport+import Simplex.Messaging.Transport.KeepAlive+import System.IO.Error+import UnliftIO.Exception (IOException)+import qualified UnliftIO.Exception as E++-- | Connect to passed TCP host:port and pass handle to the client.+runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> Maybe KeepAliveOpts -> (c -> m a) -> m a+runTransportClient host port keyHash keepAliveOpts client = do+ let clientParams = mkTLSClientParams host port keyHash+ c <- liftIO $ startTCPClient host port clientParams keepAliveOpts+ client c `E.finally` liftIO (closeConnection c)++startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> Maybe KeepAliveOpts -> IO c+startTCPClient host port clientParams keepAliveOpts = withSocketsDo $ resolve >>= tryOpen err+ where+ err :: IOException+ err = mkIOError NoSuchThing "no address" Nothing Nothing++ resolve :: IO [AddrInfo]+ resolve =+ let hints = defaultHints {addrSocketType = Stream}+ in getAddrInfo (Just hints) (Just host) (Just port)++ tryOpen :: IOException -> [AddrInfo] -> IO c+ tryOpen e [] = E.throwIO e+ tryOpen _ (addr : as) =+ E.try (open addr) >>= either (`tryOpen` as) pure++ open :: AddrInfo -> IO c+ open addr = do+ sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)+ connect sock $ addrAddress addr+ mapM_ (setSocketKeepAlive sock) keepAliveOpts+ ctx <- connectTLS clientParams sock+ getClientConnection ctx++mkTLSClientParams :: HostName -> ServiceName -> C.KeyHash -> T.ClientParams+mkTLSClientParams host port keyHash = do+ let p = B.pack port+ (T.defaultParamsClient host p)+ { T.clientShared = def,+ T.clientHooks = def {T.onServerCertificate = \_ _ _ -> validateCertificateChain keyHash host p},+ T.clientSupported = supportedParameters+ }++validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]+validateCertificateChain _ _ _ (X.CertificateChain []) = pure [XV.EmptyChain]+validateCertificateChain _ _ _ (X.CertificateChain [_]) = pure [XV.EmptyChain]+validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_, caCert]) =+ if Fingerprint kh == XV.getFingerprint caCert X.HashSHA256+ then x509validate+ else pure [XV.UnknownCA]+ where+ x509validate :: IO [XV.FailedReason]+ x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc+ where+ hooks = XV.defaultHooks+ checks = XV.defaultChecks+ certStore = XS.makeCertificateStore sc+ cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)+ serviceID = (host, port)+validateCertificateChain _ _ _ _ = pure [XV.AuthorityTooDeep]
+ src/Simplex/Messaging/Transport/KeepAlive.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE CApiFFI #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE NamedFieldPuns #-}++module Simplex.Messaging.Transport.KeepAlive where++import Foreign.C (CInt (..))+import Network.Socket++data KeepAliveOpts = KeepAliveOpts+ { keepIdle :: Int,+ keepIntvl :: Int,+ keepCnt :: Int+ }++defaultKeepAliveOpts :: KeepAliveOpts+defaultKeepAliveOpts =+ KeepAliveOpts+ { keepIdle = 30,+ keepIntvl = 15,+ keepCnt = 4+ }++_SOL_TCP :: CInt+_SOL_TCP = 6++#if defined(mingw32_HOST_OS)+-- Windows++-- The values are copied from windows::Win32::Networking::WinSock+-- https://microsoft.github.io/windows-docs-rs/doc/windows/Win32/Networking/WinSock/index.html++_TCP_KEEPIDLE :: CInt+_TCP_KEEPIDLE = 3++_TCP_KEEPINTVL :: CInt+_TCP_KEEPINTVL = 17++_TCP_KEEPCNT :: CInt+_TCP_KEEPCNT = 16++#else+-- Mac/Linux++#if defined(darwin_HOST_OS)+foreign import capi "netinet/tcp.h value TCP_KEEPALIVE" _TCP_KEEPIDLE :: CInt+#else+foreign import capi "netinet/tcp.h value TCP_KEEPIDLE" _TCP_KEEPIDLE :: CInt+#endif++foreign import capi "netinet/tcp.h value TCP_KEEPINTVL" _TCP_KEEPINTVL :: CInt++foreign import capi "netinet/tcp.h value TCP_KEEPCNT" _TCP_KEEPCNT :: CInt++#endif++setSocketKeepAlive :: Socket -> KeepAliveOpts -> IO ()+setSocketKeepAlive sock KeepAliveOpts {keepCnt, keepIdle, keepIntvl} = do+ setSocketOption sock KeepAlive 1+ setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPIDLE) keepIdle+ setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPINTVL) keepIntvl+ setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPCNT) keepCnt
+ src/Simplex/Messaging/Transport/Server.hs view
@@ -0,0 +1,107 @@+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Simplex.Messaging.Transport.Server+ ( runTransportServer,+ runTCPServer,+ loadTLSServerParams,+ loadFingerprint,+ serverHandshake,+ )+where++import Control.Monad.Except+import Control.Monad.IO.Unlift+import qualified Crypto.Store.X509 as SX+import Data.Default (def)+import Data.Set (Set)+import qualified Data.Set as S+import qualified Data.X509 as X+import Data.X509.Validation (Fingerprint (..))+import qualified Data.X509.Validation as XV+import Network.Socket+import qualified Network.TLS as T+import Simplex.Messaging.Transport+import System.Exit (exitFailure)+import UnliftIO.Concurrent+import qualified UnliftIO.Exception as E+import UnliftIO.STM++-- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.+--+-- All accepted connections are passed to the passed function.+runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> (c -> m ()) -> m ()+runTransportServer started port serverParams server = do+ u <- askUnliftIO+ liftIO $ do+ clients <- newTVarIO S.empty+ E.bracket+ (startTCPServer started port)+ (closeServer started clients)+ $ \sock -> forever $ do+ (connSock, _) <- accept sock+ tid <- forkIO $ connectClient u connSock `E.catch` \(_ :: E.SomeException) -> pure ()+ atomically . modifyTVar' clients $ S.insert tid+ where+ connectClient :: UnliftIO m -> Socket -> IO ()+ connectClient u connSock =+ E.bracket+ (connectTLS serverParams connSock >>= getServerConnection)+ closeConnection+ (unliftIO u . server)++runTCPServer :: TMVar Bool -> ServiceName -> (Socket -> IO ()) -> IO ()+runTCPServer started port server = do+ clients <- newTVarIO S.empty+ E.bracket+ (startTCPServer started port)+ (closeServer started clients)+ $ \sock -> forever $ do+ (connSock, _) <- accept sock+ tid <- forkIO $ server connSock `E.catch` \(_ :: E.SomeException) -> pure ()+ atomically . modifyTVar' clients $ S.insert tid++closeServer :: TMVar Bool -> TVar (Set ThreadId) -> Socket -> IO ()+closeServer started clients sock = do+ readTVarIO clients >>= mapM_ killThread+ close sock+ void . atomically $ tryPutTMVar started False++startTCPServer :: TMVar Bool -> ServiceName -> IO Socket+startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted+ where+ resolve =+ let hints = defaultHints {addrFlags = [AI_PASSIVE], addrSocketType = Stream}+ in head <$> getAddrInfo (Just hints) Nothing (Just port)+ open addr = do+ sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)+ setSocketOption sock ReuseAddr 1+ withFdSocket sock setCloseOnExecIfNeeded+ bind sock $ addrAddress addr+ listen sock 1024+ return sock+ setStarted sock = atomically (tryPutTMVar started True) >> pure sock++loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams+loadTLSServerParams caCertificateFile certificateFile privateKeyFile =+ fromCredential <$> loadServerCredential+ where+ loadServerCredential :: IO T.Credential+ loadServerCredential =+ T.credentialLoadX509Chain certificateFile [caCertificateFile] privateKeyFile >>= \case+ Right credential -> pure credential+ Left _ -> putStrLn "invalid credential" >> exitFailure+ fromCredential :: T.Credential -> T.ServerParams+ fromCredential credential =+ def+ { T.serverWantClientCert = False,+ T.serverShared = def {T.sharedCredentials = T.Credentials [credential]},+ T.serverHooks = def,+ T.serverSupported = supportedParameters+ }++loadFingerprint :: FilePath -> IO Fingerprint+loadFingerprint certificateFile = do+ (cert : _) <- SX.readSignedObject certificateFile+ pure $ XV.getFingerprint (cert :: X.SignedExact X.Certificate) X.HashSHA256
src/Simplex/Messaging/Util.hs view
@@ -58,6 +58,10 @@ ifM ba t f = ba >>= \b -> if b then t else f {-# INLINE ifM #-} +whenM :: Monad m => m Bool -> m () -> m ()+whenM b a = ifM b a $ pure ()+{-# INLINE whenM #-}+ unlessM :: Monad m => m Bool -> m () -> m () unlessM b = ifM b $ pure () {-# INLINE unlessM #-}
tests/AgentTests.hs view
@@ -62,9 +62,11 @@ smpAgentTest3_1_1 $ testSubscription t it "should send notifications to client when server disconnects" $ smpAgentServerTest $ testSubscrNotification t- describe "Message delivery" $ do+ describe "Message delivery and server reconnection" $ do it "should deliver messages after losing server connection and re-connecting" $ smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t+ it "should connect to the server when server goes up if it initially was down" $+ smpAgentTestN [] $ testServerConnectionAfterError t it "should deliver pending messages after agent restarting" $ smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t it "should concurrently deliver messages to connections without blocking" $@@ -126,25 +128,25 @@ bob <# ("", "alice", CON) alice <# ("", "bob", CON) -- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4- alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 4)- alice <# ("", "bob", SENT 4)- bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False- bob #: ("12", "alice", "ACK 4") #> ("12", "alice", OK)- alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 5)+ alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 5) alice <# ("", "bob", SENT 5)+ bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False+ bob #: ("12", "alice", "ACK 5") #> ("12", "alice", OK)+ alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 6)+ alice <# ("", "bob", SENT 6) bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False- bob #: ("13", "alice", "ACK 5") #> ("13", "alice", OK)- bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 6)- bob <# ("", "alice", SENT 6)- alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False- alice #: ("3a", "bob", "ACK 6") #> ("3a", "bob", OK)- bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 7)+ bob #: ("13", "alice", "ACK 6") #> ("13", "alice", OK)+ bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 7) bob <# ("", "alice", SENT 7)+ alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False+ alice #: ("3a", "bob", "ACK 7") #> ("3a", "bob", OK)+ bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 8)+ bob <# ("", "alice", SENT 8) alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False- alice #: ("4a", "bob", "ACK 7") #> ("4a", "bob", OK)+ alice #: ("4a", "bob", "ACK 8") #> ("4a", "bob", OK) alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)- bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 8)- bob <# ("", "alice", MERR 8 (SMP AUTH))+ bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 9)+ bob <# ("", "alice", MERR 9 (SMP AUTH)) alice #: ("6", "bob", "DEL") #> ("6", "bob", OK) alice #:# "nothing else should be delivered to alice" @@ -159,25 +161,25 @@ bob <# ("", aliceConn, INFO "alice's connInfo") bob <# ("", aliceConn, CON) alice <# ("", bobConn, CON)- alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 4)- alice <# ("", bobConn, SENT 4)- bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False- bob #: ("12", aliceConn, "ACK 4") #> ("12", aliceConn, OK)- alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 5)+ alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 5) alice <# ("", bobConn, SENT 5)+ bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False+ bob #: ("12", aliceConn, "ACK 5") #> ("12", aliceConn, OK)+ alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 6)+ alice <# ("", bobConn, SENT 6) bob <#= \case ("", c, Msg "how are you?") -> c == aliceConn; _ -> False- bob #: ("13", aliceConn, "ACK 5") #> ("13", aliceConn, OK)- bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 6)- bob <# ("", aliceConn, SENT 6)- alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False- alice #: ("3a", bobConn, "ACK 6") #> ("3a", bobConn, OK)- bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 7)+ bob #: ("13", aliceConn, "ACK 6") #> ("13", aliceConn, OK)+ bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 7) bob <# ("", aliceConn, SENT 7)+ alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False+ alice #: ("3a", bobConn, "ACK 7") #> ("3a", bobConn, OK)+ bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 8)+ bob <# ("", aliceConn, SENT 8) alice <#= \case ("", c, Msg "message 1") -> c == bobConn; _ -> False- alice #: ("4a", bobConn, "ACK 7") #> ("4a", bobConn, OK)+ alice #: ("4a", bobConn, "ACK 8") #> ("4a", bobConn, OK) alice #: ("5", bobConn, "OFF") #> ("5", bobConn, OK)- bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 8)- bob <# ("", aliceConn, MERR 8 (SMP AUTH))+ bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 9)+ bob <# ("", aliceConn, MERR 9 (SMP AUTH)) alice #: ("6", bobConn, "DEL") #> ("6", bobConn, OK) alice #:# "nothing else should be delivered to alice" @@ -194,10 +196,10 @@ alice <# ("", "bob", INFO "bob's connInfo 2") alice <# ("", "bob", CON) bob <# ("", "alice", CON)- alice #: ("3", "bob", "SEND :hi") #> ("3", "bob", MID 4)- alice <# ("", "bob", SENT 4)+ alice #: ("3", "bob", "SEND :hi") #> ("3", "bob", MID 5)+ alice <# ("", "bob", SENT 5) bob <#= \case ("", "alice", Msg "hi") -> True; _ -> False- bob #: ("13", "alice", "ACK 4") #> ("13", "alice", OK)+ bob #: ("13", "alice", "ACK 5") #> ("13", "alice", OK) tom #: ("21", "alice", "JOIN " <> cReq' <> " 14\ntom's connInfo") #> ("21", "alice", OK) ("", "alice_contact", Right (REQ aInvId' "tom's connInfo")) <- (alice <#:)@@ -207,10 +209,10 @@ alice <# ("", "tom", INFO "tom's connInfo 2") alice <# ("", "tom", CON) tom <# ("", "alice", CON)- alice #: ("5", "tom", "SEND :hi there") #> ("5", "tom", MID 4)- alice <# ("", "tom", SENT 4)+ alice #: ("5", "tom", "SEND :hi there") #> ("5", "tom", MID 5)+ alice <# ("", "tom", SENT 5) tom <#= \case ("", "alice", Msg "hi there") -> True; _ -> False- tom #: ("23", "alice", "ACK 4") #> ("23", "alice", OK)+ tom #: ("23", "alice", "ACK 5") #> ("23", "alice", OK) testContactConnRandomIds :: Transport c => TProxy c -> c -> c -> IO () testContactConnRandomIds _ alice bob = do@@ -230,10 +232,10 @@ alice <# ("", bobConn, CON) bob <# ("", aliceConn, CON) - alice #: ("3", bobConn, "SEND :hi") #> ("3", bobConn, MID 4)- alice <# ("", bobConn, SENT 4)+ alice #: ("3", bobConn, "SEND :hi") #> ("3", bobConn, MID 5)+ alice <# ("", bobConn, SENT 5) bob <#= \case ("", c, Msg "hi") -> c == aliceConn; _ -> False- bob #: ("13", aliceConn, "ACK 4") #> ("13", aliceConn, OK)+ bob #: ("13", aliceConn, "ACK 5") #> ("13", aliceConn, OK) testRejectContactRequest :: Transport c => TProxy c -> c -> c -> IO () testRejectContactRequest _ alice bob = do@@ -250,20 +252,20 @@ testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO () testSubscription _ alice1 alice2 bob = do (alice1, "alice") `connect` (bob, "bob")- bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 4)- bob <# ("", "alice", SENT 4)- alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False- alice1 #: ("1", "bob", "ACK 4") #> ("1", "bob", OK)- bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 5)+ bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 5) bob <# ("", "alice", SENT 5)+ alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False+ alice1 #: ("1", "bob", "ACK 5") #> ("1", "bob", OK)+ bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 6)+ bob <# ("", "alice", SENT 6) alice1 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False- alice1 #: ("2", "bob", "ACK 5") #> ("2", "bob", OK)+ alice1 #: ("2", "bob", "ACK 6") #> ("2", "bob", OK) alice2 #: ("21", "bob", "SUB") #> ("21", "bob", OK) alice1 <# ("", "bob", END)- bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 6)- bob <# ("", "alice", SENT 6)+ bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 7)+ bob <# ("", "alice", SENT 7) alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False- alice2 #: ("22", "bob", "ACK 6") #> ("22", "bob", OK)+ alice2 #: ("22", "bob", "ACK 7") #> ("22", "bob", OK) alice1 #:# "nothing else should be delivered to alice1" testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()@@ -279,40 +281,77 @@ testMsgDeliveryServerRestart t alice bob = do withServer $ do connect (alice, "alice") (bob, "bob")- bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 4)- bob <# ("", "alice", SENT 4)+ bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 5)+ bob <# ("", "alice", SENT 5) alice <#= \case ("", "bob", Msg "hi") -> True; _ -> False- alice #: ("11", "bob", "ACK 4") #> ("11", "bob", OK)+ alice #: ("11", "bob", "ACK 5") #> ("11", "bob", OK) alice #:# "nothing else delivered before the server is killed" alice <# ("", "bob", DOWN)- bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 5)+ bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 6) bob #:# "nothing else delivered before the server is restarted" alice #:# "nothing else delivered before the server is restarted" withServer $ do- bob <# ("", "alice", SENT 5)+ bob <# ("", "alice", SENT 6) alice <# ("", "bob", UP) alice <#= \case ("", "bob", Msg "hello again") -> True; _ -> False- alice #: ("12", "bob", "ACK 5") #> ("12", "bob", OK)+ alice #: ("12", "bob", "ACK 6") #> ("12", "bob", OK) removeFile testStoreLogFile where withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` () +testServerConnectionAfterError :: forall c. Transport c => TProxy c -> [c] -> IO ()+testServerConnectionAfterError t _ = do+ withAgent1 $ \bob -> do+ withAgent2 $ \alice -> do+ withServer $ do+ connect (bob, "bob") (alice, "alice")++ bob <# ("", "alice", DOWN)+ alice <# ("", "bob", DOWN)+ alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 5)+ alice #:# "nothing else delivered before the server is restarted"+ bob #:# "nothing else delivered before the server is restarted"++ withAgent1 $ \bob -> do+ withAgent2 $ \alice -> do+ bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))+ alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))+ withServer $ do+ alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 5; _ -> False+ alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 5; _ -> False+ bob <# ("", "alice", UP)+ bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False+ bob #: ("2", "alice", "ACK 5") #> ("2", "alice", OK)+ alice #: ("1", "bob", "SEND 11\nhello again") #> ("1", "bob", MID 6)+ alice <# ("", "bob", SENT 6)+ bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False++ removeFile testStoreLogFile+ removeFile testDB+ removeFile testDB2+ where+ withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()+ withAgent1 = withAgent agentTestPort testDB+ withAgent2 = withAgent agentTestPort2 testDB2+ withAgent :: String -> String -> (c -> IO a) -> IO a+ withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort+ testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO () testMsgDeliveryAgentRestart t bob = do withAgent $ \alice -> do withServer $ do connect (bob, "bob") (alice, "alice")- alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 4)- alice <# ("", "bob", SENT 4)+ alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 5)+ alice <# ("", "bob", SENT 5) bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False- bob #: ("11", "alice", "ACK 4") #> ("11", "alice", OK)+ bob #: ("11", "alice", "ACK 5") #> ("11", "alice", OK) bob #:# "nothing else delivered before the server is down" bob <# ("", "alice", DOWN)- alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 5)+ alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 6) alice #:# "nothing else delivered before the server is restarted" bob #:# "nothing else delivered before the server is restarted" @@ -322,11 +361,11 @@ alice <#= \case (corrId, "bob", cmd) -> (corrId == "3" && cmd == OK)- || (corrId == "" && cmd == SENT 5)+ || (corrId == "" && cmd == SENT 6) _ -> False bob <# ("", "alice", UP) bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False- bob #: ("12", "alice", "ACK 5") #> ("12", "alice", OK)+ bob #: ("12", "alice", "ACK 6") #> ("12", "alice", OK) removeFile testStoreLogFile removeFile testDB@@ -354,11 +393,11 @@ -- alice <# ("", "bob", SENT 1) -- bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False -- bob #: ("12", "alice", "ACK 1") #> ("12", "alice", OK)- bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 5)- bob <# ("", "alice", SENT 5)+ bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 6)+ bob <# ("", "alice", SENT 6) -- if delivery is blocked it won't go further alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False- alice #: ("3", "bob", "ACK 5") #> ("3", "bob", OK)+ alice #: ("3", "bob", "ACK 6") #> ("3", "bob", OK) testMsgDeliveryQuotaExceeded :: Transport c => TProxy c -> c -> c -> IO () testMsgDeliveryQuotaExceeded _ alice bob = do@@ -371,9 +410,9 @@ alice <#= \case ("", "bob", SENT m) -> m == mId; _ -> False (_, "bob", Right (MID _)) <- alice #: ("5", "bob", "SEND :over quota") - alice #: ("1", "bob2", "SEND :hello") #> ("1", "bob2", MID 4)+ alice #: ("1", "bob2", "SEND :hello") #> ("1", "bob2", MID 5) -- if delivery is blocked it won't go further- alice <# ("", "bob2", SENT 4)+ alice <# ("", "bob2", SENT 5) connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO () connect (h1, name1) (h2, name2) = do@@ -391,10 +430,10 @@ ("m1", name2', Right (MID mId)) <- h1 #: ("m1", name2, "SEND :" <> msg) name2' `shouldBe` name2 h1 <#= \case ("", n, SENT m) -> n == name2 && m == mId; _ -> False- ("", name1', Right (MSG MsgMeta {recipient = (msgId, _)} msg')) <- (h2 <#:)+ ("", name1', Right (MSG MsgMeta {recipient = (msgId', _)} msg')) <- (h2 <#:) name1' `shouldBe` name1 msg' `shouldBe` msg- h2 #: ("m2", name1, "ACK " <> bshow msgId) =#> \case ("m2", n, OK) -> n == name1; _ -> False+ h2 #: ("m2", name1, "ACK " <> bshow msgId') =#> \case ("m2", n, OK) -> n == name1; _ -> False -- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString) -- connect' h1 h2 = do
tests/AgentTests/FunctionalAPITests.hs view
@@ -10,7 +10,7 @@ import Control.Monad.Except (ExceptT, runExceptT) import Control.Monad.IO.Unlift import SMPAgentClient-import SMPClient (withSmpServer)+import SMPClient (testPort, withSmpServer, withSmpServerStoreLogOn) import Simplex.Messaging.Agent import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..)) import Simplex.Messaging.Agent.Protocol@@ -44,6 +44,8 @@ withSmpServer t testAsyncJoiningOfflineBeforeActivation it "should connect with both clients going offline" $ withSmpServer t testAsyncBothOffline+ it "should connect on the second attempt if server was offline" $+ testAsyncServerOffline t it "should notify after HELLO timeout" $ withSmpServer t testAsyncHelloTimeout @@ -59,26 +61,26 @@ get alice ##> ("", bobId, CON) get bob ##> ("", aliceId, INFO "alice's connInfo") get bob ##> ("", aliceId, CON)- -- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4- 4 <- sendMessage alice bobId "hello"- get alice ##> ("", bobId, SENT 4)- 5 <- sendMessage alice bobId "how are you?"+ -- message IDs 1 to 4 get assigned to control messages, so first MSG is assigned ID 5+ 5 <- sendMessage alice bobId "hello" get alice ##> ("", bobId, SENT 5)+ 6 <- sendMessage alice bobId "how are you?"+ get alice ##> ("", bobId, SENT 6) get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False- ackMessage bob aliceId 4- get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False ackMessage bob aliceId 5- 6 <- sendMessage bob aliceId "hello too"- get bob ##> ("", aliceId, SENT 6)- 7 <- sendMessage bob aliceId "message 1"+ get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False+ ackMessage bob aliceId 6+ 7 <- sendMessage bob aliceId "hello too" get bob ##> ("", aliceId, SENT 7)+ 8 <- sendMessage bob aliceId "message 1"+ get bob ##> ("", aliceId, SENT 8) get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False- ackMessage alice bobId 6- get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False ackMessage alice bobId 7+ get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False+ ackMessage alice bobId 8 suspendConnection alice bobId- 8 <- sendMessage bob aliceId "message 2"- get bob ##> ("", aliceId, MERR 8 (SMP AUTH))+ 9 <- sendMessage bob aliceId "message 2"+ get bob ##> ("", aliceId, MERR 9 (SMP AUTH)) deleteConnection alice bobId liftIO $ noMessages alice "nothing else should be delivered to alice" pure ()@@ -148,6 +150,30 @@ exchangeGreetings alice' bobId bob' aliceId pure () +testAsyncServerOffline :: ATransport -> IO ()+testAsyncServerOffline t = do+ alice <- getSMPAgentClient cfg+ bob <- getSMPAgentClient cfg {dbFile = testDB2}+ -- create connection and shutdown the server+ Right (bobId, cReq) <- withSmpServerStoreLogOn t testPort $ \_ ->+ runExceptT $ createConnection alice SCMInvitation+ -- connection fails+ Left (BROKER NETWORK) <- runExceptT $ joinConnection bob cReq "bob's connInfo"+ ("", bobId1, DOWN) <- get alice+ bobId1 `shouldBe` bobId+ -- connection succeeds after server start+ Right () <- withSmpServerStoreLogOn t testPort $ \_ -> runExceptT $ do+ ("", bobId2, UP) <- get alice+ liftIO $ bobId2 `shouldBe` bobId+ aliceId <- joinConnection bob cReq "bob's connInfo"+ ("", _, CONF confId "bob's connInfo") <- get alice+ allowConnection alice bobId confId "alice's connInfo"+ get alice ##> ("", bobId, CON)+ get bob ##> ("", aliceId, INFO "alice's connInfo")+ get bob ##> ("", aliceId, CON)+ exchangeGreetings alice bobId bob aliceId+ pure ()+ testAsyncHelloTimeout :: IO () testAsyncHelloTimeout = do alice <- getSMPAgentClient cfg@@ -161,11 +187,11 @@ exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO () exchangeGreetings alice bobId bob aliceId = do- 4 <- sendMessage alice bobId "hello"- get alice ##> ("", bobId, SENT 4)+ 5 <- sendMessage alice bobId "hello"+ get alice ##> ("", bobId, SENT 5) get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False- ackMessage bob aliceId 4- 5 <- sendMessage bob aliceId "hello too"- get bob ##> ("", aliceId, SENT 5)+ ackMessage bob aliceId 5+ 6 <- sendMessage bob aliceId "hello too"+ get bob ##> ("", aliceId, SENT 6) get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False- ackMessage alice bobId 5+ ackMessage alice bobId 6
tests/AgentTests/SQLiteTests.hs view
@@ -51,7 +51,7 @@ -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous -- IO operations on multiple similarly named files; error seems to be environment specific r <- randomIO :: IO Word32- createSQLiteStore (testDB <> show r) 4 Migrations.app+ createSQLiteStore (testDB <> show r) 4 Migrations.app True removeStore :: SQLiteStore -> IO () removeStore store = do@@ -173,7 +173,9 @@ SndQueue { server = SMPServer "smp.simplex.im" "5223" testKeyHash, sndId = "3456",+ sndPublicKey = Nothing, sndPrivateKey = testPrivateSignKey,+ e2ePubKey = Nothing, e2eDhSecret = testDhSecret, status = New }@@ -303,7 +305,9 @@ SndQueue { server = SMPServer "smp.simplex.im" "5223" testKeyHash, sndId = "2345",+ sndPublicKey = Nothing, sndPrivateKey = testPrivateSignKey,+ e2ePubKey = Nothing, e2eDhSecret = testDhSecret, status = New }@@ -393,7 +397,7 @@ sndMsgId = externalSndId, broker = (brokerId, ts) },- msgType = A_MSG_,+ msgType = AM_A_MSG_, msgBody = hw, internalHash, externalPrevSndHash = "hash_from_sender"@@ -422,7 +426,7 @@ { internalId, internalSndId, internalTs = ts,- msgType = A_MSG_,+ msgType = AM_A_MSG_, msgBody = hw, internalHash, prevMsgHash = internalHash
tests/CoreTests/ProtocolErrorTests.hs view
@@ -1,8 +1,9 @@ module CoreTests.ProtocolErrorTests where -import Simplex.Messaging.Agent.Protocol (AgentErrorType, agentErrorTypeP, serializeAgentError, serializeSmpErrorType, smpErrorTypeP)+import Simplex.Messaging.Agent.Protocol (AgentErrorType) import Simplex.Messaging.Parsers (parseAll) import Simplex.Messaging.Protocol (ErrorType)+import Simplex.Messaging.Encoding.String import Test.Hspec import Test.Hspec.QuickCheck (modifyMaxSuccess) import Test.QuickCheck@@ -11,8 +12,8 @@ protocolErrorTests = modifyMaxSuccess (const 1000) $ do describe "errors parsing / serializing" $ do it "should parse SMP protocol errors" . property $ \err ->- parseAll smpErrorTypeP (serializeSmpErrorType err)+ parseAll strP (strEncode err) == Right (err :: ErrorType) it "should parse SMP agent errors" . property $ \err ->- parseAll agentErrorTypeP (serializeAgentError err)+ parseAll strP (strEncode err) == Right (err :: AgentErrorType)
tests/SMPAgentClient.hs view
@@ -20,12 +20,14 @@ withSmpServerOn, withSmpServerThreadOn, )-import Simplex.Messaging.Agent (runSMPAgentBlocking) import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval+import Simplex.Messaging.Agent.Server (runSMPAgentBlocking) import Simplex.Messaging.Client (SMPClientConfig (..), smpDefaultConfig) import Simplex.Messaging.Transport+import Simplex.Messaging.Transport.Client+import Simplex.Messaging.Transport.KeepAlive import Test.Hspec import UnliftIO.Concurrent import UnliftIO.Directory@@ -156,7 +158,7 @@ cfg = defaultAgentConfig { tcpPort = agentTestPort,- smpServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"],+ initialSMPServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"], tbqSize = 1, dbFile = testDB, smpCfg =@@ -173,7 +175,7 @@ withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =- let cfg' = cfg {tcpPort = port', dbFile = db', smpServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}+ let cfg' = cfg {tcpPort = port', dbFile = db', initialSMPServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]} in serverBracket (\started -> runSMPAgentBlocking t started cfg') afterProcess@@ -189,7 +191,7 @@ testSMPAgentClientOn :: (Transport c, MonadUnliftIO m) => ServiceName -> (c -> m a) -> m a testSMPAgentClientOn port' client = do- runTransportClient agentTestHost port' testKeyHash $ \h -> do+ runTransportClient agentTestHost port' testKeyHash (Just defaultKeepAliveOpts) $ \h -> do line <- liftIO $ getLn h if line == "Welcome to SMP agent v" <> B.pack simplexMQVersion then client h
tests/SMPClient.hs view
@@ -21,6 +21,8 @@ import Simplex.Messaging.Server.Env.STM import Simplex.Messaging.Server.StoreLog (openReadStoreLog) import Simplex.Messaging.Transport+import Simplex.Messaging.Transport.Client+import Simplex.Messaging.Transport.KeepAlive import Test.Hspec import UnliftIO.Concurrent import qualified UnliftIO.Exception as E@@ -44,7 +46,7 @@ testSMPClient :: (Transport c, MonadUnliftIO m) => (THandle c -> m a) -> m a testSMPClient client =- runTransportClient testHost testPort testKeyHash $ \h ->+ runTransportClient testHost testPort testKeyHash (Just defaultKeepAliveOpts) $ \h -> liftIO (runExceptT $ clientHandshake h testKeyHash) >>= \case Right th -> client th Left e -> error $ show e@@ -59,6 +61,9 @@ queueIdBytes = 24, msgIdBytes = 24, storeLog = Nothing,+ allowNewQueues = True,+ messageTTL = Just $ 7 * 86400, -- seconds, 7 days+ expireMessagesInterval = Just 21600_000000, -- microseconds, 6 hours caCertificateFile = "tests/fixtures/ca.crt", privateKeyFile = "tests/fixtures/server.key", certificateFile = "tests/fixtures/server.crt"@@ -67,16 +72,16 @@ withSmpServerStoreLogOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a withSmpServerStoreLogOn t port' client = do s <- liftIO $ openReadStoreLog testStoreLogFile+ withSmpServerConfigOn t cfg {storeLog = Just s} port' client++withSmpServerConfigOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServerConfig -> ServiceName -> (ThreadId -> m a) -> m a+withSmpServerConfigOn t cfg' port' = serverBracket- (\started -> runSMPServerBlocking started cfg {transports = [(port', t)], storeLog = Just s})+ (\started -> runSMPServerBlocking started cfg' {transports = [(port', t)]}) (pure ())- client withSmpServerThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a-withSmpServerThreadOn t port' =- serverBracket- (\started -> runSMPServerBlocking started cfg {transports = [(port', t)]})- (pure ())+withSmpServerThreadOn t = withSmpServerConfigOn t cfg serverBracket :: MonadUnliftIO m => (TMVar Bool -> m ()) -> m () -> (ThreadId -> m a) -> m a serverBracket process afterProcess f = do
tests/ServerTests.hs view
@@ -9,7 +9,7 @@ module ServerTests where -import Control.Concurrent (ThreadId, killThread)+import Control.Concurrent (ThreadId, killThread, threadDelay) import Control.Concurrent.STM import Control.Exception (SomeException, try) import Control.Monad.Except (forM, forM_, runExceptT)@@ -21,6 +21,7 @@ import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String import Simplex.Messaging.Protocol+import Simplex.Messaging.Server.Env.STM (ServerConfig (..)) import Simplex.Messaging.Transport import System.Directory (removeFile) import System.TimeIt (timeItT)@@ -29,18 +30,23 @@ import Test.Hspec serverTests :: ATransport -> Spec-serverTests t = do+serverTests t@(ATransport t') = do describe "SMP syntax" $ syntaxTests t describe "SMP queues" $ do describe "NEW and KEY commands, SEND messages" $ testCreateSecure t describe "NEW, OFF and DEL commands, SEND messages" $ testCreateDelete t describe "Stress test" $ stressTest t+ describe "allowNewQueues setting" $ testAllowNewQueues t' describe "SMP messages" $ do describe "duplex communication over 2 SMP connections" $ testDuplex t describe "switch subscription to another TCP connection" $ testSwitchSub t describe "Store log" $ testWithStoreLog t describe "Timing of AUTH error" $ testTiming t describe "Message notifications" $ testMessageNotifications t+ describe "Message expiration" $ do+ testMsgExpireOnSend t'+ testMsgExpireOnInterval t'+ testMsgNOTExpireOnInterval t' pattern Resp :: CorrId -> QueueId -> BrokerMsg -> SignedTransmission BrokerMsg pattern Resp corrId queueId command <- (_, _, (corrId, queueId, Right command))@@ -204,6 +210,16 @@ closeConnection $ connection h2 subscribeQueues h3 +testAllowNewQueues :: forall c. Transport c => TProxy c -> Spec+testAllowNewQueues t =+ it "should prohibit creating new queues with allowNewQueues = False" $ do+ withSmpServerConfigOn (ATransport t) cfg {allowNewQueues = False} testPort $ \_ ->+ testSMPClient @c $ \h -> do+ (rPub, rKey) <- C.generateSignatureKeyPair C.SEd448+ (dhPub, _ :: C.PrivateKeyX25519) <- C.generateKeyPair'+ Resp "abcd" "" (ERR AUTH) <- signSendRecv h rKey ("abcd", "", NEW rPub dhPub)+ pure ()+ testDuplex :: ATransport -> Spec testDuplex (ATransport t) = it "should create 2 simplex connections and exchange messages" $@@ -465,6 +481,56 @@ 1000 `timeout` tGet @BrokerMsg nh1 >>= \case Nothing -> return () Just _ -> error "nothing else should be delivered to the 1st notifier's TCP connection"++testMsgExpireOnSend :: forall c. Transport c => TProxy c -> Spec+testMsgExpireOnSend t =+ it "should expire messages that are not received before messageTTL on SEND" $ do+ (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519+ withSmpServerConfigOn (ATransport t) cfg {messageTTL = Just 1} testPort $ \_ ->+ testSMPClient @c $ \sh -> do+ (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub+ let dec nonce = C.cbDecrypt dhShared (C.cbNonce nonce)+ Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, SEND "hello (should expire)")+ threadDelay 2500000+ Resp "2" _ OK <- signSendRecv sh sKey ("2", sId, SEND "hello (should NOT expire)")+ testSMPClient @c $ \rh -> do+ Resp "3" _ (MSG mId _ msg) <- signSendRecv rh rKey ("3", rId, SUB)+ (dec mId msg, Right "hello (should NOT expire)") #== "delivered"+ 1000 `timeout` tGet @BrokerMsg rh >>= \case+ Nothing -> return ()+ Just _ -> error "nothing else should be delivered"++testMsgExpireOnInterval :: forall c. Transport c => TProxy c -> Spec+testMsgExpireOnInterval t =+ it "should expire messages that are not received before messageTTL after expiry interval" $ do+ (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519+ withSmpServerConfigOn (ATransport t) cfg {messageTTL = Just 1, expireMessagesInterval = Just 1000000} testPort $ \_ ->+ testSMPClient @c $ \sh -> do+ (sId, rId, rKey, _) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub+ Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, SEND "hello (should expire)")+ threadDelay 2500000+ testSMPClient @c $ \rh -> do+ Resp "2" _ OK <- signSendRecv rh rKey ("2", rId, SUB)+ 1000 `timeout` tGet @BrokerMsg rh >>= \case+ Nothing -> return ()+ Just _ -> error "nothing should be delivered"++testMsgNOTExpireOnInterval :: forall c. Transport c => TProxy c -> Spec+testMsgNOTExpireOnInterval t =+ it "should NOT expire messages that are not received before messageTTL if expiry interval is not set" $ do+ (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519+ withSmpServerConfigOn (ATransport t) cfg {messageTTL = Just 1, expireMessagesInterval = Nothing} testPort $ \_ ->+ testSMPClient @c $ \sh -> do+ (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub+ let dec nonce = C.cbDecrypt dhShared (C.cbNonce nonce)+ Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, SEND "hello (should NOT expire)")+ threadDelay 2500000+ testSMPClient @c $ \rh -> do+ Resp "2" _ (MSG mId _ msg) <- signSendRecv rh rKey ("2", rId, SUB)+ (dec mId msg, Right "hello (should NOT expire)") #== "delivered"+ 1000 `timeout` tGet @BrokerMsg rh >>= \case+ Nothing -> return ()+ Just _ -> error "nothing else should be delivered" samplePubKey :: C.APublicVerifyKey samplePubKey = C.APublicVerifyKey C.SEd25519 "MCowBQYDK2VwAyEAfAOflyvbJv1fszgzkQ6buiZJVgSpQWsucXq7U6zjMgY="