dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-09-01 20:18:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

agent: add server msg IDs to agent logs, disable 2 tests (#934)

Browse Source 
* agent: race when using agent via TCP

* remove logs

* enable tests

* comments

* use different databases for different clients

* enable all tests
This commit is contained in:
Evgeny Poberezkin
2023-12-26 17:50:39 +00:00
committed by GitHub
parent 7ddeca50e4
commit 577e3cf14d
5 changed files with 63 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/smp-agent/Main.hs
+1
View File
@@ -35,6 +35,7 @@ servers =
logCfg :: LogConfig
logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
-- Warning: this SMP agent server is experimental - it does not work correctly with multiple connected TCP clients in some cases.
main :: IO ()
main = do
putStrLn $ "SMP agent listening on port " ++ tcpPort (cfg :: AgentConfig)
src/Simplex/Messaging/Agent.hs
+37 -37
View File
@@ -1930,9 +1930,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
let e2eDh = C.dh' e2ePubKey e2ePrivKey
decryptClientMessage e2eDh clientMsg >>= \case
(SMP.PHConfirmation senderKey, AgentConfirmation {e2eEncryption_, encConnInfo, agentVersion}) ->
smpConfirmation conn senderKey e2ePubKey e2eEncryption_ encConnInfo phVer agentVersion >> ack
smpConfirmation srvMsgId conn senderKey e2ePubKey e2eEncryption_ encConnInfo phVer agentVersion >> ack
(SMP.PHEmpty, AgentInvitation {connReq, connInfo}) ->
smpInvitation conn connReq connInfo >> ack
smpInvitation srvMsgId conn connReq connInfo >> ack
_ -> prohibited >> ack
(Just e2eDh, Nothing) -> do
decryptClientMessage e2eDh clientMsg >>= \case
@@ -1961,20 +1961,20 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
Right (Just (msgId, msgMeta, aMessage, rcPrev)) -> do
conn'' <- resetRatchetSync
case aMessage of
HELLO -> helloMsg conn'' >> ackDel msgId
REPLY cReq -> replyMsg conn'' cReq >> ackDel msgId
HELLO -> helloMsg srvMsgId conn'' >> ackDel msgId
REPLY cReq -> replyMsg srvMsgId conn'' cReq >> ackDel msgId
-- note that there is no ACK sent for A_MSG, it is sent with agent's user ACK command
A_MSG body -> do
logServer "<--" c srv rId "MSG <MSG>"
logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret srvMsgId
notify $ MSG msgMeta msgFlags body
A_RCVD rcpts -> qDuplex conn'' "RCVD" $ messagesRcvd rcpts msgMeta
QCONT addr -> qDuplexAckDel conn'' "QCONT" $ continueSending addr
QADD qs -> qDuplexAckDel conn'' "QADD" $ qAddMsg qs
QKEY qs -> qDuplexAckDel conn'' "QKEY" $ qKeyMsg qs
QUSE qs -> qDuplexAckDel conn'' "QUSE" $ qUseMsg qs
QCONT addr -> qDuplexAckDel conn'' "QCONT" $ continueSending srvMsgId addr
QADD qs -> qDuplexAckDel conn'' "QADD" $ qAddMsg srvMsgId qs
QKEY qs -> qDuplexAckDel conn'' "QKEY" $ qKeyMsg srvMsgId qs
QUSE qs -> qDuplexAckDel conn'' "QUSE" $ qUseMsg srvMsgId qs
-- no action needed for QTEST
-- any message in the new queue will mark it active and trigger deletion of the old queue
QTEST _ -> logServer "<--" c srv rId "MSG <QTEST>" >> ackDel msgId
QTEST _ -> logServer "<--" c srv rId ("MSG <QTEST>:" <> logSecret srvMsgId) >> ackDel msgId
EREADY _ -> qDuplexAckDel conn'' "EREADY" $ ereadyMsg rcPrev
where
qDuplexAckDel :: Connection c -> String -> (Connection 'CDuplex -> m ()) -> m ()
@@ -1996,7 +1996,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
| otherwise -> do
liftEither (parse smpP (AGENT A_MESSAGE) agentMsgBody) >>= \case
AgentMessage _ (A_MSG body) -> do
logServer "<--" c srv rId "MSG <MSG>"
logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret srvMsgId
notify $ MSG msgMeta msgFlags body
_ -> pure ()
_ -> checkDuplicateHash e encryptedMsgHash >> ack
@@ -2102,9 +2102,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
parseMessage :: Encoding a => ByteString -> m a
parseMessage = liftEither . parse smpP (AGENT A_MESSAGE)
smpConfirmation :: Connection c -> C.APublicVerifyKey -> C.PublicKeyX25519 -> Maybe (CR.E2ERatchetParams 'C.X448) -> ByteString -> Version -> Version -> m ()
smpConfirmation conn' senderKey e2ePubKey e2eEncryption encConnInfo smpClientVersion agentVersion = do
logServer "<--" c srv rId "MSG <CONF>"
smpConfirmation :: SMP.MsgId -> Connection c -> C.APublicVerifyKey -> C.PublicKeyX25519 -> Maybe (CR.E2ERatchetParams 'C.X448) -> ByteString -> Version -> Version -> m ()
smpConfirmation srvMsgId conn' senderKey e2ePubKey e2eEncryption encConnInfo smpClientVersion agentVersion = do
logServer "<--" c srv rId $ "MSG <CONF>:" <> logSecret srvMsgId
AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
unless
(agentVersion `isCompatible` smpAgentVRange && smpClientVersion `isCompatible` smpClientVRange)
@@ -2148,9 +2148,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
_ -> prohibited
_ -> prohibited
helloMsg :: Connection c -> m ()
helloMsg conn' = do
logServer "<--" c srv rId "MSG <HELLO>"
helloMsg :: SMP.MsgId -> Connection c -> m ()
helloMsg srvMsgId conn' = do
logServer "<--" c srv rId $ "MSG <HELLO>:" <> logSecret srvMsgId
case status of
Active -> prohibited
_ ->
@@ -2170,9 +2170,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
let cData' = toConnData conn'
void $ enqueueMessage c cData' sq SMP.MsgFlags {notification = True} HELLO
replyMsg :: Connection c -> NonEmpty SMPQueueInfo -> m ()
replyMsg conn' smpQueues = do
logServer "<--" c srv rId "MSG <REPLY>"
replyMsg :: SMP.MsgId -> Connection c -> NonEmpty SMPQueueInfo -> m ()
replyMsg srvMsgId conn' smpQueues = do
logServer "<--" c srv rId $ "MSG <REPLY>:" <> logSecret srvMsgId
case duplexHandshake of
Just True -> prohibited
_ -> case conn' of
@@ -2182,11 +2182,11 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
connectReplyQueues c cData' ownConnInfo smpQueues `catchAgentError` (notify . ERR)
_ -> prohibited
continueSending :: (SMPServer, SMP.SenderId) -> Connection 'CDuplex -> m ()
continueSending addr (DuplexConnection _ _ sqs) =
continueSending :: SMP.MsgId -> (SMPServer, SMP.SenderId) -> Connection 'CDuplex -> m ()
continueSending srvMsgId addr (DuplexConnection _ _ sqs) =
case findQ addr sqs of
Just sq -> do
logServer "<--" c srv rId "MSG <QCONT>"
logServer "<--" c srv rId $ "MSG <QCONT>:" <> logSecret srvMsgId
atomically $
TM.lookup (qAddress sq) (smpDeliveryWorkers c)
>>= mapM_ (\(_, retryLock) -> tryPutTMVar retryLock ())
@@ -2194,7 +2194,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
messagesRcvd :: NonEmpty AMessageReceipt -> MsgMeta -> Connection 'CDuplex -> m ()
messagesRcvd rcpts msgMeta@MsgMeta {broker = (srvMsgId, _)} _ = do
logServer "<--" c srv rId "MSG <RCPT>"
logServer "<--" c srv rId $ "MSG <RCPT>:" <> logSecret srvMsgId
rs <- forM rcpts $ \rcpt -> clientReceipt rcpt `catchAgentError` \e -> notify (ERR e) $> Nothing
case L.nonEmpty . catMaybes $ L.toList rs of
Just rs' -> notify $ RCVD msgMeta rs' -- client must ACK once processed
@@ -2215,9 +2215,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
pure $ Just rcpt
-- processed by queue sender
qAddMsg :: NonEmpty (SMPQueueUri, Maybe SndQAddr) -> Connection 'CDuplex -> m ()
qAddMsg ((_, Nothing) :| _) _ = qError "adding queue without switching is not supported"
qAddMsg ((qUri, Just addr) :| _) (DuplexConnection cData' rqs sqs) = do
qAddMsg :: SMP.MsgId -> NonEmpty (SMPQueueUri, Maybe SndQAddr) -> Connection 'CDuplex -> m ()
qAddMsg _ ((_, Nothing) :| _) _ = qError "adding queue without switching is not supported"
qAddMsg srvMsgId ((qUri, Just addr) :| _) (DuplexConnection cData' rqs sqs) = do
when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
clientVRange <- asks $ smpClientVRange . config
case qUri `compatibleVersion` clientVRange of
@@ -2235,7 +2235,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
sq2 <- withStore c $ \db -> addConnSndQueue db connId sq''
case (sndPublicKey, e2ePubKey) of
(Just sndPubKey, Just dhPublicKey) -> do
logServer "<--" c srv rId $ "MSG <QADD> " <> logSecret (senderId queueAddress)
logServer "<--" c srv rId $ "MSG <QADD>:" <> logSecret srvMsgId <> " " <> logSecret (senderId queueAddress)
let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}}
void . enqueueMessages c cData' sqs SMP.noMsgFlags $ QKEY [(sqInfo', sndPubKey)]
sq1 <- withStore' c $ \db -> setSndSwitchStatus db sq $ Just SSSendingQKEY
@@ -2248,8 +2248,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
_ -> throwError $ AGENT A_VERSION
-- processed by queue recipient
qKeyMsg :: NonEmpty (SMPQueueInfo, SndPublicVerifyKey) -> Connection 'CDuplex -> m ()
qKeyMsg ((qInfo, senderKey) :| _) conn'@(DuplexConnection cData' rqs _) = do
qKeyMsg :: SMP.MsgId -> NonEmpty (SMPQueueInfo, SndPublicVerifyKey) -> Connection 'CDuplex -> m ()
qKeyMsg srvMsgId ((qInfo, senderKey) :| _) conn'@(DuplexConnection cData' rqs _) = do
when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
clientVRange <- asks $ smpClientVRange . config
unless (qInfo `isCompatible` clientVRange) . throwError $ AGENT A_VERSION
@@ -2257,7 +2257,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
Just rq'@RcvQueue {rcvId, e2ePrivKey = dhPrivKey, smpClientVersion = cVer, status = status'}
| status' == New || status' == Confirmed -> do
checkRQSwchStatus rq RSSendingQADD
logServer "<--" c srv rId $ "MSG <QKEY> " <> logSecret senderId
logServer "<--" c srv rId $ "MSG <QKEY>:" <> logSecret srvMsgId <> " " <> logSecret senderId
let dhSecret = C.dh' dhPublicKey dhPrivKey
withStore' c $ \db -> setRcvQueueConfirmedE2E db rq' dhSecret $ min cVer cVer'
enqueueCommand c "" connId (Just smpServer) $ AInternalCommand $ ICQSecure rcvId senderKey
@@ -2269,16 +2269,16 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
-- processed by queue sender
-- mark queue as Secured and to start sending messages to it
qUseMsg :: NonEmpty ((SMPServer, SMP.SenderId), Bool) -> Connection 'CDuplex -> m ()
qUseMsg :: SMP.MsgId -> NonEmpty ((SMPServer, SMP.SenderId), Bool) -> Connection 'CDuplex -> m ()
-- NOTE: does not yet support the change of the primary status during the rotation
qUseMsg ((addr, _primary) :| _) (DuplexConnection cData' rqs sqs) = do
qUseMsg srvMsgId ((addr, _primary) :| _) (DuplexConnection cData' rqs sqs) = do
when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
case findQ addr sqs of
Just sq'@SndQueue {dbReplaceQueueId = Just replaceQId} -> do
case find ((replaceQId ==) . dbQId) sqs of
Just sq1 -> do
checkSQSwchStatus sq1 SSSendingQKEY
logServer "<--" c srv rId $ "MSG <QUSE> " <> logSecret (snd addr)
logServer "<--" c srv rId $ "MSG <QUSE>:" <> logSecret srvMsgId <> " " <> logSecret (snd addr)
withStore' c $ \db -> setSndQueueStatus db sq' Secured
let sq'' = (sq' :: SndQueue) {status = Secured}
-- sending QTEST to the new queue only, the old one will be removed if sent successfully
@@ -2300,9 +2300,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
when (isNothing rcSnd) . void $
enqueueMessages' c cData' sqs SMP.MsgFlags {notification = True} (EREADY lastExternalSndId)
smpInvitation :: Connection c -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> m ()
smpInvitation conn' connReq@(CRInvitationUri crData _) cInfo = do
logServer "<--" c srv rId "MSG <KEY>"
smpInvitation :: SMP.MsgId -> Connection c -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> m ()
smpInvitation srvMsgId conn' connReq@(CRInvitationUri crData _) cInfo = do
logServer "<--" c srv rId $ "MSG <KEY>:" <> logSecret srvMsgId
case conn' of
ContactConnection {} -> do
g <- asks random
src/Simplex/Messaging/Agent/Client.hs
+1 -1
View File
@@ -1085,7 +1085,7 @@ disableQueuesNtfs = sendTSessionBatches "NDEL" 90 id $ sendBatch disableSMPQueue
sendAck :: AgentMonad m => AgentClient -> RcvQueue -> MsgId -> m ()
sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = do
withSMPClient c rq "ACK" $ \smp ->
withSMPClient c rq ("ACK:" <> logSecret msgId) $ \smp ->
ackSMPMessage smp rcvPrivateKey rcvId msgId
atomically $ releaseGetLock c rq
tests/AgentTests.hs
+20 -15
View File
@@ -20,6 +20,7 @@ import Control.Concurrent
import Control.Monad (forM_)
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import Data.Maybe (fromJust)
import Data.Type.Equality
import GHC.Stack (withFrozenCallStack)
import Network.HTTP.Types (urlEncode)
@@ -45,9 +46,10 @@ agentTests (ATransport t) = do
describe "Migration tests" migrationTests
describe "SMP agent protocol syntax" $ syntaxTests t
describe "Establishing duplex connection (via agent protocol)" $ do
it "should connect via one server and one agent" $ do
-- These tests are disabled because the agent does not work correctly with multiple connected TCP clients
xit "should connect via one server and one agent" $ do
smpAgentTest2_1_1 $ testDuplexConnection t
it "should connect via one server and one agent (random IDs)" $ do
xit "should connect via one server and one agent (random IDs)" $ do
smpAgentTest2_1_1 $ testDuplexConnRandomIds t
it "should connect via one server and 2 agents" $ do
smpAgentTest2_2_1 $ testDuplexConnection t
@@ -137,18 +139,18 @@ correctTransmission (corrId, connId, cmdOrErr) = case cmdOrErr of
Left e -> error $ show e
-- | receive message to handle `h` and validate that it is the expected one
(<#) :: Transport c => c -> AEntityTransmission 'Agent 'AEConn -> Expectation
h <# (corrId, connId, cmd) = (h <#:) `shouldReturn` (corrId, connId, Right cmd)
(<#) :: (HasCallStack, Transport c) => c -> AEntityTransmission 'Agent 'AEConn -> Expectation
h <# (corrId, connId, cmd) = timeout 5000000 (h <#:) `shouldReturn` Just (corrId, connId, Right cmd)
(<#.) :: Transport c => c -> AEntityTransmission 'Agent 'AENone -> Expectation
h <#. (corrId, connId, cmd) = (h <#:.) `shouldReturn` (corrId, connId, Right cmd)
(<#.) :: (HasCallStack, Transport c) => c -> AEntityTransmission 'Agent 'AENone -> Expectation
h <#. (corrId, connId, cmd) = timeout 5000000 (h <#:.) `shouldReturn` Just (corrId, connId, Right cmd)
-- | receive message to handle `h` and validate it using predicate `p`
(<#=) :: Transport c => c -> (AEntityTransmission 'Agent 'AEConn -> Bool) -> Expectation
h <#= p = (h <#:) >>= (`shouldSatisfy` p . correctTransmission)
(<#=) :: (HasCallStack, Transport c) => c -> (AEntityTransmission 'Agent 'AEConn -> Bool) -> Expectation
h <#= p = timeout 5000000 (h <#:) >>= (`shouldSatisfy` p . correctTransmission . fromJust)
(<#=?) :: Transport c => c -> (ATransmission 'Agent -> Bool) -> Expectation
h <#=? p = (h <#:?) >>= (`shouldSatisfy` p . correctTransmission)
(<#=?) :: (HasCallStack, Transport c) => c -> (ATransmission 'Agent -> Bool) -> Expectation
h <#=? p = timeout 5000000 (h <#:?) >>= (`shouldSatisfy` p . correctTransmission . fromJust)
-- | test that nothing is delivered to handle `h` during 10ms
(#:#) :: Transport c => c -> String -> Expectation
@@ -162,7 +164,10 @@ h #:# err = tryGet `shouldReturn` ()
pattern Msg :: MsgBody -> ACommand 'Agent e
pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody
testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO ()
pattern Msg' :: AgentMsgId -> MsgBody -> ACommand 'Agent e
pattern Msg' aMsgId msgBody <- MSG MsgMeta {integrity = MsgOk, recipient = (aMsgId, _)} _ msgBody
testDuplexConnection :: (HasCallStack, Transport c) => TProxy c -> c -> c -> IO ()
testDuplexConnection _ alice bob = do
("1", "bob", Right (INV cReq)) <- alice #: ("1", "bob", "NEW T INV subscribe")
let cReq' = strEncode cReq
@@ -175,19 +180,19 @@ testDuplexConnection _ alice bob = do
-- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4
alice #: ("3", "bob", "SEND F :hello") #> ("3", "bob", MID 4)
alice <# ("", "bob", SENT 4)
bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
bob <#= \case ("", "alice", Msg' 4 "hello") -> True; _ -> False
bob #: ("12", "alice", "ACK 4") #> ("12", "alice", OK)
alice #: ("4", "bob", "SEND F :how are you?") #> ("4", "bob", MID 5)
alice <# ("", "bob", SENT 5)
bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False
bob <#= \case ("", "alice", Msg' 5 "how are you?") -> True; _ -> False
bob #: ("13", "alice", "ACK 5") #> ("13", "alice", OK)
bob #: ("14", "alice", "SEND F 9\nhello too") #> ("14", "alice", MID 6)
bob <# ("", "alice", SENT 6)
alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
alice <#= \case ("", "bob", Msg' 6 "hello too") -> True; _ -> False
alice #: ("3a", "bob", "ACK 6") #> ("3a", "bob", OK)
bob #: ("15", "alice", "SEND F 9\nmessage 1") #> ("15", "alice", MID 7)
bob <# ("", "alice", SENT 7)
alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False
alice <#= \case ("", "bob", Msg' 7 "message 1") -> True; _ -> False
alice #: ("4a", "bob", "ACK 7") #> ("4a", "bob", OK)
alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)
bob #: ("17", "alice", "SEND F 9\nmessage 3") #> ("17", "alice", MID 8)
tests/AgentTests/FunctionalAPITests.hs
+4 -4
View File
@@ -1878,8 +1878,8 @@ testSwitch2ConnectionsAbort1 servers = do
testCreateQueueAuth :: HasCallStack => (Maybe BasicAuth, Version) -> (Maybe BasicAuth, Version) -> IO Int
testCreateQueueAuth clnt1 clnt2 = do
a <- getClient clnt1
b <- getClient clnt2
a <- getClient clnt1 testDB
b <- getClient clnt2 testDB2
r <- runRight $ do
tryError (createConnection a 1 True SCMInvitation Nothing SMSubscribe) >>= \case
Left (SMP AUTH) -> pure 0
@@ -1900,10 +1900,10 @@ testCreateQueueAuth clnt1 clnt2 = do
disconnectAgentClient b
pure r
where
getClient (clntAuth, clntVersion) =
getClient (clntAuth, clntVersion) db =
let servers = initAgentServers {smp = userServers [ProtoServerWithAuth testSMPServer clntAuth]}
smpCfg = (defaultClientConfig :: ProtocolClientConfig) {serverVRange = mkVersionRange 4 clntVersion}
in getSMPAgentClient' agentCfg {smpCfg} servers testDB
in getSMPAgentClient' agentCfg {smpCfg} servers db
testSMPServerConnectionTest :: ATransport -> Maybe BasicAuth -> SMPServerWithAuth -> IO (Maybe ProtocolTestFailure)
testSMPServerConnectionTest t newQueueBasicAuth srv =