simplexmq 0.3.1 → 0.3.2
raw patch · 20 files changed
+728/−482 lines, 20 filesdep +websocketsdep ~QuickCheckdep ~cryptonitedep ~template-haskell
Dependencies added: websockets
Dependency ranges changed: QuickCheck, cryptonite, template-haskell, unliftio-core
Files
- CHANGELOG.md +5/−0
- apps/smp-agent/Main.hs +3/−1
- apps/smp-server/Main.hs +184/−87
- simplexmq.cabal +23/−18
- src/Simplex/Messaging/Agent.hs +13/−12
- src/Simplex/Messaging/Agent/Protocol.hs +6/−7
- src/Simplex/Messaging/Agent/Store/SQLite.hs +1/−1
- src/Simplex/Messaging/Client.hs +83/−80
- src/Simplex/Messaging/Protocol.hs +4/−4
- src/Simplex/Messaging/Server.hs +16/−11
- src/Simplex/Messaging/Server/Env/STM.hs +2/−1
- src/Simplex/Messaging/Server/StoreLog.hs +6/−0
- src/Simplex/Messaging/Transport.hs +107/−60
- src/Simplex/Messaging/Transport/WebSockets.hs +62/−0
- tests/AgentTests.hs +32/−33
- tests/AgentTests/SQLiteTests.hs +63/−60
- tests/SMPAgentClient.hs +42/−36
- tests/SMPClient.hs +30/−29
- tests/ServerTests.hs +39/−40
- tests/Test.hs +7/−2
CHANGELOG.md view
@@ -1,3 +1,8 @@+# 0.3.2++- Support websockets+- SMP server CLI commands+ # 0.3.1 - Released to hackage.org
apps/smp-agent/Main.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeApplications #-} module Main where @@ -8,6 +9,7 @@ import Simplex.Messaging.Agent (runSMPAgent) import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Client (smpDefaultConfig)+import Simplex.Messaging.Transport (TCP, Transport (..)) cfg :: AgentConfig cfg =@@ -28,4 +30,4 @@ main = do putStrLn $ "SMP agent listening on port " ++ tcpPort (cfg :: AgentConfig) setLogLevel LogInfo -- LogError- withGlobalLogging logCfg $ runSMPAgent cfg+ withGlobalLogging logCfg $ runSMPAgent (transport @TCP) cfg
apps/smp-server/Main.hs view
@@ -3,37 +3,45 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TypeApplications #-} module Main where -import Control.Monad (unless, when)+import Control.Monad.Except+import Control.Monad.Trans.Except import qualified Crypto.Store.PKCS8 as S import Data.ByteString.Base64 (encode) import qualified Data.ByteString.Char8 as B import Data.Char (toLower)-import Data.Functor (($>))-import Data.Ini (lookupValue, readIniFile)+import Data.Ini (Ini, lookupValue, readIniFile)+import Data.Text (Text) import qualified Data.Text as T import Data.X509 (PrivKey (PrivKeyRSA))+import Network.Socket (ServiceName) import Options.Applicative import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Server (runSMPServer) import Simplex.Messaging.Server.Env.STM-import Simplex.Messaging.Server.StoreLog (StoreLog, openReadStoreLog)-import System.Directory (createDirectoryIfMissing, doesFileExist)+import Simplex.Messaging.Server.StoreLog (StoreLog, openReadStoreLog, storeLogFilePath)+import Simplex.Messaging.Transport (ATransport (..), TCP, Transport (..))+import Simplex.Messaging.Transport.WebSockets (WS)+import System.Directory (createDirectoryIfMissing, doesFileExist, removeFile) import System.Exit (exitFailure) import System.FilePath (combine) import System.IO (IOMode (..), hFlush, stdout) -cfg :: ServerConfig-cfg =+defaultServerPort :: ServiceName+defaultServerPort = "5223"++serverConfig :: ServerConfig+serverConfig = ServerConfig- { tcpPort = "5223",- tbqSize = 16,+ { tbqSize = 16, queueIdBytes = 12, msgIdBytes = 6,- storeLog = Nothing,- -- key is loaded from the file server_key in /etc/opt/simplex directory+ -- below parameters are set based on ini file /etc/opt/simplex/smp-server.ini+ transports = undefined,+ storeLog = undefined, serverPrivateKey = undefined } @@ -49,77 +57,163 @@ defaultStoreLogFile :: FilePath defaultStoreLogFile = combine logDir "smp-server-store.log" +iniFile :: FilePath+iniFile = combine cfgDir "smp-server.ini"++defaultKeyFile :: FilePath+defaultKeyFile = combine cfgDir "server_key"+ main :: IO () main = do opts <- getServerOpts- putStrLn "SMP Server (-h for help)"- ini <- readCreateIni opts- storeLog <- openStoreLog ini- pk <- readCreateKey- B.putStrLn $ "transport key hash: " <> serverKeyHash pk- putStrLn $ "listening on port " <> tcpPort cfg- runSMPServer cfg {serverPrivateKey = pk, storeLog}+ case serverCommand opts of+ ServerInit ->+ runExceptT (getConfig opts) >>= \case+ Right cfg -> do+ putStrLn "Error: server is already initialized. Start it with `smp-server start` command"+ printConfig cfg+ exitFailure+ Left _ -> do+ cfg <- initializeServer opts+ putStrLn "Server was initialized. Start it with `smp-server start` command"+ printConfig cfg+ ServerStart ->+ runExceptT (getConfig opts) >>= \case+ Right cfg -> runServer cfg+ Left e -> do+ putStrLn $ "Server is not initialized: " <> e+ putStrLn "Initialize server with `smp-server init` command"+ exitFailure+ ServerDelete -> do+ deleteServer+ putStrLn "Server key, config file and store log deleted" +getConfig :: ServerOpts -> ExceptT String IO ServerConfig+getConfig opts = do+ ini <- readIni+ pk <- readKey ini+ storeLog <- liftIO $ openStoreLog opts ini+ pure $ makeConfig ini pk storeLog++makeConfig :: IniOpts -> C.FullPrivateKey -> Maybe (StoreLog 'ReadMode) -> ServerConfig+makeConfig IniOpts {serverPort, enableWebsockets} pk storeLog =+ let transports = (serverPort, transport @TCP) : [("80", transport @WS) | enableWebsockets]+ in serverConfig {serverPrivateKey = pk, storeLog, transports}++printConfig :: ServerConfig -> IO ()+printConfig ServerConfig {serverPrivateKey, storeLog} = do+ B.putStrLn $ "transport key hash: " <> serverKeyHash serverPrivateKey+ putStrLn $ case storeLog of+ Just s -> "store log: " <> storeLogFilePath s+ Nothing -> "store log disabled"++initializeServer :: ServerOpts -> IO ServerConfig+initializeServer opts = do+ ini <- createIni opts+ pk <- createKey ini+ storeLog <- openStoreLog opts ini+ pure $ makeConfig ini pk storeLog++runServer :: ServerConfig -> IO ()+runServer cfg = do+ printConfig cfg+ forM_ (transports cfg) $ \(port, ATransport t) ->+ putStrLn $ "listening on port " <> port <> " (" <> transportName t <> ")"+ runSMPServer cfg++deleteServer :: IO ()+deleteServer = do+ ini <- runExceptT readIni+ deleteIfExists iniFile+ case ini of+ Right IniOpts {storeLogFile, serverKeyFile} -> do+ deleteIfExists storeLogFile+ deleteIfExists serverKeyFile+ Left _ -> do+ deleteIfExists defaultKeyFile+ deleteIfExists defaultStoreLogFile+ data IniOpts = IniOpts { enableStoreLog :: Bool,- storeLogFile :: FilePath+ storeLogFile :: FilePath,+ serverKeyFile :: FilePath,+ serverPort :: ServiceName,+ enableWebsockets :: Bool } -readCreateIni :: ServerOpts -> IO IniOpts-readCreateIni ServerOpts {configFile} = do- createDirectoryIfMissing True cfgDir- doesFileExist configFile >>= (`unless` createIni)- readIni+readIni :: ExceptT String IO IniOpts+readIni = do+ fileExists iniFile+ ini <- ExceptT $ readIniFile iniFile+ let enableStoreLog = (== Right "on") $ lookupValue "STORE_LOG" "enable" ini+ storeLogFile = opt defaultStoreLogFile "STORE_LOG" "file" ini+ serverKeyFile = opt defaultKeyFile "TRANSPORT" "key_file" ini+ serverPort = opt defaultServerPort "TRANSPORT" "port" ini+ enableWebsockets = (== Right "on") $ lookupValue "TRANSPORT" "websockets" ini+ pure IniOpts {enableStoreLog, storeLogFile, serverKeyFile, serverPort, enableWebsockets} where- readIni :: IO IniOpts- readIni = do- ini <- either exitError pure =<< readIniFile configFile- let enableStoreLog = (== Right "on") $ lookupValue "STORE_LOG" "enable" ini- storeLogFile = either (const defaultStoreLogFile) T.unpack $ lookupValue "STORE_LOG" "file" ini- pure IniOpts {enableStoreLog, storeLogFile}- exitError e = do- putStrLn $ "error reading config file " <> configFile <> ": " <> e- exitFailure- createIni :: IO ()- createIni = do- confirm $ "Save default ini file to " <> configFile- writeFile- configFile- "[STORE_LOG]\n\- \# The server uses STM memory to store SMP queues and messages,\n\- \# that will be lost on restart (e.g., as with redis).\n\- \# This option enables saving SMP queues to append only log,\n\- \# and restoring them when the server is started.\n\- \# Log is compacted on start (deleted queues are removed).\n\- \# The messages in the queues are not logged.\n\- \\n\- \# enable: on\n\- \# file: /var/opt/simplex/smp-server-store.log\n"+ opt :: String -> Text -> Text -> Ini -> String+ opt def section key ini = either (const def) T.unpack $ lookupValue section key ini -readCreateKey :: IO C.FullPrivateKey-readCreateKey = do- createDirectoryIfMissing True cfgDir- let path = combine cfgDir "server_key"- hasKey <- doesFileExist path- (if hasKey then readKey else createKey) path+createIni :: ServerOpts -> IO IniOpts+createIni ServerOpts {enableStoreLog} = do+ writeFile iniFile $+ "[STORE_LOG]\n\+ \# The server uses STM memory to store SMP queues and messages,\n\+ \# that will be lost on restart (e.g., as with redis).\n\+ \# This option enables saving SMP queues to append only log,\n\+ \# and restoring them when the server is started.\n\+ \# Log is compacted on start (deleted queues are removed).\n\+ \# The messages in the queues are not logged.\n\n"+ <> (if enableStoreLog then "" else "# ")+ <> "enable: on\n\+ \# file: "+ <> defaultStoreLogFile+ <> "\n\n\+ \[TRANSPORT]\n\n\+ \# key_file: "+ <> defaultKeyFile+ <> "\n\+ \# port: "+ <> defaultServerPort+ <> "\n\+ \websockets: on\n"+ pure+ IniOpts+ { enableStoreLog,+ storeLogFile = defaultStoreLogFile,+ serverKeyFile = defaultKeyFile,+ serverPort = defaultServerPort,+ enableWebsockets = True+ }++readKey :: IniOpts -> ExceptT String IO C.FullPrivateKey+readKey IniOpts {serverKeyFile} = do+ fileExists serverKeyFile+ liftIO (S.readKeyFile serverKeyFile) >>= \case+ [S.Unprotected (PrivKeyRSA pk)] -> pure $ C.FullPrivateKey pk+ [_] -> err "not RSA key"+ [] -> err "invalid key file format"+ _ -> err "more than one key" where- createKey :: FilePath -> IO C.FullPrivateKey- createKey path = do- confirm "Generate new server key pair"- (_, pk) <- C.generateKeyPair newKeySize- S.writeKeyFile S.TraditionalFormat path [PrivKeyRSA $ C.rsaPrivateKey pk]- pure pk- readKey :: FilePath -> IO C.FullPrivateKey- readKey path = do- S.readKeyFile path >>= \case- [S.Unprotected (PrivKeyRSA pk)] -> pure $ C.FullPrivateKey pk- [_] -> errorExit "not RSA key"- [] -> errorExit "invalid key file format"- _ -> errorExit "more than one key"- where- errorExit :: String -> IO b- errorExit e = putStrLn (e <> ": " <> path) >> exitFailure+ err :: String -> ExceptT String IO b+ err e = throwE $ e <> ": " <> serverKeyFile +createKey :: IniOpts -> IO C.FullPrivateKey+createKey IniOpts {serverKeyFile} = do+ createDirectoryIfMissing True cfgDir+ (_, pk) <- C.generateKeyPair newKeySize+ S.writeKeyFile S.TraditionalFormat serverKeyFile [PrivKeyRSA $ C.rsaPrivateKey pk]+ pure pk++fileExists :: FilePath -> ExceptT String IO ()+fileExists path = do+ exists <- liftIO $ doesFileExist path+ unless exists . throwE $ "file " <> path <> " not found"++deleteIfExists :: FilePath -> IO ()+deleteIfExists path = doesFileExist path >>= (`when` removeFile path)+ confirm :: String -> IO () confirm msg = do putStr $ msg <> " (y/N): "@@ -130,38 +224,41 @@ serverKeyHash :: C.FullPrivateKey -> B.ByteString serverKeyHash = encode . C.unKeyHash . C.publicKeyHash . C.publicKey -openStoreLog :: IniOpts -> IO (Maybe (StoreLog 'ReadMode))-openStoreLog IniOpts {enableStoreLog, storeLogFile = f}- | enableStoreLog = do+openStoreLog :: ServerOpts -> IniOpts -> IO (Maybe (StoreLog 'ReadMode))+openStoreLog ServerOpts {enableStoreLog = l} IniOpts {enableStoreLog = l', storeLogFile = f}+ | l || l' = do createDirectoryIfMissing True logDir- putStrLn ("store log: " <> f) Just <$> openReadStoreLog f- | otherwise = putStrLn "store log disabled" $> Nothing+ | otherwise = pure Nothing -newtype ServerOpts = ServerOpts- { configFile :: FilePath+data ServerOpts = ServerOpts+ { serverCommand :: ServerCommand,+ enableStoreLog :: Bool } +data ServerCommand = ServerInit | ServerStart | ServerDelete+ serverOpts :: Parser ServerOpts serverOpts = ServerOpts- <$> strOption- ( long "config"- <> short 'c'- <> metavar "INI_FILE"- <> help ("config file (" <> defaultIniFile <> ")")- <> value defaultIniFile+ <$> subparser+ ( command "init" (info (pure ServerInit) (progDesc "Initialize server: generate server key and ini file"))+ <> command "start" (info (pure ServerStart) (progDesc "Start server (ini: /etc/opt/simplex/smp-server.ini)"))+ <> command "delete" (info (pure ServerDelete) (progDesc "Delete server key, ini file and store log")) )- where- defaultIniFile = combine cfgDir "smp-server.ini"+ <*> switch+ ( long "store-log"+ <> short 'l'+ <> help "enable store log for SMP queues persistence"+ ) getServerOpts :: IO ServerOpts-getServerOpts = execParser opts+getServerOpts = customExecParser p opts where+ p = prefs showHelpOnEmpty opts = info (serverOpts <**> helper) ( fullDesc <> header "Simplex Messaging Protocol (SMP) Server"- <> progDesc "Start server with INI_FILE (created on first run)" )
simplexmq.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 386cce2d4e066eae195c77973524bda82fab223da4b8f504e9932f05b562594f+-- hash: 1f213808753ff701ca8bcdc7b3296d99346fdbe9e0b6fbcfb372f928bcffa525 name: simplexmq-version: 0.3.1+version: 0.3.2 synopsis: SimpleXMQ message broker description: This package includes <./docs/Simplex-Messaging-Server.html server>, <./docs/Simplex-Messaging-Client.html client> and@@ -50,6 +50,7 @@ Simplex.Messaging.Server.QueueStore.STM Simplex.Messaging.Server.StoreLog Simplex.Messaging.Transport+ Simplex.Messaging.Transport.WebSockets Simplex.Messaging.Util other-modules: Paths_simplexmq@@ -57,7 +58,7 @@ src ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns build-depends:- QuickCheck ==2.13.*+ QuickCheck ==2.14.* , ansi-terminal ==0.10.* , asn1-encoding ==0.9.* , asn1-types ==0.3.*@@ -67,7 +68,7 @@ , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.* , containers ==0.6.*- , cryptonite ==0.26.*+ , cryptonite ==0.27.* , directory ==1.3.* , filepath ==1.4.* , generic-random ==1.3.*@@ -80,12 +81,13 @@ , simple-logger ==0.1.* , sqlite-simple ==0.4.* , stm ==2.5.*- , template-haskell ==2.15.*+ , template-haskell ==2.16.* , text ==1.2.* , time ==1.9.* , transformers ==0.5.* , unliftio ==0.2.*- , unliftio-core ==0.1.*+ , unliftio-core ==0.2.*+ , websockets ==0.12.* , x509 ==1.7.* default-language: Haskell2010 @@ -97,7 +99,7 @@ apps/smp-agent ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded build-depends:- QuickCheck ==2.13.*+ QuickCheck ==2.14.* , ansi-terminal ==0.10.* , asn1-encoding ==0.9.* , asn1-types ==0.3.*@@ -107,7 +109,7 @@ , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.* , containers ==0.6.*- , cryptonite ==0.26.*+ , cryptonite ==0.27.* , directory ==1.3.* , filepath ==1.4.* , generic-random ==1.3.*@@ -121,12 +123,13 @@ , simplexmq , sqlite-simple ==0.4.* , stm ==2.5.*- , template-haskell ==2.15.*+ , template-haskell ==2.16.* , text ==1.2.* , time ==1.9.* , transformers ==0.5.* , unliftio ==0.2.*- , unliftio-core ==0.1.*+ , unliftio-core ==0.2.*+ , websockets ==0.12.* , x509 ==1.7.* default-language: Haskell2010 @@ -138,7 +141,7 @@ apps/smp-server ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded build-depends:- QuickCheck ==2.13.*+ QuickCheck ==2.14.* , ansi-terminal ==0.10.* , asn1-encoding ==0.9.* , asn1-types ==0.3.*@@ -148,7 +151,7 @@ , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.* , containers ==0.6.*- , cryptonite ==0.26.*+ , cryptonite ==0.27.* , cryptostore ==0.2.* , directory ==1.3.* , filepath ==1.4.*@@ -165,12 +168,13 @@ , simplexmq , sqlite-simple ==0.4.* , stm ==2.5.*- , template-haskell ==2.15.*+ , template-haskell ==2.16.* , text ==1.2.* , time ==1.9.* , transformers ==0.5.* , unliftio ==0.2.*- , unliftio-core ==0.1.*+ , unliftio-core ==0.2.*+ , websockets ==0.12.* , x509 ==1.7.* default-language: Haskell2010 @@ -190,7 +194,7 @@ ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns build-depends: HUnit ==1.6.*- , QuickCheck ==2.13.*+ , QuickCheck ==2.14.* , ansi-terminal ==0.10.* , asn1-encoding ==0.9.* , asn1-types ==0.3.*@@ -200,7 +204,7 @@ , base64-bytestring >=1.0 && <1.3 , bytestring ==0.10.* , containers ==0.6.*- , cryptonite ==0.26.*+ , cryptonite ==0.27.* , directory ==1.3.* , filepath ==1.4.* , generic-random ==1.3.*@@ -216,12 +220,13 @@ , simplexmq , sqlite-simple ==0.4.* , stm ==2.5.*- , template-haskell ==2.15.*+ , template-haskell ==2.16.* , text ==1.2.* , time ==1.9.* , timeit ==2.0.* , transformers ==0.5.* , unliftio ==0.2.*- , unliftio-core ==0.1.*+ , unliftio-core ==0.2.*+ , websockets ==0.12.* , x509 ==1.7.* default-language: Haskell2010
src/Simplex/Messaging/Agent.hs view
@@ -51,9 +51,8 @@ import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Protocol (CorrId (..), MsgBody, SenderPublicKey) import qualified Simplex.Messaging.Protocol as SMP-import Simplex.Messaging.Transport (putLn, runTCPServer)+import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), runTransportServer) import Simplex.Messaging.Util (bshow)-import System.IO (Handle) import System.Random (randomR) import UnliftIO.Async (race_) import qualified UnliftIO.Exception as E@@ -62,19 +61,21 @@ -- | 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) => AgentConfig -> m ()-runSMPAgent cfg = newEmptyTMVarIO >>= (`runSMPAgentBlocking` cfg)+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) => TMVar Bool -> AgentConfig -> m ()-runSMPAgentBlocking started cfg@AgentConfig {tcpPort} = runReaderT smpAgent =<< newSMPAgentEnv cfg+runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()+runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort} = runReaderT (smpAgent t) =<< newSMPAgentEnv cfg where- smpAgent :: (MonadUnliftIO m', MonadReader Env m') => m' ()- smpAgent = runTCPServer started tcpPort $ \h -> do- liftIO $ putLn h "Welcome to SMP v0.3.1 agent"+ smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()+ smpAgent _ = runTransportServer started tcpPort $ \(h :: c) -> do+ liftIO $ putLn h "Welcome to SMP v0.3.2 agent" c <- getSMPAgentClient logConnection c True race_ (connectClient h c) (runSMPAgentClient c)@@ -87,7 +88,7 @@ cfg <- asks config atomically $ newAgentClient n cfg -connectClient :: MonadUnliftIO m => Handle -> AgentClient -> m ()+connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m () connectClient h c = race_ (send h c) (receive h c) logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()@@ -103,7 +104,7 @@ s2 <- connectSQLiteStore db race_ (subscriber c s1) (client c s2) -receive :: forall m. MonadUnliftIO m => Handle -> AgentClient -> m ()+receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m () receive h c@AgentClient {rcvQ, sndQ} = forever $ do (corrId, cAlias, cmdOrErr) <- tGet SClient h case cmdOrErr of@@ -115,7 +116,7 @@ logClient c "-->" t atomically $ writeTBQueue q t -send :: MonadUnliftIO m => Handle -> AgentClient -> m ()+send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m () send h c@AgentClient {sndQ} = forever $ do t <- atomically $ readTBQueue sndQ tPut h t
src/Simplex/Messaging/Agent/Protocol.hs view
@@ -102,9 +102,8 @@ SenderPublicKey, ) import qualified Simplex.Messaging.Protocol as SMP-import Simplex.Messaging.Transport (TransportError, getLn, putLn, serializeTransportError, transportErrorP)+import Simplex.Messaging.Transport (Transport (..), TransportError, serializeTransportError, transportErrorP) import Simplex.Messaging.Util-import System.IO import Test.QuickCheck (Arbitrary (..)) import Text.Read import UnliftIO.Exception@@ -540,23 +539,23 @@ serializeMsg body = bshow (B.length body) <> "\n" <> body -- | Send raw (unparsed) SMP agent protocol transmission to TCP connection.-tPutRaw :: Handle -> ARawTransmission -> IO ()+tPutRaw :: Transport c => c -> ARawTransmission -> IO () tPutRaw h (corrId, connAlias, command) = do putLn h corrId putLn h connAlias putLn h command -- | Receive raw (unparsed) SMP agent protocol transmission from TCP connection.-tGetRaw :: Handle -> IO ARawTransmission+tGetRaw :: Transport c => c -> IO ARawTransmission tGetRaw h = (,,) <$> getLn h <*> getLn h <*> getLn h -- | Send SMP agent protocol command (or response) to TCP connection.-tPut :: MonadIO m => Handle -> ATransmission p -> m ()+tPut :: (Transport c, MonadIO m) => c -> ATransmission p -> m () tPut h (CorrId corrId, connAlias, command) = liftIO $ tPutRaw h (corrId, connAlias, serializeCommand command) -- | Receive client and agent transmissions from TCP connection.-tGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (ATransmissionOrError p)+tGet :: forall c m p. (Transport c, MonadIO m) => SAParty p -> c -> m (ATransmissionOrError p) tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody where tParseLoadBody :: ARawTransmission -> m (ATransmissionOrError p)@@ -595,7 +594,7 @@ ':' : body -> return . Right $ B.pack body str -> case readMaybe str :: Maybe Int of Just size -> liftIO $ do- body <- B.hGet h size+ body <- cGet h size s <- getLn h return $ if B.null s then Right body else Left $ CMD SIZE Nothing -> return . Left $ CMD SYNTAX
src/Simplex/Messaging/Agent/Store/SQLite.hs view
@@ -104,7 +104,7 @@ if tLim > t && DB.sqlError e == DB.ErrorBusy then do threadDelay t- loop (t * 3 `div` 2) (tLim - t)+ loop (t * 9 `div` 8) (tLim - t) else E.throwIO e instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteStore m where
src/Simplex/Messaging/Client.hs view
@@ -7,6 +7,7 @@ {-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-} -- | -- Module : Simplex.Messaging.Client@@ -60,9 +61,9 @@ import Simplex.Messaging.Agent.Protocol (SMPServer (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Protocol-import Simplex.Messaging.Transport (THandle (..), TransportError, clientHandshake, runTCPClient)+import Simplex.Messaging.Transport (ATransport (..), TCP, THandle (..), TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)+import Simplex.Messaging.Transport.WebSockets (WS) import Simplex.Messaging.Util (bshow, liftError, raceAny_)-import System.IO import System.Timeout -- | 'SMPClient' is a handle used to send commands to a specific SMP server.@@ -92,7 +93,7 @@ { -- | size of TBQueue to use for server commands and responses qSize :: Natural, -- | default SMP server port if port is not specified in SMPServer- defaultPort :: ServiceName,+ defaultTransport :: (ServiceName, ATransport), -- | timeout of TCP commands (microseconds) tcpTimeout :: Int, -- | period for SMP ping commands (microseconds)@@ -107,7 +108,7 @@ smpDefaultConfig = SMPClientConfig { qSize = 16,- defaultPort = "5223",+ defaultTransport = ("5223", transport @TCP), tcpTimeout = 4_000_000, smpPing = 30_000_000, smpCommandSize = 256@@ -126,88 +127,90 @@ -- 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@SMPServer {host, port, keyHash}- SMPClientConfig {qSize, defaultPort, tcpTimeout, smpPing}- msgQ- disconnected = do- c <- atomically mkSMPClient- thVar <- newEmptyTMVarIO- action <-- async $- runTCPClient host (fromMaybe defaultPort port) (client c thVar)- `finally` atomically (putTMVar thVar $ Left SMPNetworkError)- tHandle <- tcpTimeout `timeout` atomically (takeTMVar thVar)- pure $ case tHandle of- Just (Right THandle {blockSize}) -> Right c {action, blockSize}- Just (Left e) -> Left e- Nothing -> Left SMPNetworkError- where- mkSMPClient :: STM SMPClient- mkSMPClient = do- connected <- newTVar False- clientCorrId <- newTVar 0- sentCommands <- newTVar M.empty- sndQ <- newTBQueue qSize- rcvQ <- newTBQueue qSize- return- SMPClient- { action = undefined,- blockSize = undefined,- connected,- smpServer,- tcpTimeout,- clientCorrId,- sentCommands,- sndQ,- rcvQ,- msgQ- }+getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ disconnected =+ atomically mkSMPClient >>= runClient useTransport+ where+ mkSMPClient :: STM SMPClient+ mkSMPClient = do+ connected <- newTVar False+ clientCorrId <- newTVar 0+ sentCommands <- newTVar M.empty+ sndQ <- newTBQueue qSize+ rcvQ <- newTBQueue qSize+ return+ SMPClient+ { action = undefined,+ blockSize = undefined,+ connected,+ smpServer,+ tcpTimeout,+ clientCorrId,+ sentCommands,+ sndQ,+ rcvQ,+ msgQ+ } - client :: SMPClient -> TMVar (Either SMPClientError THandle) -> Handle -> IO ()- client c thVar h =- runExceptT (clientHandshake h keyHash) >>= \case- Right th -> clientTransport c thVar th- Left e -> atomically . putTMVar thVar . Left $ SMPTransportError e+ runClient :: (ServiceName, ATransport) -> SMPClient -> IO (Either SMPClientError SMPClient)+ runClient (port', ATransport t) c = do+ thVar <- newEmptyTMVarIO+ action <-+ async $+ runTransportClient (host smpServer) port' (client t c thVar)+ `finally` atomically (putTMVar thVar $ Left SMPNetworkError)+ bSize <- tcpTimeout `timeout` atomically (takeTMVar thVar)+ pure $ case bSize of+ Just (Right blockSize) -> Right c {action, blockSize}+ Just (Left e) -> Left e+ Nothing -> Left SMPNetworkError - clientTransport :: SMPClient -> TMVar (Either SMPClientError THandle) -> THandle -> IO ()- clientTransport c thVar th = do- atomically $ do- writeTVar (connected c) True- putTMVar thVar $ Right th- raceAny_ [send c th, process c, receive c th, ping c]- `finally` disconnected+ useTransport :: (ServiceName, ATransport)+ useTransport = case port smpServer of+ Nothing -> defaultTransport cfg+ Just "80" -> ("80", transport @WS)+ Just p -> (p, transport @TCP) - send :: SMPClient -> THandle -> IO ()- send SMPClient {sndQ} h = forever $ atomically (readTBQueue sndQ) >>= tPut h+ client :: forall c. Transport c => TProxy c -> SMPClient -> TMVar (Either SMPClientError Int) -> c -> IO ()+ client _ c thVar h =+ runExceptT (clientHandshake h $ keyHash smpServer) >>= \case+ Left e -> atomically . putTMVar thVar . Left $ SMPTransportError e+ Right th -> do+ atomically $ do+ writeTVar (connected c) True+ putTMVar thVar . Right $ blockSize (th :: THandle c)+ raceAny_ [send c th, process c, receive c th, ping c]+ `finally` disconnected - receive :: SMPClient -> THandle -> IO ()- receive SMPClient {rcvQ} h = forever $ tGet fromServer h >>= atomically . writeTBQueue rcvQ+ send :: Transport c => SMPClient -> THandle c -> IO ()+ send SMPClient {sndQ} h = forever $ atomically (readTBQueue sndQ) >>= tPut h - ping :: SMPClient -> IO ()- ping c = forever $ do- threadDelay smpPing- runExceptT $ sendSMPCommand c Nothing "" (Cmd SSender PING)+ receive :: Transport c => SMPClient -> THandle c -> IO ()+ receive SMPClient {rcvQ} h = forever $ tGet fromServer h >>= atomically . writeTBQueue rcvQ - process :: SMPClient -> IO ()- process SMPClient {rcvQ, sentCommands} = forever $ do- (_, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ- cs <- readTVarIO sentCommands- case M.lookup corrId cs of- Nothing -> do- case respOrErr of- Right (Cmd SBroker cmd) -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)- -- TODO send everything else to errQ and log in agent- _ -> return ()- Just Request {queueId, responseVar} -> atomically $ do- modifyTVar sentCommands $ M.delete corrId- putTMVar responseVar $- if queueId == qId- then case respOrErr of- Left e -> Left $ SMPResponseError e- Right (Cmd _ (ERR e)) -> Left $ SMPServerError e- Right r -> Right r- else Left SMPUnexpectedResponse+ ping :: SMPClient -> IO ()+ ping c = forever $ do+ threadDelay smpPing+ runExceptT $ sendSMPCommand c Nothing "" (Cmd SSender PING)++ process :: SMPClient -> IO ()+ process SMPClient {rcvQ, sentCommands} = forever $ do+ (_, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ+ cs <- readTVarIO sentCommands+ case M.lookup corrId cs of+ Nothing -> do+ case respOrErr of+ Right (Cmd SBroker cmd) -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)+ -- TODO send everything else to errQ and log in agent+ _ -> return ()+ Just Request {queueId, responseVar} -> atomically $ do+ modifyTVar sentCommands $ M.delete corrId+ putTMVar responseVar $+ if queueId == qId+ then case respOrErr of+ Left e -> Left $ SMPResponseError e+ Right (Cmd _ (ERR e)) -> Left $ SMPServerError e+ Right r -> Right r+ else Left SMPUnexpectedResponse -- | Disconnects SMP client from the server and terminates client threads. closeSMPClient :: SMPClient -> IO ()
src/Simplex/Messaging/Protocol.hs view
@@ -80,7 +80,7 @@ import Generic.Random (genericArbitraryU) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Parsers-import Simplex.Messaging.Transport (THandle, TransportError (..), tGetEncrypted, tPutEncrypted)+import Simplex.Messaging.Transport (THandle, Transport, TransportError (..), tGetEncrypted, tPutEncrypted) import Simplex.Messaging.Util import Test.QuickCheck (Arbitrary (..)) @@ -293,7 +293,7 @@ serializeErrorType = bshow -- | Send signed SMP transmission to TCP transport.-tPut :: THandle -> SignedRawTransmission -> IO (Either TransportError ())+tPut :: Transport c => THandle c -> SignedRawTransmission -> IO (Either TransportError ()) tPut th (C.Signature sig, t) = tPutEncrypted th $ encode sig <> " " <> t <> " " @@ -315,14 +315,14 @@ _ -> Left $ CMD PROHIBITED -- | Receive and parse transmission from the TCP transport.-tGetParse :: THandle -> IO (Either TransportError RawTransmission)+tGetParse :: Transport c => THandle c -> IO (Either TransportError RawTransmission) tGetParse th = (>>= parse transmissionP TEBadBlock) <$> tGetEncrypted th -- | Receive client and server transmissions. -- -- The first argument is used to limit allowed senders. -- 'fromClient' or 'fromServer' should be used here.-tGet :: forall m. MonadIO m => (Cmd -> Either ErrorType Cmd) -> THandle -> m SignedTransmissionOrError+tGet :: forall c m. (Transport c, MonadIO m) => (Cmd -> Either ErrorType Cmd) -> THandle c -> m SignedTransmissionOrError tGet fromParty th = liftIO (tGetParse th) >>= decodeParseValidate where decodeParseValidate :: Either TransportError RawTransmission -> m SignedTransmissionOrError
src/Simplex/Messaging/Server.hs view
@@ -36,6 +36,7 @@ import Data.Functor (($>)) import qualified Data.Map.Strict as M import Data.Time.Clock+import Network.Socket (ServiceName) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Protocol import Simplex.Messaging.Server.Env.STM@@ -46,7 +47,6 @@ import Simplex.Messaging.Server.StoreLog import Simplex.Messaging.Transport import Simplex.Messaging.Util-import UnliftIO.Async import UnliftIO.Concurrent import UnliftIO.Exception import UnliftIO.IO@@ -56,24 +56,29 @@ -- -- See a full server here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-server/Main.hs runSMPServer :: (MonadRandom m, MonadUnliftIO m) => ServerConfig -> m ()-runSMPServer cfg = newEmptyTMVarIO >>= (`runSMPServerBlocking` cfg)+runSMPServer cfg = do+ started <- newEmptyTMVarIO+ runSMPServerBlocking started cfg -- | Runs an SMP server 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). runSMPServerBlocking :: (MonadRandom m, MonadUnliftIO m) => TMVar Bool -> ServerConfig -> m ()-runSMPServerBlocking started cfg@ServerConfig {tcpPort} = do+runSMPServerBlocking started cfg@ServerConfig {transports} = do env <- newEnv cfg runReaderT smpServer env where- smpServer :: (MonadUnliftIO m, MonadReader Env m) => m ()+ smpServer :: (MonadUnliftIO m', MonadReader Env m') => m' () smpServer = do s <- asks server- race_ (runTCPServer started tcpPort runClient) (serverThread s)+ raceAny_ (serverThread s : map runServer transports) `finally` withLog closeStoreLog - serverThread :: MonadUnliftIO m => Server -> m ()+ runServer :: (MonadUnliftIO m', MonadReader Env m') => (ServiceName, ATransport) -> m' ()+ runServer (tcpPort, ATransport t) = runTransportServer started tcpPort (runClient t)++ serverThread :: MonadUnliftIO m' => Server -> m' () serverThread Server {subscribedQ, subscribers} = forever . atomically $ do (rId, clnt) <- readTBQueue subscribedQ cs <- readTVar subscribers@@ -82,14 +87,14 @@ Nothing -> return () writeTVar subscribers $ M.insert rId clnt cs -runClient :: (MonadUnliftIO m, MonadReader Env m) => Handle -> m ()-runClient h = do+runClient :: (Transport c, MonadUnliftIO m, MonadReader Env m) => TProxy c -> c -> m ()+runClient _ h = do keyPair <- asks serverKeyPair liftIO (runExceptT $ serverHandshake h keyPair) >>= \case Right th -> runClientTransport th Left _ -> pure () -runClientTransport :: (MonadUnliftIO m, MonadReader Env m) => THandle -> m ()+runClientTransport :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> m () runClientTransport th = do q <- asks $ tbqSize . config c <- atomically $ newClient q@@ -106,7 +111,7 @@ Sub {subThread = SubThread t} -> killThread t _ -> return () -receive :: (MonadUnliftIO m, MonadReader Env m) => THandle -> Client -> m ()+receive :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> Client -> m () receive h Client {rcvQ} = forever $ do (signature, (corrId, queueId, cmdOrError)) <- tGet fromClient h t <- case cmdOrError of@@ -114,7 +119,7 @@ Right cmd -> verifyTransmission (signature, (corrId, queueId, cmd)) atomically $ writeTBQueue rcvQ t -send :: MonadUnliftIO m => THandle -> Client -> m ()+send :: (Transport c, MonadUnliftIO m) => THandle c -> Client -> m () send h Client {sndQ} = forever $ do t <- atomically $ readTBQueue sndQ liftIO $ tPut h ("", serializeTransmission t)
src/Simplex/Messaging/Server/Env/STM.hs view
@@ -18,11 +18,12 @@ import Simplex.Messaging.Server.QueueStore (QueueRec (..)) import Simplex.Messaging.Server.QueueStore.STM import Simplex.Messaging.Server.StoreLog+import Simplex.Messaging.Transport (ATransport) import System.IO (IOMode (..)) import UnliftIO.STM data ServerConfig = ServerConfig- { tcpPort :: ServiceName,+ { transports :: [(ServiceName, ATransport)], tbqSize :: Natural, queueIdBytes :: Int, msgIdBytes :: Int,
src/Simplex/Messaging/Server/StoreLog.hs view
@@ -10,6 +10,7 @@ ( StoreLog, -- constructors are not exported openWriteStoreLog, openReadStoreLog,+ storeLogFilePath, closeStoreLog, logCreateQueue, logSecureQueue,@@ -87,6 +88,11 @@ openReadStoreLog f = do doesFileExist f >>= (`unless` writeFile f "") ReadStoreLog f <$> openFile f ReadMode++storeLogFilePath :: StoreLog a -> FilePath+storeLogFilePath = \case+ WriteStoreLog f _ -> f+ ReadStoreLog f _ -> f closeStoreLog :: StoreLog a -> IO () closeStoreLog = \case
src/Simplex/Messaging/Transport.hs view
@@ -9,6 +9,7 @@ {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-} -- | -- Module : Simplex.Messaging.Transport@@ -23,13 +24,18 @@ -- -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a module Simplex.Messaging.Transport- ( -- * TCP transport- runTCPServer,- runTCPClient,- putLn,- getLn,- trimCR,+ ( -- * Transport connection class+ Transport (..),+ TProxy (..),+ ATransport (..), + -- * Transport over TCP+ runTransportServer,+ runTransportClient,++ -- * TCP transport+ TCP (..),+ -- * SMP encrypted transport THandle (..), TransportError (..),@@ -39,6 +45,9 @@ tGetEncrypted, serializeTransportError, transportErrorP,++ -- * Trim trailing CR+ trimCR, ) where @@ -72,20 +81,53 @@ import UnliftIO.Concurrent import UnliftIO.Exception (Exception, IOException) import qualified UnliftIO.Exception as E-import qualified UnliftIO.IO as IO import UnliftIO.STM --- * TCP transport+-- * Transport connection class --- | Run TCP server on passed port and signal when server started and stopped via passed TMVar.+class Transport c where+ transport :: ATransport+ transport = ATransport (TProxy @c)++ transportName :: TProxy c -> String++ -- | Upgrade client socket to connection (used in the server)+ getServerConnection :: Socket -> IO c++ -- | Upgrade server socket to connection (used in the client)+ getClientConnection :: Socket -> IO c++ -- | Close connection+ closeConnection :: c -> IO ()++ -- | Read fixed number of bytes from connection+ cGet :: c -> Int -> IO ByteString++ -- | Write bytes to connection+ cPut :: c -> ByteString -> IO ()++ -- | Receive ByteString from connection, allowing LF or CRLF termination.+ getLn :: c -> IO ByteString++ -- | Send ByteString to connection terminating it with CRLF.+ putLn :: c -> ByteString -> IO ()+ putLn c = cPut c . (<> "\r\n")++data TProxy c = TProxy++data ATransport = forall c. Transport c => ATransport (TProxy c)++-- * Transport over TCP++-- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar. ----- All accepted TCP connection handles are passed to the passed function.-runTCPServer :: MonadUnliftIO m => TMVar Bool -> ServiceName -> (Handle -> m ()) -> m ()-runTCPServer started port server = do+-- All accepted connections are passed to the passed function.+runTransportServer :: (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> (c -> m ()) -> m ()+runTransportServer started port server = do clients <- newTVarIO S.empty E.bracket (liftIO $ startTCPServer started port) (liftIO . closeServer clients) \sock -> forever $ do- h <- liftIO $ acceptTCPConn sock- tid <- forkFinally (server h) (const $ IO.hClose h)+ c <- liftIO $ acceptConnection sock+ tid <- forkFinally (server c) (const $ liftIO $ closeConnection c) atomically . modifyTVar clients $ S.insert tid where closeServer :: TVar (Set ThreadId) -> Socket -> IO ()@@ -93,6 +135,8 @@ readTVarIO clients >>= mapM_ killThread close sock void . atomically $ tryPutTMVar started False+ acceptConnection :: Transport c => Socket -> IO c+ acceptConnection sock = accept sock >>= getServerConnection . fst startTCPServer :: TMVar Bool -> ServiceName -> IO Socket startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted@@ -107,18 +151,15 @@ bind sock $ addrAddress addr listen sock 1024 return sock- setStarted sock = atomically (putTMVar started True) >> pure sock--acceptTCPConn :: Socket -> IO Handle-acceptTCPConn sock = accept sock >>= getSocketHandle . fst+ setStarted sock = atomically (tryPutTMVar started True) >> pure sock -- | Connect to passed TCP host:port and pass handle to the client.-runTCPClient :: MonadUnliftIO m => HostName -> ServiceName -> (Handle -> m a) -> m a-runTCPClient host port client = do- h <- liftIO $ startTCPClient host port- client h `E.finally` IO.hClose h+runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> (c -> m a) -> m a+runTransportClient host port client = do+ c <- liftIO $ startTCPClient host port+ client c `E.finally` liftIO (closeConnection c) -startTCPClient :: HostName -> ServiceName -> IO Handle+startTCPClient :: forall c. Transport c => HostName -> ServiceName -> IO c startTCPClient host port = withSocketsDo $ resolve >>= tryOpen err where err :: IOException@@ -129,17 +170,30 @@ let hints = defaultHints {addrSocketType = Stream} in getAddrInfo (Just hints) (Just host) (Just port) - tryOpen :: IOException -> [AddrInfo] -> IO Handle+ tryOpen :: IOException -> [AddrInfo] -> IO c tryOpen e [] = E.throwIO e tryOpen _ (addr : as) = E.try (open addr) >>= either (`tryOpen` as) pure - open :: AddrInfo -> IO Handle+ open :: AddrInfo -> IO c open addr = do sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) connect sock $ addrAddress addr- getSocketHandle sock+ getClientConnection sock +-- * TCP transport++newtype TCP = TCP {tcpHandle :: Handle}++instance Transport TCP where+ transportName _ = "TCP"+ getServerConnection = fmap TCP . getSocketHandle+ getClientConnection = getServerConnection+ closeConnection = hClose . tcpHandle+ cGet = B.hGet . tcpHandle+ cPut = B.hPut . tcpHandle+ getLn = fmap trimCR . B.hGetLine . tcpHandle+ getSocketHandle :: Socket -> IO Handle getSocketHandle conn = do h <- socketToHandle conn ReadWriteMode@@ -148,14 +202,6 @@ hSetBuffering h LineBuffering return h --- | Send ByteString to TCP connection handle terminating it with CRLF.-putLn :: Handle -> ByteString -> IO ()-putLn h = B.hPut h . (<> "\r\n")---- | Receive ByteString from TCP connection handle, allowing LF or CRLF termination.-getLn :: Handle -> IO ByteString-getLn h = trimCR <$> B.hGetLine h- -- | Trim trailing CR from ByteString. trimCR :: ByteString -> ByteString trimCR "" = ""@@ -170,7 +216,7 @@ major (SMPVersion a b _ _) = (a, b) currentSMPVersion :: SMPVersion-currentSMPVersion = SMPVersion 0 3 1 0+currentSMPVersion = SMPVersion 0 3 2 0 serializeSMPVersion :: SMPVersion -> ByteString serializeSMPVersion (SMPVersion a b c d) = B.intercalate "." [bshow a, bshow b, bshow c, bshow d]@@ -180,9 +226,9 @@ let ver = A.decimal <* A.char '.' in SMPVersion <$> ver <*> ver <*> ver <*> A.decimal --- | The handle for SMP encrypted transport connection over TCP.-data THandle = THandle- { handle :: Handle,+-- | The handle for SMP encrypted transport connection over Transport .+data THandle c = THandle+ { connection :: c, sndKey :: SessionKey, rcvKey :: SessionKey, blockSize :: Int@@ -255,16 +301,16 @@ TEHandshake e -> bshow e -- | Encrypt and send block to SMP encrypted transport.-tPutEncrypted :: THandle -> ByteString -> IO (Either TransportError ())-tPutEncrypted THandle {handle = h, sndKey, blockSize} block =+tPutEncrypted :: Transport c => THandle c -> ByteString -> IO (Either TransportError ())+tPutEncrypted THandle {connection = c, sndKey, blockSize} block = encryptBlock sndKey (blockSize - C.authTagSize) block >>= \case Left _ -> pure $ Left TEEncrypt- Right (authTag, msg) -> Right <$> B.hPut h (C.authTagToBS authTag <> msg)+ Right (authTag, msg) -> Right <$> cPut c (C.authTagToBS authTag <> msg) -- | Receive and decrypt block from SMP encrypted transport.-tGetEncrypted :: THandle -> IO (Either TransportError ByteString)-tGetEncrypted THandle {handle = h, rcvKey, blockSize} =- B.hGet h blockSize >>= decryptBlock rcvKey >>= \case+tGetEncrypted :: Transport c => THandle c -> IO (Either TransportError ByteString)+tGetEncrypted THandle {connection = c, rcvKey, blockSize} =+ cGet c blockSize >>= decryptBlock rcvKey >>= \case Left _ -> pure $ Left TEDecrypt Right "" -> ioe_EOF Right msg -> pure $ Right msg@@ -294,14 +340,14 @@ -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a -- -- The numbers in function names refer to the steps in the document.-serverHandshake :: Handle -> C.FullKeyPair -> ExceptT TransportError IO THandle-serverHandshake h (k, pk) = do+serverHandshake :: forall c. Transport c => c -> C.FullKeyPair -> ExceptT TransportError IO (THandle c)+serverHandshake c (k, pk) = do liftIO sendHeaderAndPublicKey_1 encryptedKeys <- receiveEncryptedKeys_4 -- TODO server currently ignores blockSize returned by the client -- this is reserved for future support of streams ClientHandshake {blockSize = _, sndKey, rcvKey} <- decryptParseKeys_5 encryptedKeys- th <- liftIO $ transportHandle h rcvKey sndKey transportBlockSize -- keys are swapped here+ th <- liftIO $ transportHandle c rcvKey sndKey transportBlockSize -- keys are swapped here sendWelcome_6 th pure th where@@ -309,17 +355,18 @@ sendHeaderAndPublicKey_1 = do let sKey = C.encodePubKey k header = ServerHeader {blockSize = transportBlockSize, keySize = B.length sKey}- B.hPut h $ binaryServerHeader header <> sKey+ cPut c $ binaryServerHeader header+ cPut c sKey receiveEncryptedKeys_4 :: ExceptT TransportError IO ByteString receiveEncryptedKeys_4 =- liftIO (B.hGet h $ C.publicKeySize k) >>= \case+ liftIO (cGet c $ C.publicKeySize k) >>= \case "" -> throwE $ TEHandshake TERMINATED ks -> pure ks decryptParseKeys_5 :: ByteString -> ExceptT TransportError IO ClientHandshake decryptParseKeys_5 encKeys = liftError (const $ TEHandshake DECRYPT) (C.decryptOAEP pk encKeys) >>= liftEither . parseClientHandshake- sendWelcome_6 :: THandle -> ExceptT TransportError IO ()+ sendWelcome_6 :: THandle c -> ExceptT TransportError IO () sendWelcome_6 th = ExceptT . tPutEncrypted th $ serializeSMPVersion currentSMPVersion <> " " -- | Client SMP encrypted transport handshake.@@ -327,23 +374,23 @@ -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#appendix-a -- -- The numbers in function names refer to the steps in the document.-clientHandshake :: Handle -> Maybe C.KeyHash -> ExceptT TransportError IO THandle-clientHandshake h keyHash = do+clientHandshake :: forall c. Transport c => c -> Maybe C.KeyHash -> ExceptT TransportError IO (THandle c)+clientHandshake c keyHash = do (k, blkSize) <- getHeaderAndPublicKey_1_2 -- TODO currently client always uses the blkSize returned by the server keys@ClientHandshake {sndKey, rcvKey} <- liftIO $ generateKeys_3 blkSize sendEncryptedKeys_4 k keys- th <- liftIO $ transportHandle h sndKey rcvKey blkSize+ th <- liftIO $ transportHandle c sndKey rcvKey blkSize getWelcome_6 th >>= checkVersion pure th where getHeaderAndPublicKey_1_2 :: ExceptT TransportError IO (C.PublicKey, Int) getHeaderAndPublicKey_1_2 = do- header <- liftIO (B.hGet h serverHeaderSize)+ header <- liftIO (cGet c serverHeaderSize) ServerHeader {blockSize, keySize} <- liftEither $ parse serverHeaderP (TEHandshake HEADER) header when (blockSize < transportBlockSize || blockSize > maxTransportBlockSize) $ throwError $ TEHandshake HEADER- s <- liftIO $ B.hGet h keySize+ s <- liftIO $ cGet c keySize maybe (pure ()) (validateKeyHash_2 s) keyHash key <- liftEither $ parseKey s pure (key, blockSize)@@ -363,8 +410,8 @@ sendEncryptedKeys_4 :: C.PublicKey -> ClientHandshake -> ExceptT TransportError IO () sendEncryptedKeys_4 k keys = liftError (const $ TEHandshake ENCRYPT) (C.encryptOAEP k $ serializeClientHandshake keys)- >>= liftIO . B.hPut h- getWelcome_6 :: THandle -> ExceptT TransportError IO SMPVersion+ >>= liftIO . cPut c+ getWelcome_6 :: THandle c -> ExceptT TransportError IO SMPVersion getWelcome_6 th = ExceptT $ (>>= parseSMPVersion) <$> tGetEncrypted th parseSMPVersion :: ByteString -> Either TransportError SMPVersion parseSMPVersion = first (const $ TEHandshake VERSION) . A.parseOnly (smpVersionP <* A.space)@@ -428,13 +475,13 @@ parseClientHandshake :: ByteString -> Either TransportError ClientHandshake parseClientHandshake = parse clientHandshakeP $ TEHandshake AES_KEYS -transportHandle :: Handle -> SessionKey -> SessionKey -> Int -> IO THandle-transportHandle h sk rk blockSize = do+transportHandle :: c -> SessionKey -> SessionKey -> Int -> IO (THandle c)+transportHandle c sk rk blockSize = do sndCounter <- newTVarIO 0 rcvCounter <- newTVarIO 0 pure THandle- { handle = h,+ { connection = c, sndKey = sk {counter = sndCounter}, rcvKey = rk {counter = rcvCounter}, blockSize
+ src/Simplex/Messaging/Transport/WebSockets.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE InstanceSigs #-}++module Simplex.Messaging.Transport.WebSockets (WS (..)) where++import qualified Control.Exception as E+import Data.ByteString.Char8 (ByteString)+import qualified Data.ByteString.Char8 as B+import Network.Socket (Socket)+import Network.WebSockets+import Network.WebSockets.Stream (Stream)+import qualified Network.WebSockets.Stream as S+import Simplex.Messaging.Transport (TProxy, Transport (..), TransportError (..), trimCR)++data WS = WS {wsStream :: Stream, wsConnection :: Connection}++websocketsOpts :: ConnectionOptions+websocketsOpts =+ defaultConnectionOptions+ { connectionCompressionOptions = NoCompression,+ connectionFramePayloadSizeLimit = SizeLimit 8192,+ connectionMessageDataSizeLimit = SizeLimit 65536+ }++instance Transport WS where+ transportName :: TProxy WS -> String+ transportName _ = "WebSockets"++ getServerConnection :: Socket -> IO WS+ getServerConnection sock = do+ s <- S.makeSocketStream sock+ WS s <$> acceptClientRequest s+ where+ acceptClientRequest :: Stream -> IO Connection+ acceptClientRequest s = makePendingConnectionFromStream s websocketsOpts >>= acceptRequest++ getClientConnection :: Socket -> IO WS+ getClientConnection sock = do+ s <- S.makeSocketStream sock+ WS s <$> sendClientRequest s+ where+ sendClientRequest :: Stream -> IO Connection+ sendClientRequest s = newClientConnection s "" "/" websocketsOpts []++ closeConnection :: WS -> IO ()+ closeConnection = S.close . wsStream++ cGet :: WS -> Int -> IO ByteString+ cGet c n = do+ s <- receiveData (wsConnection c)+ if B.length s == n+ then pure s+ else E.throwIO TEBadBlock++ cPut :: WS -> ByteString -> IO ()+ cPut = sendBinaryData . wsConnection++ getLn :: WS -> IO ByteString+ getLn c = do+ s <- trimCR <$> receiveData (wsConnection c)+ if B.null s || B.last s /= '\n'+ then E.throwIO TEBadBlock+ else pure $ B.init s
tests/AgentTests.hs view
@@ -5,49 +5,40 @@ {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE ScopedTypeVariables #-} module AgentTests where -import AgentTests.SQLiteTests (storeStressTest, storeTests)+import AgentTests.SQLiteTests (storeTests) import Control.Concurrent import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import SMPAgentClient import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)-import System.IO (Handle)+import Simplex.Messaging.Transport (ATransport (..), TProxy (..), Transport (..)) import System.Timeout import Test.Hspec -agentTests :: Spec-agentTests = do- describe "SQLite store" do- storeTests- storeStressTest- describe "SMP agent protocol syntax" syntaxTests+agentTests :: ATransport -> Spec+agentTests (ATransport t) = do+ describe "SQLite store" storeTests+ describe "SMP agent protocol syntax" $ syntaxTests t describe "Establishing duplex connection" do it "should connect via one server and one agent" $- smpAgentTest2_1_1 testDuplexConnection+ smpAgentTest2_1_1 $ testDuplexConnection t it "should connect via one server and 2 agents" $- smpAgentTest2_2_1 testDuplexConnection+ smpAgentTest2_2_1 $ testDuplexConnection t it "should connect via 2 servers and 2 agents" $- smpAgentTest2_2_2 testDuplexConnection+ smpAgentTest2_2_2 $ testDuplexConnection t describe "Connection subscriptions" do it "should connect via one server and one agent" $- smpAgentTest3_1_1 testSubscription+ smpAgentTest3_1_1 $ testSubscription t it "should send notifications to client when server disconnects" $- smpAgentServerTest testSubscrNotification---- | simple test for one command with the expected response-(>#>) :: ARawTransmission -> ARawTransmission -> Expectation-command >#> response = smpAgentTest command `shouldReturn` response---- | simple test for one command with a predicate for the expected response-(>#>=) :: ARawTransmission -> ((ByteString, ByteString, [ByteString]) -> Bool) -> Expectation-command >#>= p = smpAgentTest command >>= (`shouldSatisfy` p . \(cId, cAlias, cmd) -> (cId, cAlias, B.words cmd))+ smpAgentServerTest $ testSubscrNotification t -- | send transmission `t` to handle `h` and get response-(#:) :: Handle -> (ByteString, ByteString, ByteString) -> IO (ATransmissionOrError 'Agent)+(#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (ATransmissionOrError 'Agent) h #: t = tPutRaw h t >> tGet SAgent h -- | action and expected response@@ -66,15 +57,15 @@ Left e -> error $ show e -- | receive message to handle `h` and validate that it is the expected one-(<#) :: Handle -> ATransmission 'Agent -> Expectation+(<#) :: Transport c => c -> ATransmission 'Agent -> Expectation h <# (corrId, cAlias, cmd) = tGet SAgent h `shouldReturn` (corrId, cAlias, Right cmd) -- | receive message to handle `h` and validate it using predicate `p`-(<#=) :: Handle -> (ATransmission 'Agent -> Bool) -> Expectation+(<#=) :: Transport c => c -> (ATransmission 'Agent -> Bool) -> Expectation h <#= p = tGet SAgent h >>= (`shouldSatisfy` p . correctTransmission) -- | test that nothing is delivered to handle `h` during 10ms-(#:#) :: Handle -> String -> Expectation+(#:#) :: Transport c => c -> String -> Expectation h #:# err = tryGet `shouldReturn` () where tryGet =@@ -85,8 +76,8 @@ pattern Msg :: MsgBody -> ACommand 'Agent pattern Msg msgBody <- MSG {msgBody, msgIntegrity = MsgOk} -testDuplexConnection :: Handle -> Handle -> IO ()-testDuplexConnection alice bob = do+testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO ()+testDuplexConnection _ alice bob = do ("1", "bob", Right (INV qInfo)) <- alice #: ("1", "bob", "NEW") let qInfo' = serializeSmpQueueInfo qInfo bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)@@ -104,8 +95,8 @@ alice #: ("6", "bob", "DEL") #> ("6", "bob", OK) alice #:# "nothing else should be delivered to alice" -testSubscription :: Handle -> Handle -> Handle -> IO ()-testSubscription alice1 alice2 bob = do+testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO ()+testSubscription _ alice1 alice2 bob = do ("1", "bob", Right (INV qInfo)) <- alice1 #: ("1", "bob", "NEW") let qInfo' = serializeSmpQueueInfo qInfo bob #: ("11", "alice", "JOIN " <> qInfo') #> ("", "alice", CON)@@ -120,8 +111,8 @@ alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False alice1 #:# "nothing else should be delivered to alice1" -testSubscrNotification :: (ThreadId, ThreadId) -> Handle -> IO ()-testSubscrNotification (server, _) client = do+testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()+testSubscrNotification _ (server, _) client = do client #: ("1", "conn1", "NEW") =#> \case ("1", "conn1", INV _) -> True; _ -> False client #:# "nothing should be delivered to client before the server is killed" killThread server@@ -130,8 +121,8 @@ samplePublicKey :: ByteString samplePublicKey = "rsa:MIIBoDANBgkqhkiG9w0BAQEFAAOCAY0AMIIBiAKCAQEAtn1NI2tPoOGSGfad0aUg0tJ0kG2nzrIPGLiz8wb3dQSJC9xkRHyzHhEE8Kmy2cM4q7rNZIlLcm4M7oXOTe7SC4x59bLQG9bteZPKqXu9wk41hNamV25PWQ4zIcIRmZKETVGbwN7jFMpH7wxLdI1zzMArAPKXCDCJ5ctWh4OWDI6OR6AcCtEj+toCI6N6pjxxn5VigJtwiKhxYpoUJSdNM60wVEDCSUrZYBAuDH8pOxPfP+Tm4sokaFDTIG3QJFzOjC+/9nW4MUjAOFll9PCp9kaEFHJ/YmOYKMWNOCCPvLS6lxA83i0UaardkNLNoFS5paWfTlroxRwOC2T6PwO2ywKBgDjtXcSED61zK1seocQMyGRINnlWdhceD669kIHju/f6kAayvYKW3/lbJNXCmyinAccBosO08/0sUxvtuniIo18kfYJE0UmP1ReCjhMP+O+yOmwZJini/QelJk/Pez8IIDDWnY1qYQsN/q7ocjakOYrpGG7mig6JMFpDJtD6istR" -syntaxTests :: Spec-syntaxTests = do+syntaxTests :: forall c. Transport c => TProxy c -> Spec+syntaxTests t = do it "unknown command" $ ("1", "5678", "HELLO") >#> ("1", "5678", "ERR CMD SYNTAX") describe "NEW" do describe "valid" do@@ -151,3 +142,11 @@ describe "invalid" do -- TODO: JOIN is not merged yet - to be added it "no parameters" $ ("321", "", "JOIN") >#> ("321", "", "ERR CMD SYNTAX")+ where+ -- simple test for one command with the expected response+ (>#>) :: ARawTransmission -> ARawTransmission -> Expectation+ command >#> response = smpAgentTest t command `shouldReturn` response++ -- simple test for one command with a predicate for the expected response+ (>#>=) :: ARawTransmission -> ((ByteString, ByteString, [ByteString]) -> Bool) -> Expectation+ command >#>= p = smpAgentTest t command >>= (`shouldSatisfy` p . \(cId, cAlias, cmd) -> (cId, cAlias, B.words cmd))
tests/AgentTests/SQLiteTests.hs view
@@ -5,7 +5,7 @@ {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE RecordWildCards #-} -module AgentTests.SQLiteTests (storeTests, storeStressTest) where+module AgentTests.SQLiteTests (storeTests) where import Control.Concurrent.Async (concurrently_) import Control.Monad (replicateM_)@@ -62,45 +62,48 @@ -- TODO add null port tests storeTests :: Spec-storeTests = withStore do- describe "store setup" do- testCompiledThreadsafe- testForeignKeysEnabled- describe "store methods" do- describe "Queue and Connection management" do- describe "createRcvConn" do- testCreateRcvConn- testCreateRcvConnDuplicate- describe "createSndConn" do- testCreateSndConn- testCreateSndConnDuplicate- describe "getAllConnAliases" testGetAllConnAliases- describe "getRcvConn" testGetRcvConn- describe "deleteConn" do- testDeleteRcvConn- testDeleteSndConn- testDeleteDuplexConn- describe "upgradeRcvConnToDuplex" do- testUpgradeRcvConnToDuplex- describe "upgradeSndConnToDuplex" do- testUpgradeSndConnToDuplex- describe "set Queue status" do- describe "setRcvQueueStatus" do- testSetRcvQueueStatus- testSetRcvQueueStatusNoQueue- describe "setSndQueueStatus" do- testSetSndQueueStatus- testSetSndQueueStatusNoQueue- testSetQueueStatusDuplex- describe "Msg management" do- describe "create Msg" do- testCreateRcvMsg- testCreateSndMsg- testCreateRcvAndSndMsgs+storeTests = do+ withStore2 do+ describe "stress test" testConcurrentWrites+ withStore do+ describe "store setup" do+ testCompiledThreadsafe+ testForeignKeysEnabled+ describe "store methods" do+ describe "Queue and Connection management" do+ describe "createRcvConn" do+ testCreateRcvConn+ testCreateRcvConnDuplicate+ describe "createSndConn" do+ testCreateSndConn+ testCreateSndConnDuplicate+ describe "getAllConnAliases" testGetAllConnAliases+ describe "getRcvConn" testGetRcvConn+ describe "deleteConn" do+ testDeleteRcvConn+ testDeleteSndConn+ testDeleteDuplexConn+ describe "upgradeRcvConnToDuplex" do+ testUpgradeRcvConnToDuplex+ describe "upgradeSndConnToDuplex" do+ testUpgradeSndConnToDuplex+ describe "set Queue status" do+ describe "setRcvQueueStatus" do+ testSetRcvQueueStatus+ testSetRcvQueueStatusNoQueue+ describe "setSndQueueStatus" do+ testSetSndQueueStatus+ testSetSndQueueStatusNoQueue+ testSetQueueStatusDuplex+ describe "Msg management" do+ describe "create Msg" do+ testCreateRcvMsg+ testCreateSndMsg+ testCreateRcvAndSndMsgs -storeStressTest :: Spec-storeStressTest = withStore2 $- it "should pass stress test on multiple concurrent write transactions" $ \(s1, s2) -> do+testConcurrentWrites :: SpecWith (SQLiteStore, SQLiteStore)+testConcurrentWrites =+ it "should complete multiple concurrent write transactions w/t sqlite busy errors" $ \(s1, s2) -> do _ <- runExceptT $ createRcvConn s1 rcvQueue1 concurrently_ (runTest s1) (runTest s2) where@@ -111,13 +114,13 @@ createRcvMsg store rcvQueue1 rcvMsgData testCompiledThreadsafe :: SpecWith SQLiteStore-testCompiledThreadsafe = do+testCompiledThreadsafe = it "compiled sqlite library should be threadsafe" $ \store -> do compileOptions <- DB.query_ (dbConn store) "pragma COMPILE_OPTIONS;" :: IO [[T.Text]] compileOptions `shouldNotContain` [["THREADSAFE=0"]] testForeignKeysEnabled :: SpecWith SQLiteStore-testForeignKeysEnabled = do+testForeignKeysEnabled = it "foreign keys should be enabled" $ \store -> do let inconsistentQuery = [sql|@@ -156,7 +159,7 @@ } testCreateRcvConn :: SpecWith SQLiteStore-testCreateRcvConn = do+testCreateRcvConn = it "should create RcvConnection and add SndQueue" $ \store -> do createRcvConn store rcvQueue1 `returnsResult` ()@@ -168,14 +171,14 @@ `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 sndQueue1) testCreateRcvConnDuplicate :: SpecWith SQLiteStore-testCreateRcvConnDuplicate = do+testCreateRcvConnDuplicate = it "should throw error on attempt to create duplicate RcvConnection" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 createRcvConn store rcvQueue1 `throwsError` SEConnDuplicate testCreateSndConn :: SpecWith SQLiteStore-testCreateSndConn = do+testCreateSndConn = it "should create SndConnection and add RcvQueue" $ \store -> do createSndConn store sndQueue1 `returnsResult` ()@@ -187,14 +190,14 @@ `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 sndQueue1) testCreateSndConnDuplicate :: SpecWith SQLiteStore-testCreateSndConnDuplicate = do+testCreateSndConnDuplicate = it "should throw error on attempt to create duplicate SndConnection" $ \store -> do _ <- runExceptT $ createSndConn store sndQueue1 createSndConn store sndQueue1 `throwsError` SEConnDuplicate testGetAllConnAliases :: SpecWith SQLiteStore-testGetAllConnAliases = do+testGetAllConnAliases = it "should get all conn aliases" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 _ <- runExceptT $ createSndConn store sndQueue1 {connAlias = "conn2"}@@ -202,7 +205,7 @@ `returnsResult` ["conn1" :: ConnAlias, "conn2" :: ConnAlias] testGetRcvConn :: SpecWith SQLiteStore-testGetRcvConn = do+testGetRcvConn = it "should get connection using rcv queue id and server" $ \store -> do let smpServer = SMPServer "smp.simplex.im" (Just "5223") testKeyHash let recipientId = "1234"@@ -211,7 +214,7 @@ `returnsResult` SomeConn SCRcv (RcvConnection (connAlias (rcvQueue1 :: RcvQueue)) rcvQueue1) testDeleteRcvConn :: SpecWith SQLiteStore-testDeleteRcvConn = do+testDeleteRcvConn = it "should create RcvConnection and delete it" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 getConn store "conn1"@@ -223,7 +226,7 @@ `throwsError` SEConnNotFound testDeleteSndConn :: SpecWith SQLiteStore-testDeleteSndConn = do+testDeleteSndConn = it "should create SndConnection and delete it" $ \store -> do _ <- runExceptT $ createSndConn store sndQueue1 getConn store "conn1"@@ -235,7 +238,7 @@ `throwsError` SEConnNotFound testDeleteDuplexConn :: SpecWith SQLiteStore-testDeleteDuplexConn = do+testDeleteDuplexConn = it "should create DuplexConnection and delete it" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1@@ -248,7 +251,7 @@ `throwsError` SEConnNotFound testUpgradeRcvConnToDuplex :: SpecWith SQLiteStore-testUpgradeRcvConnToDuplex = do+testUpgradeRcvConnToDuplex = it "should throw error on attempt to add SndQueue to SndConnection or DuplexConnection" $ \store -> do _ <- runExceptT $ createSndConn store sndQueue1 let anotherSndQueue =@@ -268,7 +271,7 @@ `throwsError` SEBadConnType CDuplex testUpgradeSndConnToDuplex :: SpecWith SQLiteStore-testUpgradeSndConnToDuplex = do+testUpgradeSndConnToDuplex = it "should throw error on attempt to add RcvQueue to RcvConnection or DuplexConnection" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 let anotherRcvQueue =@@ -290,7 +293,7 @@ `throwsError` SEBadConnType CDuplex testSetRcvQueueStatus :: SpecWith SQLiteStore-testSetRcvQueueStatus = do+testSetRcvQueueStatus = it "should update status of RcvQueue" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 getConn store "conn1"@@ -301,7 +304,7 @@ `returnsResult` SomeConn SCRcv (RcvConnection "conn1" rcvQueue1 {status = Confirmed}) testSetSndQueueStatus :: SpecWith SQLiteStore-testSetSndQueueStatus = do+testSetSndQueueStatus = it "should update status of SndQueue" $ \store -> do _ <- runExceptT $ createSndConn store sndQueue1 getConn store "conn1"@@ -312,7 +315,7 @@ `returnsResult` SomeConn SCSnd (SndConnection "conn1" sndQueue1 {status = Confirmed}) testSetQueueStatusDuplex :: SpecWith SQLiteStore-testSetQueueStatusDuplex = do+testSetQueueStatusDuplex = it "should update statuses of RcvQueue and SndQueue in DuplexConnection" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1@@ -328,13 +331,13 @@ `returnsResult` SomeConn SCDuplex (DuplexConnection "conn1" rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed}) testSetRcvQueueStatusNoQueue :: SpecWith SQLiteStore-testSetRcvQueueStatusNoQueue = do+testSetRcvQueueStatusNoQueue = xit "should throw error on attempt to update status of non-existent RcvQueue" $ \store -> do setRcvQueueStatus store rcvQueue1 Confirmed `throwsError` SEConnNotFound testSetSndQueueStatusNoQueue :: SpecWith SQLiteStore-testSetSndQueueStatusNoQueue = do+testSetSndQueueStatusNoQueue = xit "should throw error on attempt to update status of non-existent SndQueue" $ \store -> do setSndQueueStatus store sndQueue1 Confirmed `throwsError` SEConnNotFound@@ -367,7 +370,7 @@ `returnsResult` () testCreateRcvMsg :: SpecWith SQLiteStore-testCreateRcvMsg = do+testCreateRcvMsg = it "should reserve internal ids and create a RcvMsg" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 -- TODO getMsg to check message@@ -392,7 +395,7 @@ `returnsResult` () testCreateSndMsg :: SpecWith SQLiteStore-testCreateSndMsg = do+testCreateSndMsg = it "should create a SndMsg and return InternalId and PrevSndMsgHash" $ \store -> do _ <- runExceptT $ createSndConn store sndQueue1 -- TODO getMsg to check message@@ -400,7 +403,7 @@ testCreateSndMsg' store "hash_dummy" sndQueue1 $ mkSndMsgData (InternalId 2) (InternalSndId 2) "new_hash_dummy" testCreateRcvAndSndMsgs :: SpecWith SQLiteStore-testCreateRcvAndSndMsgs = do+testCreateRcvAndSndMsgs = it "should create multiple RcvMsg and SndMsg, correctly ordering internal Ids and returning previous state" $ \store -> do _ <- runExceptT $ createRcvConn store rcvQueue1 _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1
tests/SMPAgentClient.hs view
@@ -3,6 +3,7 @@ {-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-} module SMPAgentClient where @@ -28,7 +29,6 @@ import Test.Hspec import UnliftIO.Concurrent import UnliftIO.Directory-import UnliftIO.IO agentTestHost :: HostName agentTestHost = "localhost"@@ -51,44 +51,50 @@ testDB3 :: String testDB3 = "tests/tmp/smp-agent3.test.protocol.db" -smpAgentTest :: ARawTransmission -> IO ARawTransmission-smpAgentTest cmd = runSmpAgentTest $ \h -> tPutRaw h cmd >> tGetRaw h+smpAgentTest :: forall c. Transport c => TProxy c -> ARawTransmission -> IO ARawTransmission+smpAgentTest _ cmd = runSmpAgentTest $ \(h :: c) -> tPutRaw h cmd >> tGetRaw h -runSmpAgentTest :: (MonadUnliftIO m, MonadRandom m) => (Handle -> m a) -> m a-runSmpAgentTest test = withSmpServer . withSmpAgent $ testSMPAgentClient test+runSmpAgentTest :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => (c -> m a) -> m a+runSmpAgentTest test = withSmpServer t . withSmpAgent t $ testSMPAgentClient test+ where+ t = transport @c -runSmpAgentServerTest :: (MonadUnliftIO m, MonadRandom m) => ((ThreadId, ThreadId) -> Handle -> m a) -> m a+runSmpAgentServerTest :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => ((ThreadId, ThreadId) -> c -> m a) -> m a runSmpAgentServerTest test =- withSmpServerThreadOn testPort $- \server -> withSmpAgentThreadOn (agentTestPort, testPort, testDB) $+ withSmpServerThreadOn t testPort $+ \server -> withSmpAgentThreadOn t (agentTestPort, testPort, testDB) $ \agent -> testSMPAgentClient $ test (server, agent)+ where+ t = transport @c -smpAgentServerTest :: ((ThreadId, ThreadId) -> Handle -> IO ()) -> Expectation+smpAgentServerTest :: Transport c => ((ThreadId, ThreadId) -> c -> IO ()) -> Expectation smpAgentServerTest test' = runSmpAgentServerTest test' `shouldReturn` () -runSmpAgentTestN :: forall m a. (MonadUnliftIO m, MonadRandom m) => [(ServiceName, ServiceName, String)] -> ([Handle] -> m a) -> m a-runSmpAgentTestN agents test = withSmpServer $ run agents []+runSmpAgentTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => [(ServiceName, ServiceName, String)] -> ([c] -> m a) -> m a+runSmpAgentTestN agents test = withSmpServer t $ run agents [] where- run :: [(ServiceName, ServiceName, String)] -> [Handle] -> m a+ run :: [(ServiceName, ServiceName, String)] -> [c] -> m a run [] hs = test hs- run (a@(p, _, _) : as) hs = withSmpAgentOn a $ testSMPAgentClientOn p $ \h -> run as (h : hs)+ run (a@(p, _, _) : as) hs = withSmpAgentOn t a $ testSMPAgentClientOn p $ \h -> run as (h : hs)+ t = transport @c -runSmpAgentTestN_1 :: forall m a. (MonadUnliftIO m, MonadRandom m) => Int -> ([Handle] -> m a) -> m a-runSmpAgentTestN_1 nClients test = withSmpServer . withSmpAgent $ run nClients []+runSmpAgentTestN_1 :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => Int -> ([c] -> m a) -> m a+runSmpAgentTestN_1 nClients test = withSmpServer t . withSmpAgent t $ run nClients [] where- run :: Int -> [Handle] -> m a+ run :: Int -> [c] -> m a run 0 hs = test hs run n hs = testSMPAgentClient $ \h -> run (n - 1) (h : hs)+ t = transport @c -smpAgentTestN :: [(ServiceName, ServiceName, String)] -> ([Handle] -> IO ()) -> Expectation+smpAgentTestN :: Transport c => [(ServiceName, ServiceName, String)] -> ([c] -> IO ()) -> Expectation smpAgentTestN agents test' = runSmpAgentTestN agents test' `shouldReturn` () -smpAgentTestN_1 :: Int -> ([Handle] -> IO ()) -> Expectation+smpAgentTestN_1 :: Transport c => Int -> ([c] -> IO ()) -> Expectation smpAgentTestN_1 n test' = runSmpAgentTestN_1 n test' `shouldReturn` () -smpAgentTest2_2_2 :: (Handle -> Handle -> IO ()) -> Expectation+smpAgentTest2_2_2 :: forall c. Transport c => (c -> c -> IO ()) -> Expectation smpAgentTest2_2_2 test' =- withSmpServerOn testPort2 $+ withSmpServerOn (transport @c) testPort2 $ smpAgentTestN [ (agentTestPort, testPort, testDB), (agentTestPort2, testPort2, testDB2)@@ -98,7 +104,7 @@ _test [h1, h2] = test' h1 h2 _test _ = error "expected 2 handles" -smpAgentTest2_2_1 :: (Handle -> Handle -> IO ()) -> Expectation+smpAgentTest2_2_1 :: Transport c => (c -> c -> IO ()) -> Expectation smpAgentTest2_2_1 test' = smpAgentTestN [ (agentTestPort, testPort, testDB),@@ -109,13 +115,13 @@ _test [h1, h2] = test' h1 h2 _test _ = error "expected 2 handles" -smpAgentTest2_1_1 :: (Handle -> Handle -> IO ()) -> Expectation+smpAgentTest2_1_1 :: Transport c => (c -> c -> IO ()) -> Expectation smpAgentTest2_1_1 test' = smpAgentTestN_1 2 _test where _test [h1, h2] = test' h1 h2 _test _ = error "expected 2 handles" -smpAgentTest3 :: (Handle -> Handle -> Handle -> IO ()) -> Expectation+smpAgentTest3 :: Transport c => (c -> c -> c -> IO ()) -> Expectation smpAgentTest3 test' = smpAgentTestN [ (agentTestPort, testPort, testDB),@@ -127,7 +133,7 @@ _test [h1, h2, h3] = test' h1 h2 h3 _test _ = error "expected 3 handles" -smpAgentTest3_1_1 :: (Handle -> Handle -> Handle -> IO ()) -> Expectation+smpAgentTest3_1_1 :: Transport c => (c -> c -> c -> IO ()) -> Expectation smpAgentTest3_1_1 test' = smpAgentTestN_1 3 _test where _test [h1, h2, h3] = test' h1 h2 h3@@ -145,31 +151,31 @@ smpCfg = smpDefaultConfig { qSize = 1,- defaultPort = testPort,+ defaultTransport = (testPort, transport @TCP), tcpTimeout = 500_000 } } -withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a-withSmpAgentThreadOn (port', smpPort', db') =+withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a+withSmpAgentThreadOn t (port', smpPort', db') = let cfg' = cfg {tcpPort = port', dbFile = db', smpServers = L.fromList [SMPServer "localhost" (Just smpPort') testKeyHash]} in serverBracket- (`runSMPAgentBlocking` cfg')+ (\started -> runSMPAgentBlocking t started cfg') (removeFile db') -withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => (ServiceName, ServiceName, String) -> m a -> m a-withSmpAgentOn (port', smpPort', db') = withSmpAgentThreadOn (port', smpPort', db') . const+withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m a -> m a+withSmpAgentOn t (port', smpPort', db') = withSmpAgentThreadOn t (port', smpPort', db') . const -withSmpAgent :: (MonadUnliftIO m, MonadRandom m) => m a -> m a-withSmpAgent = withSmpAgentOn (agentTestPort, testPort, testDB)+withSmpAgent :: (MonadUnliftIO m, MonadRandom m) => ATransport -> m a -> m a+withSmpAgent t = withSmpAgentOn t (agentTestPort, testPort, testDB) -testSMPAgentClientOn :: MonadUnliftIO m => ServiceName -> (Handle -> m a) -> m a+testSMPAgentClientOn :: (Transport c, MonadUnliftIO m) => ServiceName -> (c -> m a) -> m a testSMPAgentClientOn port' client = do- runTCPClient agentTestHost port' $ \h -> do+ runTransportClient agentTestHost port' $ \h -> do line <- liftIO $ getLn h- if line == "Welcome to SMP v0.3.1 agent"+ if line == "Welcome to SMP v0.3.2 agent" then client h else error $ "wrong welcome message: " <> B.unpack line -testSMPAgentClient :: MonadUnliftIO m => (Handle -> m a) -> m a+testSMPAgentClient :: (Transport c, MonadUnliftIO m) => (c -> m a) -> m a testSMPAgentClient = testSMPAgentClientOn agentTestPort
tests/SMPClient.hs view
@@ -4,6 +4,7 @@ {-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-} module SMPClient where @@ -44,9 +45,9 @@ testStoreLogFile :: FilePath testStoreLogFile = "tests/tmp/smp-server-store.log" -testSMPClient :: MonadUnliftIO m => (THandle -> m a) -> m a+testSMPClient :: (Transport c, MonadUnliftIO m) => (THandle c -> m a) -> m a testSMPClient client =- runTCPClient testHost testPort $ \h ->+ runTransportClient testHost testPort $ \h -> liftIO (runExceptT $ clientHandshake h testKeyHash) >>= \case Right th -> client th Left e -> error $ show e@@ -54,7 +55,7 @@ cfg :: ServerConfig cfg = ServerConfig- { tcpPort = testPort,+ { transports = undefined, tbqSize = 1, queueIdBytes = 12, msgIdBytes = 6,@@ -91,18 +92,18 @@ \TmKzSAw7iVWwEUZR/PeiEKazqrpp9VU=" } -withSmpServerStoreLogOn :: (MonadUnliftIO m, MonadRandom m) => ServiceName -> (ThreadId -> m a) -> m a-withSmpServerStoreLogOn port client = do+withSmpServerStoreLogOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a+withSmpServerStoreLogOn t port client = do s <- liftIO $ openReadStoreLog testStoreLogFile serverBracket- (\started -> runSMPServerBlocking started cfg {tcpPort = port, storeLog = Just s})+ (\started -> runSMPServerBlocking started cfg {transports = [(port, t)], storeLog = Just s}) (pure ()) client -withSmpServerThreadOn :: (MonadUnliftIO m, MonadRandom m) => ServiceName -> (ThreadId -> m a) -> m a-withSmpServerThreadOn port =+withSmpServerThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a+withSmpServerThreadOn t port = serverBracket- (\started -> runSMPServerBlocking started cfg {tcpPort = port})+ (\started -> runSMPServerBlocking started cfg {transports = [(port, t)]}) (pure ()) serverBracket :: MonadUnliftIO m => (TMVar Bool -> m ()) -> m () -> (ThreadId -> m a) -> m a@@ -118,49 +119,49 @@ Nothing -> error $ "server did not " <> s _ -> pure () -withSmpServerOn :: (MonadUnliftIO m, MonadRandom m) => ServiceName -> m a -> m a-withSmpServerOn port = withSmpServerThreadOn port . const+withSmpServerOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> m a -> m a+withSmpServerOn t port = withSmpServerThreadOn t port . const -withSmpServer :: (MonadUnliftIO m, MonadRandom m) => m a -> m a-withSmpServer = withSmpServerOn testPort+withSmpServer :: (MonadUnliftIO m, MonadRandom m) => ATransport -> m a -> m a+withSmpServer t = withSmpServerOn t testPort -runSmpTest :: (MonadUnliftIO m, MonadRandom m) => (THandle -> m a) -> m a-runSmpTest test = withSmpServer $ testSMPClient test+runSmpTest :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => (THandle c -> m a) -> m a+runSmpTest test = withSmpServer (transport @c) $ testSMPClient test -runSmpTestN :: forall m a. (MonadUnliftIO m, MonadRandom m) => Int -> ([THandle] -> m a) -> m a-runSmpTestN nClients test = withSmpServer $ run nClients []+runSmpTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => Int -> ([THandle c] -> m a) -> m a+runSmpTestN nClients test = withSmpServer (transport @c) $ run nClients [] where- run :: Int -> [THandle] -> m a+ run :: Int -> [THandle c] -> m a run 0 hs = test hs run n hs = testSMPClient $ \h -> run (n - 1) (h : hs) -smpServerTest :: RawTransmission -> IO RawTransmission-smpServerTest cmd = runSmpTest $ \h -> tPutRaw h cmd >> tGetRaw h+smpServerTest :: forall c. Transport c => TProxy c -> RawTransmission -> IO RawTransmission+smpServerTest _ cmd = runSmpTest $ \(h :: THandle c) -> tPutRaw h cmd >> tGetRaw h -smpTest :: (THandle -> IO ()) -> Expectation-smpTest test' = runSmpTest test' `shouldReturn` ()+smpTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation+smpTest _ test' = runSmpTest test' `shouldReturn` () -smpTestN :: Int -> ([THandle] -> IO ()) -> Expectation+smpTestN :: Transport c => Int -> ([THandle c] -> IO ()) -> Expectation smpTestN n test' = runSmpTestN n test' `shouldReturn` () -smpTest2 :: (THandle -> THandle -> IO ()) -> Expectation-smpTest2 test' = smpTestN 2 _test+smpTest2 :: Transport c => TProxy c -> (THandle c -> THandle c -> IO ()) -> Expectation+smpTest2 _ test' = smpTestN 2 _test where _test [h1, h2] = test' h1 h2 _test _ = error "expected 2 handles" -smpTest3 :: (THandle -> THandle -> THandle -> IO ()) -> Expectation-smpTest3 test' = smpTestN 3 _test+smpTest3 :: Transport c => TProxy c -> (THandle c -> THandle c -> THandle c -> IO ()) -> Expectation+smpTest3 _ test' = smpTestN 3 _test where _test [h1, h2, h3] = test' h1 h2 h3 _test _ = error "expected 3 handles" -tPutRaw :: THandle -> RawTransmission -> IO ()+tPutRaw :: Transport c => THandle c -> RawTransmission -> IO () tPutRaw h (sig, corrId, queueId, command) = do let t = B.intercalate " " [corrId, queueId, command] void $ tPut h (C.Signature sig, t) -tGetRaw :: THandle -> IO RawTransmission+tGetRaw :: Transport c => THandle c -> IO RawTransmission tGetRaw h = do ("", (CorrId corrId, qId, Right cmd)) <- tGet fromServer h pure ("", corrId, encode qId, serializeCommand cmd)
tests/ServerTests.hs view
@@ -28,25 +28,25 @@ rsaKeySize :: Int rsaKeySize = 2048 `div` 8 -serverTests :: Spec-serverTests = do- describe "SMP syntax" syntaxTests+serverTests :: ATransport -> Spec+serverTests t = do+ describe "SMP syntax" $ syntaxTests t describe "SMP queues" do- describe "NEW and KEY commands, SEND messages" testCreateSecure- describe "NEW, OFF and DEL commands, SEND messages" testCreateDelete+ describe "NEW and KEY commands, SEND messages" $ testCreateSecure t+ describe "NEW, OFF and DEL commands, SEND messages" $ testCreateDelete t describe "SMP messages" do- describe "duplex communication over 2 SMP connections" testDuplex- describe "switch subscription to another SMP queue" testSwitchSub- describe "Store log" testWithStoreLog- describe "Timing of AUTH error" testTiming+ describe "duplex communication over 2 SMP connections" $ testDuplex t+ describe "switch subscription to another SMP queue" $ testSwitchSub t+ describe "Store log" $ testWithStoreLog t+ describe "Timing of AUTH error" $ testTiming t pattern Resp :: CorrId -> QueueId -> Command 'Broker -> SignedTransmissionOrError pattern Resp corrId queueId command <- ("", (corrId, queueId, Right (Cmd SBroker command))) -sendRecv :: THandle -> (ByteString, ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError+sendRecv :: Transport c => THandle c -> (ByteString, ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError sendRecv h (sgn, corrId, qId, cmd) = tPutRaw h (sgn, corrId, encode qId, cmd) >> tGet fromServer h -signSendRecv :: THandle -> C.SafePrivateKey -> (ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError+signSendRecv :: Transport c => THandle c -> C.SafePrivateKey -> (ByteString, ByteString, ByteString) -> IO SignedTransmissionOrError signSendRecv h pk (corrId, qId, cmd) = do let t = B.intercalate " " [corrId, encode qId, cmd] Right sig <- runExceptT $ C.sign pk t@@ -56,16 +56,13 @@ cmdSEND :: ByteString -> ByteString cmdSEND msg = serializeCommand (Cmd SSender . SEND $ msg) -(>#>) :: RawTransmission -> RawTransmission -> Expectation-command >#> response = smpServerTest command `shouldReturn` response- (#==) :: (HasCallStack, Eq a, Show a) => (a, a) -> String -> Assertion (actual, expected) #== message = assertEqual message expected actual -testCreateSecure :: Spec-testCreateSecure =+testCreateSecure :: ATransport -> Spec+testCreateSecure (ATransport t) = it "should create (NEW) and secure (KEY) queue" $- smpTest \h -> do+ smpTest t $ \h -> do (rPub, rKey) <- C.generateKeyPair rsaKeySize Resp "abcd" rId1 (IDS rId sId) <- signSendRecv h rKey ("abcd", "", "NEW " <> C.serializePubKey rPub) (rId1, "") #== "creates queue"@@ -114,10 +111,10 @@ Resp "dabc" _ err5 <- sendRecv h ("", "dabc", sId, "SEND 5 hello ") (err5, ERR AUTH) #== "rejects unsigned SEND" -testCreateDelete :: Spec-testCreateDelete =+testCreateDelete :: ATransport -> Spec+testCreateDelete (ATransport t) = it "should create (NEW), suspend (OFF) and delete (DEL) queue" $- smpTest2 \rh sh -> do+ smpTest2 t $ \rh sh -> do (rPub, rKey) <- C.generateKeyPair rsaKeySize Resp "abcd" rId1 (IDS rId sId) <- signSendRecv rh rKey ("abcd", "", "NEW " <> C.serializePubKey rPub) (rId1, "") #== "creates queue"@@ -182,10 +179,10 @@ Resp "cdab" _ err10 <- signSendRecv rh rKey ("cdab", rId, "SUB") (err10, ERR AUTH) #== "rejects SUB when deleted" -testDuplex :: Spec-testDuplex =+testDuplex :: ATransport -> Spec+testDuplex (ATransport t) = it "should create 2 simplex connections and exchange messages" $- smpTest2 \alice bob -> do+ smpTest2 t $ \alice bob -> do (arPub, arKey) <- C.generateKeyPair rsaKeySize Resp "abcd" _ (IDS aRcv aSnd) <- signSendRecv alice arKey ("abcd", "", "NEW " <> C.serializePubKey arPub) -- aSnd ID is passed to Bob out-of-band@@ -232,10 +229,10 @@ Resp "bcda" _ OK <- signSendRecv bob brKey ("bcda", bRcv, "ACK") (msg5, "how are you bob") #== "message received from alice" -testSwitchSub :: Spec-testSwitchSub =+testSwitchSub :: ATransport -> Spec+testSwitchSub (ATransport t) = it "should create simplex connections and switch subscription to another TCP connection" $- smpTest3 \rh1 rh2 sh -> do+ smpTest3 t $ \rh1 rh2 sh -> do (rPub, rKey) <- C.generateKeyPair rsaKeySize Resp "abcd" _ (IDS rId sId) <- signSendRecv rh1 rKey ("abcd", "", "NEW " <> C.serializePubKey rPub) Resp "bcda" _ ok1 <- sendRecv sh ("", "bcda", sId, "SEND 5 test1 ")@@ -270,15 +267,15 @@ Nothing -> return () Just _ -> error "nothing else is delivered to the 1st TCP connection" -testWithStoreLog :: Spec-testWithStoreLog =+testWithStoreLog :: ATransport -> Spec+testWithStoreLog at@(ATransport t) = it "should store simplex queues to log and restore them after server restart" $ do (sPub1, sKey1) <- C.generateKeyPair rsaKeySize (sPub2, sKey2) <- C.generateKeyPair rsaKeySize senderId1 <- newTVarIO "" senderId2 <- newTVarIO "" - withSmpServerStoreLogOn testPort . runTest $ \h -> do+ withSmpServerStoreLogOn at testPort . runTest t $ \h -> do (sId1, _, _) <- createAndSecureQueue h sPub1 atomically $ writeTVar senderId1 sId1 Resp "bcda" _ OK <- signSendRecv h sKey1 ("bcda", sId1, "SEND 5 hello ")@@ -294,13 +291,13 @@ logSize `shouldReturn` 5 - withSmpServerThreadOn testPort . runTest $ \h -> do+ withSmpServerThreadOn at testPort . runTest t $ \h -> do sId1 <- readTVarIO senderId1 -- fails if store log is disabled Resp "bcda" _ (ERR AUTH) <- signSendRecv h sKey1 ("bcda", sId1, "SEND 5 hello ") pure () - withSmpServerStoreLogOn testPort . runTest $ \h -> do+ withSmpServerStoreLogOn at testPort . runTest t $ \h -> do -- this queue is restored sId1 <- readTVarIO senderId1 Resp "bcda" _ OK <- signSendRecv h sKey1 ("bcda", sId1, "SEND 5 hello ")@@ -312,7 +309,7 @@ logSize `shouldReturn` 1 removeFile testStoreLogFile where- createAndSecureQueue :: THandle -> SenderPublicKey -> IO (SenderId, RecipientId, C.SafePrivateKey)+ createAndSecureQueue :: Transport c => THandle c -> SenderPublicKey -> IO (SenderId, RecipientId, C.SafePrivateKey) createAndSecureQueue h sPub = do (rPub, rKey) <- C.generateKeyPair rsaKeySize Resp "abcd" "" (IDS rId sId) <- signSendRecv h rKey ("abcd", "", "NEW " <> C.serializePubKey rPub)@@ -321,8 +318,8 @@ (rId', rId) #== "same queue ID" pure (sId, rId, rKey) - runTest :: (THandle -> IO ()) -> ThreadId -> Expectation- runTest test' server = do+ runTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> ThreadId -> Expectation+ runTest _ test' server = do testSMPClient test' `shouldReturn` () killThread server @@ -332,10 +329,10 @@ Right l -> pure l Left (_ :: SomeException) -> logSize -testTiming :: Spec-testTiming =+testTiming :: ATransport -> Spec+testTiming (ATransport t) = it "should have similar time for auth error, whether queue exists or not, for all key sizes" $- smpTest2 \rh sh ->+ smpTest2 t $ \rh sh -> mapM_ (testSameTiming rh sh) [ (128, 128, 100),@@ -351,7 +348,7 @@ where timeRepeat n = fmap fst . timeItT . forM_ (replicate n ()) . const similarTime t1 t2 = abs (t1 - t2) / t1 < 0.2 `shouldBe` True- testSameTiming :: THandle -> THandle -> (Int, Int, Int) -> Expectation+ testSameTiming :: Transport c => THandle c -> THandle c -> (Int, Int, Int) -> Expectation testSameTiming rh sh (senderKeySize, badKeySize, n) = do (rPub, rKey) <- C.generateKeyPair rsaKeySize Resp "abcd" "" (IDS rId sId) <- signSendRecv rh rKey ("abcd", "", "NEW " <> C.serializePubKey rPub)@@ -377,8 +374,8 @@ sampleSig :: ByteString sampleSig = "\128\207*\159eq\220i!\"\157\161\130\184\226\246\232_\\\170`\180\160\230sI\154\197\211\252\SUB\246\206ELL\t9K\ESC\196?\128\215%\222\148\NAK;9\155f\164\217e\242\156\CAN9\253\r\170\174'w\211\228?\205)\215\150\255\247z\DC115\DC1{\bn\145\rKD,K\230\202d8\233\167|7y\t_S\EM\248\EOT\216\172\167d\181\224)\137\ACKo\197j#c\217\243\228.\167\228\205\144\vr\134" -syntaxTests :: Spec-syntaxTests = do+syntaxTests :: ATransport -> Spec+syntaxTests (ATransport t) = do it "unknown command" $ ("", "abcd", "1234", "HELLO") >#> ("", "abcd", "1234", "ERR CMD SYNTAX") describe "NEW" do it "no parameters" $ (sampleSig, "bcda", "", "NEW") >#> ("", "bcda", "", "ERR CMD SYNTAX")@@ -414,3 +411,5 @@ it "wrong terminator" $ (sampleSig, "bcda", "12345678", cmd <> "=") >#> ("", "bcda", "12345678", "ERR CMD SYNTAX") it "no signature" $ ("", "cdab", "12345678", cmd) >#> ("", "cdab", "12345678", "ERR CMD NO_AUTH") it "no queue ID" $ (sampleSig, "dabc", "", cmd) >#> ("", "dabc", "", "ERR CMD NO_AUTH")+ (>#>) :: RawTransmission -> RawTransmission -> Expectation+ command >#> response = smpServerTest t command `shouldReturn` response
tests/Test.hs view
@@ -1,6 +1,10 @@+{-# LANGUAGE TypeApplications #-}+ import AgentTests import ProtocolErrorTests import ServerTests+import Simplex.Messaging.Transport (TCP, Transport (..))+import Simplex.Messaging.Transport.WebSockets (WS) import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive) import Test.Hspec @@ -9,6 +13,7 @@ createDirectoryIfMissing False "tests/tmp" hspec $ do describe "Protocol errors" protocolErrorTests- describe "SMP server" serverTests- describe "SMP client agent" agentTests+ describe "SMP server via TCP" $ serverTests (transport @TCP)+ describe "SMP server via WebSockets" $ serverTests (transport @WS)+ describe "SMP client agent" $ agentTests (transport @TCP) removeDirectoryRecursive "tests/tmp"