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

JOIN command (#4)

Browse Source 
* parse JOIN command

* parse JOIN command

* serialize NEW command

* parse and serialize CON response

* process JOIN command: SEND confirmation (WIP)

* process response to the confirmation from JOIN command

* remove comment
This commit is contained in:
Evgeny Poberezkin
2021-01-11 18:31:10 +00:00
committed by GitHub
parent 75570350a8
commit 253f4e39c9
4 changed files with 130 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Agent.hs
+61 -4
View File
@@ -24,7 +24,7 @@ import Simplex.Messaging.Agent.Store.SQLite
import Simplex.Messaging.Agent.Store.Types
import Simplex.Messaging.Agent.Transmission
import Simplex.Messaging.Server (randomBytes)
import Simplex.Messaging.Server.Transmission (Cmd (..), CorrId (..), SParty (..))
import Simplex.Messaging.Server.Transmission (Cmd (..), CorrId (..), PublicKey, SParty (..))
import qualified Simplex.Messaging.Server.Transmission as SMP
import Simplex.Messaging.Transport
import UnliftIO.Async
@@ -89,11 +89,16 @@ processCommand ::
m ()
processCommand AgentClient {respQ, servers, commands} t@(_, connAlias, cmd) =
case cmd of
NEW smpServer _ -> do
NEW smpServer -> do
srv <- getSMPServer smpServer
smpT <- mkSmpNEW smpServer
atomically $ writeTBQueue (smpSndQ srv) smpT
return ()
JOIN (SMPQueueInfo smpServer senderId encKey) _ -> do
srv <- getSMPServer smpServer
smpT <- mkConfSEND smpServer senderId encKey
atomically $ writeTBQueue (smpSndQ srv) smpT
return ()
_ -> throwError PROHIBITED
where
replyError :: ErrorType -> SomeException -> m a
@@ -133,6 +138,44 @@ processCommand AgentClient {respQ, servers, commands} t@(_, connAlias, cmd) =
atomically . modifyTVar commands $ M.insert smpCorrId req -- TODO check ID collision
return toSMP
mkConfSEND :: SMPServer -> SMP.SenderId -> PublicKey -> m SMP.Transmission
mkConfSEND smpServer senderId encryptKey = do
g <- asks idsDrg
smpCorrId <- atomically $ CorrId <$> randomBytes 4 g
senderKey <- atomically $ randomBytes 16 g -- TODO replace with cryptographic key pair
verifyKey <- atomically $ randomBytes 16 g -- TODO replace with cryptographic key pair
-- TODO create connection with NEW status, it will be upgraded to CONFIRMED status once SMP server replies OK to SEND
msg <- mkConfirmation encryptKey senderKey
let sndPrivateKey = senderKey
signKey = verifyKey
withStore $ \st ->
createSndConn st connAlias $
SendQueue
{ server = smpServer,
sndId = senderId,
sndPrivateKey,
encryptKey,
signKey,
-- verifyKey,
status = New,
ackMode = AckMode On
}
let toSMP = ("", (smpCorrId, senderId, Cmd SSender $ SMP.SEND msg))
req =
Request
{ fromClient = t,
toSMP,
state = ConfSENDRequestState {connAlias, smpServer, senderId, sndPrivateKey, encryptKey}
}
atomically . modifyTVar commands $ M.insert smpCorrId req -- TODO check ID collision
return toSMP
mkConfirmation :: PublicKey -> PublicKey -> m SMP.MsgBody
mkConfirmation _encKey senderKey = do
let msg = "KEY " <> senderKey <> "\r\n\r\n"
-- TODO encryption
return msg
processSmp :: forall m. (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
processSmp c@AgentClient {respQ, sndQ, commands} = forever $ do
(_, (smpCorrId, qId, cmdOrErr)) <- atomically $ readTBQueue respQ
@@ -142,6 +185,7 @@ processSmp c@AgentClient {respQ, sndQ, commands} = forever $ do
case req of -- TODO empty correlation ID is ok - it can be a message
Nothing -> atomically $ writeTBQueue sndQ ("", "", ERR $ BROKER smpErrCorrelationId)
Just r@Request {fromClient = (corrId, cAlias, _)} ->
-- TODO remove matched correlation ID
runExceptT (processResponse c r cmdOrErr) >>= \case
Left e -> atomically $ writeTBQueue sndQ (corrId, cAlias, ERR e)
Right _ -> return ()
@@ -162,8 +206,7 @@ processResponse
Right resp -> case resp of
Cmd SBroker (SMP.IDS recipientId senderId) -> case smpCmd of
Cmd SRecipient (SMP.NEW _) -> case (cmd, state) of
(NEW _ _, NEWRequestState {connAlias, smpServer, rcvPrivateKey}) -> do
-- TODO all good - process response
(NEW _, NEWRequestState {connAlias, smpServer, rcvPrivateKey}) -> do
g <- asks idsDrg
encryptKey <- atomically $ randomBytes 16 g -- TODO replace with cryptographic key pair
let decryptKey = encryptKey
@@ -183,6 +226,20 @@ processResponse
respond . INV $ SMPQueueInfo smpServer senderId encryptKey
_ -> throwError INTERNAL
_ -> throwError $ BROKER smpUnexpectedResponse
Cmd SBroker (SMP.OK) -> case smpCmd of
Cmd SSender (SMP.SEND _) -> case (cmd, state) of
(JOIN _ _, ConfSENDRequestState {connAlias}) -> do
withStore $ \st -> updateQueueStatus st connAlias SND Confirmed
respond OK
_ -> throwError INTERNAL
_ -> throwError $ BROKER smpUnexpectedResponse
Cmd SBroker (SMP.ERR e) -> case smpCmd of
Cmd SSender (SMP.SEND _) -> case (cmd, state) of
(JOIN _ _, ConfSENDRequestState {connAlias}) -> do
withStore $ \st -> deleteConn st connAlias
respond . ERR $ SMP e
_ -> throwError INTERNAL
_ -> throwError $ BROKER smpUnexpectedResponse
_ -> throwError UNSUPPORTED
where
respond :: ACommand 'Agent -> m ()
src/Simplex/Messaging/Agent/Env/SQLite.hs
+14 -5
View File
@@ -15,6 +15,7 @@ import Simplex.Messaging.Agent.ServerClient
import Simplex.Messaging.Agent.Store
import Simplex.Messaging.Agent.Store.SQLite
import Simplex.Messaging.Agent.Transmission
import Simplex.Messaging.Server.Transmission (PublicKey, SenderId)
import qualified Simplex.Messaging.Server.Transmission as SMP
import UnliftIO.STM
@@ -47,11 +48,19 @@ data Request = Request
state :: RequestState
}
data RequestState = NEWRequestState
{ connAlias :: ConnAlias,
smpServer :: SMPServer,
rcvPrivateKey :: PrivateKey
}
data RequestState
= NEWRequestState
{ connAlias :: ConnAlias,
smpServer :: SMPServer,
rcvPrivateKey :: PrivateKey
}
| ConfSENDRequestState
{ connAlias :: ConnAlias,
smpServer :: SMPServer,
senderId :: SenderId,
sndPrivateKey :: PrivateKey,
encryptKey :: PublicKey
}
newAgentClient :: Natural -> STM AgentClient
newAgentClient qSize = do
src/Simplex/Messaging/Agent/Store.hs
+1
View File
@@ -36,6 +36,7 @@ data SendQueue = SendQueue
sndPrivateKey :: PrivateKey,
encryptKey :: PublicKey,
signKey :: PrivateKey,
-- verifyKey :: Maybe PublicKey,
status :: QueueStatus,
ackMode :: AckMode -- whether acknowledgement is expected (via ReceiveQueue if present)
}
src/Simplex/Messaging/Agent/Transmission.hs
+54 -24
View File
@@ -70,22 +70,22 @@ data ACmd where
deriving instance Show ACmd
data ACommand (p :: AParty) where
NEW :: SMPServer -> AckMode -> ACommand Client
NEW :: SMPServer -> ACommand Client
INV :: SMPQueueInfo -> ACommand Agent
JOIN :: SMPQueueInfo -> Maybe SMPServer -> AckMode -> ACommand Client
JOIN :: SMPQueueInfo -> ReplyMode -> ACommand Client
CON :: ACommand Agent
CONF :: OtherPartyId -> ACommand Agent
LET :: OtherPartyId -> ACommand Client
SUB :: SubMode -> ACommand Client
END :: ACommand Agent
QST :: QueueDirection -> ACommand Client
STAT :: QueueDirection -> Maybe QueueStatus -> Maybe SubMode -> ACommand Agent
-- QST :: QueueDirection -> ACommand Client
-- STAT :: QueueDirection -> Maybe QueueStatus -> Maybe SubMode -> ACommand Agent
SEND :: MsgBody -> ACommand Client
MSG :: AgentMsgId -> UTCTime -> UTCTime -> MsgStatus -> MsgBody -> ACommand Agent
ACK :: AgentMsgId -> ACommand Client
RCVD :: AgentMsgId -> ACommand Agent
OFF :: ACommand Client
DEL :: ACommand Client
-- RCVD :: AgentMsgId -> ACommand Agent
-- OFF :: ACommand Client
-- DEL :: ACommand Client
OK :: ACommand Agent
ERR :: ErrorType -> ACommand Agent
@@ -95,8 +95,9 @@ data AMessage where
HELLO :: VerificationKey -> AckMode -> AMessage
REPLY :: SMPQueueInfo -> AMessage
A_MSG :: MsgBody -> AMessage
A_ACK :: AgentMsgId -> AckStatus -> AMessage
A_DEL :: AMessage
-- A_ACK :: AgentMsgId -> AckStatus -> AMessage
-- A_DEL :: AMessage
data SMPServer = SMPServer
{ host :: HostName,
@@ -120,6 +121,8 @@ newtype SubMode = SubMode Mode deriving (Show)
data SMPQueueInfo = SMPQueueInfo SMPServer SenderId EncryptionKey
deriving (Show)
data ReplyMode = ReplyOn SMPServer | ReplyOff deriving (Show)
type EncryptionKey = PublicKey
type VerificationKey = PublicKey
@@ -172,15 +175,32 @@ smpUnexpectedResponse = 3
parseCommand :: ByteString -> Either ErrorType ACmd
parseCommand command = case B.words command of
["NEW", srv] -> newConn srv . Right $ AckMode On
["NEW", srv, am] -> newConn srv $ ackMode am
["NEW", srv] -> newConn srv -- . Right $ AckMode On
-- ["NEW", srv, am] -> newConn srv $ ackMode am
["INV", qInfo] -> ACmd SAgent . INV <$> smpQueueInfo qInfo
"JOIN" : qInfo : ws -> joinConn qInfo ws
["CON"] -> Right . ACmd SAgent $ CON
"NEW" : _ -> errParams
"INV" : _ -> errParams
"JOIN" : _ -> errParams
"CON" : _ -> errParams
_ -> Left UNKNOWN
where
newConn :: ByteString -> Either ErrorType AckMode -> Either ErrorType ACmd
newConn srv am = ACmd SClient <$> liftM2 NEW (smpServer srv) am
newConn :: ByteString -> Either ErrorType ACmd
newConn srv = ACmd SClient . NEW <$> smpServer srv
joinConn :: ByteString -> [ByteString] -> Either ErrorType ACmd
joinConn qInfo ws = do
q <- smpQueueInfo qInfo
case ws of
[] -> let SMPQueueInfo srv _ _ = q in joinCmd q $ ReplyOn srv
["NO_REPLY"] -> joinCmd q ReplyOff
[srv] -> do
s <- smpServer srv
joinCmd q $ ReplyOn s
_ -> errParams
where
joinCmd q r = return $ ACmd SClient $ JOIN q r
smpServer :: ByteString -> Either ErrorType SMPServer
smpServer srv =
@@ -199,16 +219,16 @@ parseCommand command = case B.words command of
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
-- 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
-- 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
@@ -218,12 +238,22 @@ parseCommand command = case B.words command of
serializeCommand :: ACommand p -> ByteString
serializeCommand = \case
INV (SMPQueueInfo srv qId ek) -> "INV smp::" <> server srv <> "::" <> encode qId <> "::" <> encode ek
NEW srv -> "NEW " <> server srv
INV qInfo -> "INV " <> smpQueueInfo qInfo
JOIN qInfo rMode ->
"JOIN " <> smpQueueInfo qInfo <> " "
<> case rMode of
ReplyOff -> "NO_REPLY"
ReplyOn srv -> server srv
CON -> "CON"
c -> B.pack $ show c
where
server :: SMPServer -> ByteString
server SMPServer {host, port, keyHash} = B.pack $ host <> maybe "" (':' :) port <> maybe "" (('#' :) . B.unpack) keyHash
smpQueueInfo :: SMPQueueInfo -> ByteString
smpQueueInfo (SMPQueueInfo srv qId ek) = "smp::" <> server srv <> "::" <> encode qId <> "::" <> encode ek
tPutRaw :: MonadIO m => Handle -> ARawTransmission -> m ()
tPutRaw h (corrId, connAlias, command) = do
putLn h corrId
@@ -258,7 +288,7 @@ tGet party h = tGetRaw h >>= tParseLoadBody
tConnAlias :: ARawTransmission -> ACommand p -> Either ErrorType (ACommand p)
tConnAlias (_, connAlias, _) cmd = case cmd of
-- NEW has optional connAlias
NEW _ _ -> Right cmd
NEW _ -> Right cmd
-- ERROR response does not always have connAlias
ERR _ -> Right cmd
-- other responses must have connAlias