diff --git a/design/agent2.gv b/design/agent2.gv index e65fa5171..bf3f57744 100644 --- a/design/agent2.gv +++ b/design/agent2.gv @@ -71,12 +71,6 @@ digraph SMPAgent { sOutq [shape=rarrow label="srv send\nTBQueue"] sSock [label="SMP client connection TCP socket" color=blue] - subgraph clusterQueue { - graph [fontsize=11] - label="1 per SMP queue\n(+ 1 TBQueue\nw/out SMP qID)" - sCmdq [shape=rarrow label="commands\nTBQueue"] - } - subgraph clusterServerThreads { graph [fontsize=11] label="SMP client threads" @@ -89,7 +83,7 @@ digraph SMPAgent { {uAgent uProcess} -> sOutq -> sSnd -> sSock -> sAgent [color=blue] } - userState [label="connected\nservers,\nsubscribed\nqueues\n(STM)" shape="folder"] + userState [label="connected\nservers,\nsubscribed\nqueues,\nsent\ncommands\n(STM)" shape="folder"] {uAgent uProcess} -> userState [dir=both color="#FF8888"] {uAgent uProcess} -> connectionsStore [dir=both color="#880000"] @@ -98,8 +92,6 @@ digraph SMPAgent { connectClnt -> {uRcv uSnd} [style=dashed label="race" color=orange fontcolor=orange] runClnt -> {uAgent uProcess} [style=dashed label="race" color=orange fontcolor=orange] - sSnd -> sCmdq - sCmdq -> sAgent [style=dashed] sAgent -> uRespq -> uProcess uAgent -> runClient [style=dashed label="fork" color=orange fontcolor=orange] diff --git a/package.yaml b/package.yaml index 9d9e3082e..2b6ae0b3c 100644 --- a/package.yaml +++ b/package.yaml @@ -20,6 +20,7 @@ dependencies: - iso8601-time == 0.1.* - mtl - network == 3.1.* + - split == 0.2.* - sqlite-simple == 0.4.* - stm - template-haskell == 2.15.* diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index ab9ca8c3e..3386514a0 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -9,8 +9,10 @@ module Simplex.Messaging.Agent (runSMPAgent) where import Control.Monad.IO.Unlift import Control.Monad.Reader import Crypto.Random +import qualified Data.ByteString.Char8 as B import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Transmission +import Simplex.Messaging.Server.Transmission (CorrId (..)) import Simplex.Messaging.Transport import UnliftIO.Async import UnliftIO.IO @@ -35,23 +37,22 @@ runClient :: MonadUnliftIO m => AgentClient -> m () runClient c = race_ (respond c) (process c) receive :: MonadUnliftIO m => Handle -> AgentClient -> m () -receive h AgentClient {rcvQ, sndQ} = forever $ do - aCmdGet SClient h >>= \case - Right cmd -> atomically $ writeTBQueue rcvQ cmd - Left e -> atomically $ writeTBQueue sndQ $ ERR e +receive h AgentClient {rcvQ, sndQ} = + forever $ + tGet SClient h >>= \(corrId, cAlias, command) -> atomically $ case command of + Right cmd -> writeTBQueue rcvQ (corrId, cAlias, cmd) + Left e -> writeTBQueue sndQ (corrId, cAlias, ERR e) send :: MonadUnliftIO m => Handle -> AgentClient -> m () -send h AgentClient {sndQ} = forever $ do - cmd <- atomically $ readTBQueue sndQ - putLn h $ serializeCommand cmd +send h AgentClient {sndQ} = forever $ atomically (readTBQueue sndQ) >>= tPut h process :: MonadUnliftIO m => AgentClient -> m () process AgentClient {rcvQ, respQ} = forever $ do - cmd <- atomically (readTBQueue rcvQ) + (corrId, cAlias, cmd) <- atomically (readTBQueue rcvQ) liftIO $ print cmd atomically $ writeTBQueue respQ () respond :: MonadUnliftIO m => AgentClient -> m () respond AgentClient {respQ, sndQ} = forever . atomically $ do readTBQueue respQ - writeTBQueue sndQ $ ERR UNKNOWN + writeTBQueue sndQ (CorrId B.empty, B.empty, ERR UNKNOWN) diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 88a72030d..cbe906902 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -31,8 +31,8 @@ data Env = Env } data AgentClient = AgentClient - { rcvQ :: TBQueue (ACommand Client), - sndQ :: TBQueue (ACommand Agent), + { rcvQ :: TBQueue (ATransmission Client), + sndQ :: TBQueue (ATransmission Agent), respQ :: TBQueue (), servers :: Map (HostName, ServiceName) ServerClient } diff --git a/src/Simplex/Messaging/Agent/Transmission.hs b/src/Simplex/Messaging/Agent/Transmission.hs index c50c125b3..9a2b611dd 100644 --- a/src/Simplex/Messaging/Agent/Transmission.hs +++ b/src/Simplex/Messaging/Agent/Transmission.hs @@ -1,6 +1,8 @@ {-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneDeriving #-} @@ -9,14 +11,19 @@ module Simplex.Messaging.Agent.Transmission where +import Control.Monad import Control.Monad.IO.Class +-- import Numeric.Natural +import Data.ByteString.Base64 import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import Data.Kind +import Data.List.Split (splitOn) import Data.Time.Clock (UTCTime) +import Data.Type.Equality +import Data.Typeable () import Network.Socket --- import Numeric.Natural -import Simplex.Messaging.Server.Transmission (CorrId (..), Encoded, MsgBody, PublicKey, QueueId, errMessageBody) +import Simplex.Messaging.Server.Transmission (CorrId (..), Encoded, MsgBody, PublicKey, QueueId, errBadParameters, errMessageBody) import Simplex.Messaging.Transport import System.IO import Text.Read @@ -38,6 +45,11 @@ deriving instance Show (SAParty p) deriving instance Eq (SAParty p) +instance TestEquality SAParty where + testEquality SAgent SAgent = Just Refl + testEquality SClient SClient = Just Refl + testEquality _ _ = Nothing + data ACmd where ACmd :: SAParty p -> ACommand p -> ACmd @@ -72,9 +84,14 @@ data AMessage where A_ACK :: AgentMsgId -> AckStatus -> AMessage A_DEL :: AMessage -data SMPServer = SMPServer HostName ServiceName KeyFingerprint deriving (Show) +data SMPServer = SMPServer + { host :: HostName, + port :: Maybe ServiceName, + keyHash :: Maybe KeyHash + } + deriving (Show) -type KeyFingerprint = Encoded +type KeyHash = Encoded type ConnAlias = ByteString @@ -115,8 +132,57 @@ data AckStatus = AckOk | AckError AckErrorType data AckErrorType = AckUnknown | AckProhibited | AckSyntax Int -- etc. deriving (Show) +errBadInvitation :: Int +errBadInvitation = 10 + +errNoConnAlias :: Int +errNoConnAlias = 11 + parseCommand :: ByteString -> Either ErrorType ACmd -parseCommand _ = Left UNKNOWN +parseCommand command = case B.words command of + ["NEW", srv] -> newConn srv . Right $ AckMode On + ["NEW", srv, am] -> newConn srv $ ackMode am + ["INV", qInfo] -> ACmd SAgent . INV <$> smpQueueInfo qInfo + "NEW" : _ -> errParams + "INV" : _ -> errParams + _ -> Left UNKNOWN + where + newConn :: ByteString -> Either ErrorType AckMode -> Either ErrorType ACmd + newConn srv am = ACmd SClient <$> liftM2 NEW (smpServer srv) am + + smpServer :: ByteString -> Either ErrorType SMPServer + smpServer srv = + let (s, kf) = span (/= '#') $ B.unpack srv + (h, p) = span (/= ':') s + in SMPServer h (srvPart p) <$> traverse dec64 (srvPart kf) + + smpQueueInfo :: ByteString -> Either ErrorType SMPQueueInfo + smpQueueInfo qInfo = case splitOn "::" $ B.unpack qInfo of + ["smp", srv, qId, ek] -> liftM3 SMPQueueInfo (smpServer $ B.pack srv) (dec64 qId) (dec64 ek) + _ -> errInv + + dec64 :: String -> Either ErrorType ByteString + dec64 s = either (const errInv) Right . decode $ B.pack s + + srvPart :: String -> Maybe String + srvPart s = if length s > 1 then Just $ tail s else Nothing + + ackMode :: ByteString -> Either ErrorType AckMode + ackMode am = case B.split '=' am of + ["ACK", mode] -> AckMode <$> getMode mode + _ -> errParams + + getMode :: ByteString -> Either ErrorType Mode + getMode mode = case mode of + "ON" -> Right On + "OFF" -> Right Off + _ -> errParams + + errParams :: Either ErrorType a + errParams = Left $ SYNTAX errBadParameters + + errInv :: Either ErrorType a + errInv = Left $ SYNTAX errBadInvitation serializeCommand :: ACommand p -> ByteString serializeCommand = B.pack . show @@ -137,70 +203,45 @@ tGetRaw h = do tPut :: MonadIO m => Handle -> ATransmission p -> m () tPut h (corrId, connAlias, command) = tPutRaw h (bs corrId, connAlias, serializeCommand command) -fromClient :: ACmd -> Either ErrorType ACmd -fromClient = \case - ACmd SAgent _ -> Left PROHIBITED - cmd -> Right cmd - -fromAgent :: ACmd -> Either ErrorType ACmd -fromAgent = \case - ACmd SClient _ -> Left PROHIBITED - cmd -> Right cmd - -- | get client and agent transmissions --- tGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (ATransmissionOrError p) --- tGet fromParty h = tGetRaw h >>= tParseLoadBody --- where --- tParseLoadBody :: ARawTransmission -> m (ATransmissionOrError p) --- tParseLoadBody t@(corrId, connAlias, command) = do --- let cmd = parseCommand command >>= fromParty -- >>= tCredentials t --- fullCmd <- either (return . Left) cmdWithMsgBody cmd --- return (CorrId corrId, connAlias, fullCmd) +tGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (ATransmissionOrError p) +tGet party h = tGetRaw h >>= tParseLoadBody + where + tParseLoadBody :: ARawTransmission -> m (ATransmissionOrError p) + tParseLoadBody t@(corrId, connAlias, command) = do + let cmd = parseCommand command >>= fromParty >>= tConnAlias t + fullCmd <- either (return . Left) cmdWithMsgBody cmd + return (CorrId corrId, connAlias, fullCmd) --- fromParty :: ACmd -> Either ErrorType (ACommand p) --- fromParty (ACmd p cmd) --- | p == party = Right cmd --- | otherwise = Left PROHIBITED + fromParty :: ACmd -> Either ErrorType (ACommand p) + fromParty (ACmd (p :: p1) cmd) = case testEquality party p of + Just Refl -> Right cmd + _ -> Left PROHIBITED --- tCredentials :: RawTransmission -> Cmd -> Either ErrorType Cmd --- 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 --- Cmd SBroker (ERR _) -> Right cmd --- -- other responses must have queue ID --- Cmd SBroker _ --- | B.null queueId -> Left $ SYNTAX errNoConnectionId --- | otherwise -> Right cmd --- -- CREATE must NOT have signature or queue ID --- Cmd SRecipient (NEW _) --- | B.null signature && B.null queueId -> Right cmd --- | otherwise -> Left $ SYNTAX errHasCredentials --- -- SEND must have queue ID, signature is not always required --- Cmd SSender (SEND _) --- | B.null queueId -> Left $ SYNTAX errNoConnectionId --- | otherwise -> Right cmd --- -- other client commands must have both signature and queue ID --- Cmd SRecipient _ --- | B.null signature || B.null queueId -> Left $ SYNTAX errNoCredentials --- | otherwise -> Right cmd + tConnAlias :: ARawTransmission -> ACommand p -> Either ErrorType (ACommand p) + tConnAlias (_, connAlias, _) cmd = case cmd of + -- NEW has optional connAlias + NEW _ _ -> Right cmd + -- ERROR response does not always have connAlias + ERR _ -> Right cmd + -- other responses must have connAlias + _ + | B.null connAlias -> Left $ SYNTAX errNoConnAlias + | otherwise -> Right cmd --- cmdWithMsgBody :: ACommand p -> m (Either ErrorType (ACommand p)) --- cmdWithMsgBody = \case --- SEND body -> SEND <$$> getMsgBody body --- MSG agentMsgId srvTS agentTS status body -> MSG agentMsgId srvTS agentTS status <$$> getMsgBody body --- cmd -> return $ Right cmd + cmdWithMsgBody :: ACommand p -> m (Either ErrorType (ACommand p)) + cmdWithMsgBody = \case + SEND body -> SEND <$$> getMsgBody body + MSG agentMsgId srvTS agentTS status body -> MSG agentMsgId srvTS agentTS status <$$> getMsgBody body + cmd -> return $ Right cmd --- getMsgBody :: MsgBody -> m (Either ErrorType MsgBody) --- getMsgBody msgBody = --- case B.unpack msgBody of --- ':' : body -> return . Right $ B.pack body --- str -> case readMaybe str :: Maybe Int of --- Just size -> do --- body <- getBytes h size --- s <- getLn h --- return $ if B.null s then Right body else Left SIZE --- Nothing -> return . Left $ SYNTAX errMessageBody - -aCmdGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (Either ErrorType (ACommand p)) -aCmdGet _ h = getLn h >>= (\_ -> return $ Left UNKNOWN) + getMsgBody :: MsgBody -> m (Either ErrorType MsgBody) + getMsgBody msgBody = + case B.unpack msgBody of + ':' : body -> return . Right $ B.pack body + str -> case readMaybe str :: Maybe Int of + Just size -> do + body <- getBytes h size + s <- getLn h + return $ if B.null s then Right body else Left SIZE + Nothing -> return . Left $ SYNTAX errMessageBody diff --git a/src/Simplex/Messaging/Server/Transmission.hs b/src/Simplex/Messaging/Server/Transmission.hs index 628574cdb..74d5fdeba 100644 --- a/src/Simplex/Messaging/Server/Transmission.hs +++ b/src/Simplex/Messaging/Server/Transmission.hs @@ -145,7 +145,7 @@ serializeCommand = \case type Encoded = ByteString -- newtype to avoid accidentally changing order of transmission parts -newtype CorrId = CorrId {bs :: ByteString} deriving (Eq) +newtype CorrId = CorrId {bs :: ByteString} deriving (Eq, Show) instance IsString CorrId where fromString = CorrId . fromString @@ -182,8 +182,8 @@ errNoCredentials = 3 errHasCredentials :: Int errHasCredentials = 4 -errNoConnectionId :: Int -errNoConnectionId = 5 +errNoQueueId :: Int +errNoQueueId = 5 errMessageBody :: Int errMessageBody = 6 @@ -237,19 +237,19 @@ tGet fromParty h = do 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 + -- ERR response does not always have queue ID Cmd SBroker (ERR _) -> Right cmd -- other responses must have queue ID Cmd SBroker _ - | B.null queueId -> Left $ SYNTAX errNoConnectionId + | B.null queueId -> Left $ SYNTAX errNoQueueId | otherwise -> Right cmd - -- CREATE must NOT have signature or queue ID + -- NEW must NOT have signature or queue ID Cmd SRecipient (NEW _) | B.null signature && B.null queueId -> Right cmd | otherwise -> Left $ SYNTAX errHasCredentials -- SEND must have queue ID, signature is not always required Cmd SSender (SEND _) - | B.null queueId -> Left $ SYNTAX errNoConnectionId + | B.null queueId -> Left $ SYNTAX errNoQueueId | otherwise -> Right cmd -- other client commands must have both signature and queue ID Cmd SRecipient _