From 4b8f6417f85826a270d5b58d1ec0a636fcc9b800 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Mon, 28 Dec 2020 13:55:53 +0000 Subject: [PATCH] add correlationId to SMP transmissions, fix tests --- src/Simplex/Messaging/Server.hs | 38 ++++++------- src/Simplex/Messaging/Server/Transmission.hs | 8 +-- src/Simplex/Messaging/Transport.hs | 33 +++++++----- tests/SMPClient.hs | 2 +- tests/Test.hs | 57 +++++++++++--------- 5 files changed, 76 insertions(+), 62 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 85e1b3395..5508081c3 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -51,7 +51,7 @@ runSMPServer cfg@ServerConfig {tcpPort} = do (rId, clnt) <- readTBQueue subscribedQ cs <- readTVar subscribers case M.lookup rId cs of - Just Client {rcvQ} -> writeTBQueue rcvQ (rId, Cmd SBroker END) + Just Client {rcvQ} -> writeTBQueue rcvQ (B.empty, rId, Cmd SBroker END) Nothing -> return () writeTVar subscribers $ M.insert rId clnt cs @@ -75,10 +75,10 @@ cancelSub = \case receive :: (MonadUnliftIO m, MonadReader Env m) => Handle -> Client -> m () receive h Client {rcvQ} = forever $ do - (signature, (queueId, cmdOrError)) <- tGet fromClient h + (signature, (corrId, queueId, cmdOrError)) <- tGet fromClient h signed <- case cmdOrError of - Left e -> return . mkResp queueId $ ERR e - Right cmd -> verifyTransmission signature queueId cmd + Left e -> return . mkResp corrId queueId $ ERR e + Right cmd -> verifyTransmission (signature, (corrId, queueId, cmd)) atomically $ writeTBQueue rcvQ signed send :: MonadUnliftIO m => Handle -> Client -> m () @@ -86,12 +86,12 @@ send h Client {sndQ} = forever $ do signed <- atomically $ readTBQueue sndQ tPut h (B.empty, signed) -mkResp :: QueueId -> Command 'Broker -> Signed -mkResp queueId command = (queueId, Cmd SBroker command) +mkResp :: CorrelationId -> QueueId -> Command 'Broker -> Signed +mkResp corrId queueId command = (corrId, queueId, Cmd SBroker command) -verifyTransmission :: forall m. (MonadUnliftIO m, MonadReader Env m) => Signature -> QueueId -> Cmd -> m Signed -verifyTransmission signature queueId cmd = do - (queueId,) <$> case cmd of +verifyTransmission :: forall m. (MonadUnliftIO m, MonadReader Env m) => Transmission -> m Signed +verifyTransmission (signature, (corrId, queueId, cmd)) = do + (corrId,queueId,) <$> case cmd of Cmd SBroker _ -> return $ smpErr INTERNAL -- it can only be client command, because `fromClient` was used Cmd SRecipient (NEW _) -> return cmd Cmd SRecipient _ -> withQueueRec SRecipient $ verifySignature . recipientKey @@ -121,11 +121,11 @@ client clnt@Client {subscriptions, rcvQ, sndQ} Server {subscribedQ} = >>= atomically . writeTBQueue sndQ where processCommand :: Signed -> m Signed - processCommand (queueId, cmd) = do + processCommand (corrId, queueId, cmd) = do st <- asks queueStore case cmd of - Cmd SBroker END -> unsubscribeQueue $> (queueId, cmd) - Cmd SBroker _ -> return (queueId, cmd) + Cmd SBroker END -> unsubscribeQueue $> (corrId, queueId, cmd) + Cmd SBroker _ -> return (corrId, queueId, cmd) Cmd SSender (SEND msgBody) -> sendMessage st msgBody Cmd SRecipient command -> case command of NEW rKey -> createQueue st rKey @@ -136,7 +136,7 @@ client clnt@Client {subscriptions, rcvQ, sndQ} Server {subscribedQ} = DEL -> delQueueAndMsgs st where createQueue :: QueueStore -> RecipientKey -> m Signed - createQueue st rKey = mkResp B.empty <$> addSubscribe + createQueue st rKey = mkResp corrId B.empty <$> addSubscribe where addSubscribe = addQueueRetry 3 >>= \case @@ -217,7 +217,7 @@ client clnt@Client {subscriptions, rcvQ, sndQ} Server {subscribedQ} = q <- atomically $ getMsgQueue ms rId atomically (tryPeek q) >>= \case Nothing -> forkSub q $> ok - Just msg -> atomically setDelivered $> msgResp rId msg + Just msg -> atomically setDelivered $> msgResp corrId rId msg _ -> return ok where forkSub :: MsgQueue -> m () @@ -231,7 +231,7 @@ client clnt@Client {subscriptions, rcvQ, sndQ} Server {subscribedQ} = subscriber :: MsgQueue -> m () subscriber q = atomically $ do msg <- peekMsg q - writeTBQueue sndQ $ msgResp rId msg + writeTBQueue sndQ $ msgResp B.empty rId msg setSub (\s -> s {subThread = NoSub}) void setDelivered @@ -250,16 +250,16 @@ client clnt@Client {subscriptions, rcvQ, sndQ} Server {subscribedQ} = Right _ -> delMsgQueue ms queueId $> ok ok :: Signed - ok = mkResp queueId OK + ok = mkResp corrId queueId OK err :: ErrorType -> Signed - err = mkResp queueId . ERR + err = mkResp corrId queueId . ERR okResp :: Either ErrorType () -> Signed okResp = either err $ const ok - msgResp :: RecipientId -> Message -> Signed - msgResp rId Message {msgId, ts, msgBody} = mkResp rId $ MSG msgId ts msgBody + msgResp :: CorrelationId -> RecipientId -> Message -> Signed + msgResp _corrId rId Message {msgId, ts, msgBody} = mkResp _corrId rId $ MSG msgId ts msgBody randomId :: (MonadUnliftIO m, MonadReader Env m) => Int -> m Encoded randomId n = do diff --git a/src/Simplex/Messaging/Server/Transmission.hs b/src/Simplex/Messaging/Server/Transmission.hs index 11e5b250f..9287e9254 100644 --- a/src/Simplex/Messaging/Server/Transmission.hs +++ b/src/Simplex/Messaging/Server/Transmission.hs @@ -33,15 +33,15 @@ data Cmd where deriving instance Show Cmd -type Signed = (QueueId, Cmd) +type Signed = (CorrelationId, QueueId, Cmd) type Transmission = (Signature, Signed) -type SignedOrError = (QueueId, Either ErrorType Cmd) +type SignedOrError = (CorrelationId, QueueId, Either ErrorType Cmd) type TransmissionOrError = (Signature, SignedOrError) -type RawTransmission = (ByteString, ByteString, ByteString) +type RawTransmission = (ByteString, ByteString, ByteString, ByteString) data Command (a :: Party) where NEW :: RecipientKey -> Command Recipient @@ -137,6 +137,8 @@ serializeCommand = \case type Encoded = ByteString +type CorrelationId = ByteString + type PublicKey = Encoded type Signature = Encoded diff --git a/src/Simplex/Messaging/Transport.hs b/src/Simplex/Messaging/Transport.hs index c0a8d2a44..193ec99d4 100644 --- a/src/Simplex/Messaging/Transport.hs +++ b/src/Simplex/Messaging/Transport.hs @@ -80,20 +80,22 @@ getBytes :: MonadIO m => Handle -> Int -> m ByteString getBytes h = liftIO . B.hGet h tPutRaw :: MonadIO m => Handle -> RawTransmission -> m () -tPutRaw h (signature, queueId, command) = do - putLn h (encode signature) - putLn h (encode queueId) +tPutRaw h (signature, corrId, queueId, command) = do + putLn h signature + putLn h corrId + putLn h queueId putLn h command -tGetRaw :: MonadIO m => Handle -> m (Either String RawTransmission) +tGetRaw :: MonadIO m => Handle -> m RawTransmission tGetRaw h = do - signature <- decode <$> getLn h - queueId <- decode <$> getLn h + signature <- getLn h + corrId <- getLn h + queueId <- getLn h command <- getLn h - return $ liftM2 (,,command) signature queueId + return (signature, corrId, queueId, command) tPut :: MonadIO m => Handle -> Transmission -> m () -tPut h (signature, (queueId, command)) = tPutRaw h (signature, queueId, serializeCommand command) +tPut h (signature, (corrId, queueId, command)) = tPutRaw h (encode signature, corrId, encode queueId, serializeCommand command) fromClient :: Cmd -> Either ErrorType Cmd fromClient = \case @@ -108,19 +110,22 @@ fromServer = \case -- | get client and server transmissions -- `fromParty` is used to limit allowed senders - `fromClient` or `fromServer` should be used tGet :: forall m. MonadIO m => (Cmd -> Either ErrorType Cmd) -> Handle -> m TransmissionOrError -tGet fromParty h = tGetRaw h >>= either (const tError) tParseLoadBody +tGet fromParty h = do + (signature, corrId, queueId, command) <- tGetRaw h + let decodedTransmission = liftM2 (,corrId,,command) (decode signature) (decode queueId) + either (const $ tError corrId) tParseLoadBody decodedTransmission where - tError :: m TransmissionOrError - tError = return (B.empty, (B.empty, Left $ SYNTAX errBadTransmission)) + tError :: ByteString -> m TransmissionOrError + tError corrId = return (B.empty, (corrId, B.empty, Left $ SYNTAX errBadTransmission)) tParseLoadBody :: RawTransmission -> m TransmissionOrError - tParseLoadBody t@(signature, queueId, command) = do + tParseLoadBody t@(signature, corrId, queueId, command) = do let cmd = parseCommand command >>= fromParty >>= tCredentials t fullCmd <- either (return . Left) cmdWithMsgBody cmd - return (signature, (queueId, fullCmd)) + return (signature, (corrId, queueId, fullCmd)) tCredentials :: RawTransmission -> Cmd -> Either ErrorType Cmd - tCredentials (signature, queueId, _) cmd = case cmd of + tCredentials (signature, _, queueId, _) cmd = case cmd of -- IDS response should not have queue ID Cmd SBroker (IDS _ _) -> Right cmd -- ERROR response does not always have queue ID diff --git a/tests/SMPClient.hs b/tests/SMPClient.hs index 9a122e035..fccfce8fb 100644 --- a/tests/SMPClient.hs +++ b/tests/SMPClient.hs @@ -57,7 +57,7 @@ smpServerTest :: [RawTransmission] -> IO [RawTransmission] smpServerTest commands = runSmpTest \h -> mapM (sendReceive h) commands where sendReceive :: Handle -> RawTransmission -> IO RawTransmission - sendReceive h t = tPutRaw h t >> either (error "bad transmission") id <$> tGetRaw h + sendReceive h t = tPutRaw h t >> tGetRaw h smpTest :: (Handle -> IO ()) -> Expectation smpTest test' = runSmpTest test' `shouldReturn` () diff --git a/tests/Test.hs b/tests/Test.hs index 1bdd88077..4d5f43732 100644 --- a/tests/Test.hs +++ b/tests/Test.hs @@ -5,6 +5,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} +import Crypto.Random import Data.ByteString.Base64 import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B @@ -28,10 +29,16 @@ main = hspec do describe "switch subscription to another SMP queue" testSwitchSub pattern Resp :: QueueId -> Command 'Broker -> TransmissionOrError -pattern Resp queueId command = ("", (queueId, Right (Cmd SBroker command))) +pattern Resp queueId command <- ("", (_, queueId, Right (Cmd SBroker command))) -sendRecv :: Handle -> RawTransmission -> IO TransmissionOrError -sendRecv h (sgn, qId, cmd) = tPutRaw h (fromRight "" $ decode sgn, qId, cmd) >> tGet fromServer h +sendRecv :: Handle -> (ByteString, ByteString, ByteString) -> IO TransmissionOrError +sendRecv h (sgn, qId, cmd) = do + corrId <- encode <$> getRandomBytes 3 + tPutRaw h (sgn, corrId, encode qId, cmd) + t@(_, (corrId', _, _)) <- tGet fromServer h + if corrId == corrId' + then return t + else error "response correlation ID does not match request" (>#>) :: [RawTransmission] -> [RawTransmission] -> Expectation commands >#> responses = smpServerTest commands `shouldReturn` responses @@ -238,36 +245,36 @@ testSwitchSub = syntaxTests :: SpecWith () syntaxTests = do - it "unknown command" $ [("", "1234", "HELLO")] >#> [("", "1234", "ERR UNKNOWN")] + it "unknown command" $ [("", "abcd", "1234", "HELLO")] >#> [("", "abcd", "1234", "ERR UNKNOWN")] describe "NEW" do - it "no parameters" $ [("", "", "NEW")] >#> [("", "", "ERR SYNTAX 2")] - it "many parameters" $ [("", "", "NEW 1 2")] >#> [("", "", "ERR SYNTAX 2")] - it "has signature" $ [("1234", "", "NEW 1234")] >#> [("", "", "ERR SYNTAX 4")] - it "queue ID" $ [("", "1", "NEW 1234")] >#> [("", "1", "ERR SYNTAX 4")] + it "no parameters" $ [("", "bcda", "", "NEW")] >#> [("", "bcda", "", "ERR SYNTAX 2")] + it "many parameters" $ [("", "cdab", "", "NEW 1 2")] >#> [("", "cdab", "", "ERR SYNTAX 2")] + it "has signature" $ [("1234", "dabc", "", "NEW 1234")] >#> [("", "dabc", "", "ERR SYNTAX 4")] + it "queue ID" $ [("", "abcd", "12345678", "NEW 1234")] >#> [("", "abcd", "12345678", "ERR SYNTAX 4")] describe "KEY" do - it "valid syntax" $ [("1234", "1", "KEY 4567")] >#> [("", "1", "ERR AUTH")] - it "no parameters" $ [("1234", "1", "KEY")] >#> [("", "1", "ERR SYNTAX 2")] - it "many parameters" $ [("1234", "1", "KEY 1 2")] >#> [("", "1", "ERR SYNTAX 2")] - it "no signature" $ [("", "1", "KEY 4567")] >#> [("", "1", "ERR SYNTAX 3")] - it "no queue ID" $ [("1234", "", "KEY 4567")] >#> [("", "", "ERR SYNTAX 3")] + it "valid syntax" $ [("1234", "bcda", "12345678", "KEY 4567")] >#> [("", "bcda", "12345678", "ERR AUTH")] + it "no parameters" $ [("1234", "cdab", "12345678", "KEY")] >#> [("", "cdab", "12345678", "ERR SYNTAX 2")] + it "many parameters" $ [("1234", "dabc", "12345678", "KEY 1 2")] >#> [("", "dabc", "12345678", "ERR SYNTAX 2")] + it "no signature" $ [("", "abcd", "12345678", "KEY 4567")] >#> [("", "abcd", "12345678", "ERR SYNTAX 3")] + it "no queue ID" $ [("1234", "bcda", "", "KEY 4567")] >#> [("", "bcda", "", "ERR SYNTAX 3")] noParamsSyntaxTest "SUB" noParamsSyntaxTest "ACK" noParamsSyntaxTest "OFF" noParamsSyntaxTest "DEL" describe "SEND" do - it "valid syntax 1" $ [("1234", "1", "SEND :hello")] >#> [("", "1", "ERR AUTH")] - it "valid syntax 2" $ [("1234", "1", "SEND 11\nhello there\n")] >#> [("", "1", "ERR AUTH")] - it "no parameters" $ [("1234", "1", "SEND")] >#> [("", "1", "ERR SYNTAX 2")] - it "no queue ID" $ [("1234", "", "SEND :hello")] >#> [("", "", "ERR SYNTAX 5")] - it "bad message body 1" $ [("1234", "1", "SEND 11 hello")] >#> [("", "1", "ERR SYNTAX 6")] - it "bad message body 2" $ [("1234", "1", "SEND hello")] >#> [("", "1", "ERR SYNTAX 6")] - it "bigger body" $ [("1234", "1", "SEND 4\nhello\n")] >#> [("", "1", "ERR SIZE")] + it "valid syntax 1" $ [("1234", "cdab", "12345678", "SEND :hello")] >#> [("", "cdab", "12345678", "ERR AUTH")] + it "valid syntax 2" $ [("1234", "dabc", "12345678", "SEND 11\nhello there\n")] >#> [("", "dabc", "12345678", "ERR AUTH")] + it "no parameters" $ [("1234", "abcd", "12345678", "SEND")] >#> [("", "abcd", "12345678", "ERR SYNTAX 2")] + it "no queue ID" $ [("1234", "bcda", "", "SEND :hello")] >#> [("", "bcda", "", "ERR SYNTAX 5")] + it "bad message body 1" $ [("1234", "cdab", "12345678", "SEND 11 hello")] >#> [("", "cdab", "12345678", "ERR SYNTAX 6")] + it "bad message body 2" $ [("1234", "dabc", "12345678", "SEND hello")] >#> [("", "dabc", "12345678", "ERR SYNTAX 6")] + it "bigger body" $ [("1234", "abcd", "12345678", "SEND 4\nhello\n")] >#> [("", "abcd", "12345678", "ERR SIZE")] describe "broker response not allowed" do - it "OK" $ [("1234", "1", "OK")] >#> [("", "1", "ERR PROHIBITED")] + it "OK" $ [("1234", "bcda", "12345678", "OK")] >#> [("", "bcda", "12345678", "ERR PROHIBITED")] where noParamsSyntaxTest :: ByteString -> SpecWith () noParamsSyntaxTest cmd = describe (B.unpack cmd) do - it "valid syntax" $ [("1234", "1", cmd)] >#> [("", "1", "ERR AUTH")] - it "parameters" $ [("1234", "1", cmd <> " 1")] >#> [("", "1", "ERR SYNTAX 2")] - it "no signature" $ [("", "1", cmd)] >#> [("", "1", "ERR SYNTAX 3")] - it "no queue ID" $ [("1234", "", cmd)] >#> [("", "", "ERR SYNTAX 3")] + it "valid syntax" $ [("1234", "abcd", "12345678", cmd)] >#> [("", "abcd", "12345678", "ERR AUTH")] + it "parameters" $ [("1234", "bcda", "12345678", cmd <> " 1")] >#> [("", "bcda", "12345678", "ERR SYNTAX 2")] + it "no signature" $ [("", "cdab", "12345678", cmd)] >#> [("", "cdab", "12345678", "ERR SYNTAX 3")] + it "no queue ID" $ [("1234", "dabc", "", cmd)] >#> [("", "dabc", "", "ERR SYNTAX 3")]