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

newtype for server entity IDs, fix TRcvQueues (#1290)

Browse Source 
* put DRG state to IORef, split STM transaction of sending notification (#1288)

* put DRG state to IORef, split STM transaction of sending notification

* remove comment

* remove comment

* add comment

* revert version

* newtype for server entity IDs, fix TRcvQueues

* Revert "put DRG state to IORef, split STM transaction of sending notification (#1288)"

This reverts commit 517933d189.

* logServer
This commit is contained in:
Evgeny
2024-08-30 12:50:02 +01:00
committed by GitHub
parent 655e7ad7d5
commit ce6777b68d
34 changed files with 280 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/FileTransfer/Agent.hs
+2 -2
View File
@@ -74,7 +74,7 @@ import qualified Simplex.Messaging.Crypto.File as CF
import qualified Simplex.Messaging.Crypto.Lazy as LC
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String (strDecode, strEncode)
import Simplex.Messaging.Protocol (EntityId, ProtocolServer, ProtocolType (..), XFTPServer)
import Simplex.Messaging.Protocol (ProtocolServer, ProtocolType (..), XFTPServer)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Util (catchAll_, liftError, tshow, unlessM, whenM)
import System.FilePath (takeFileName, (</>))
@@ -346,7 +346,7 @@ xftpDeleteRcvFiles' c rcvFileEntityIds = do
batchFiles :: (DB.Connection -> DBRcvFileId -> IO a) -> [RcvFile] -> AM' [Either AgentErrorType a]
batchFiles f rcvFiles = withStoreBatch' c $ \db -> map (\RcvFile {rcvFileId} -> f db rcvFileId) rcvFiles
notify :: forall m e. (MonadIO m, AEntityI e) => AgentClient -> EntityId -> AEvent e -> m ()
notify :: forall m e. (MonadIO m, AEntityI e) => AgentClient -> AEntityId -> AEvent e -> m ()
notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, AEvt (sAEntity @e) cmd)
xftpSendFile' :: AgentClient -> UserId -> CryptoFile -> Int -> AM SndFileId
src/Simplex/FileTransfer/Client.hs
+3 -2
View File
@@ -50,6 +50,7 @@ import Simplex.Messaging.Protocol
ProtocolServer (..),
RecipientId,
SenderId,
pattern NoEntity,
)
import Simplex.Messaging.Transport (ALPN, HandshakeError (..), THandleAuth (..), THandleParams (..), TransportError (..), TransportPeer (..), supportedParameters)
import Simplex.Messaging.Transport.Client (TransportClientConfig, TransportHost, alpn)
@@ -222,7 +223,7 @@ createXFTPChunk ::
Maybe BasicAuth ->
ExceptT XFTPClientError IO (SenderId, NonEmpty RecipientId)
createXFTPChunk c spKey file rcps auth_ =
sendXFTPCommand c spKey "" (FNEW file rcps auth_) Nothing >>= \case
sendXFTPCommand c spKey NoEntity (FNEW file rcps auth_) Nothing >>= \case
(FRSndIds sId rIds, body) -> noFile body (sId, rIds)
(r, _) -> throwE $ unexpectedResponse r
@@ -278,7 +279,7 @@ pingXFTP :: XFTPClient -> ExceptT XFTPClientError IO ()
pingXFTP c@XFTPClient {thParams} = do
t <-
liftEither . first PCETransportError $
xftpEncodeTransmission thParams ("", "", FileCmd SFRecipient PING)
xftpEncodeTransmission thParams ("", NoEntity, FileCmd SFRecipient PING)
(r, _) <- sendXFTPTransmission c t Nothing
case r of
FRPong -> pure ()
src/Simplex/FileTransfer/Description.hs
+4 -9
View File
@@ -1,7 +1,9 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
@@ -139,12 +141,9 @@ data FileChunkReplica = FileChunkReplica
}
deriving (Eq, Show)
newtype ChunkReplicaId = ChunkReplicaId {unChunkReplicaId :: ByteString}
newtype ChunkReplicaId = ChunkReplicaId {unChunkReplicaId :: XFTPFileId}
deriving (Eq, Show)
instance StrEncoding ChunkReplicaId where
strEncode (ChunkReplicaId fid) = strEncode fid
strP = ChunkReplicaId <$> strP
deriving newtype (StrEncoding)
instance FromJSON ChunkReplicaId where
parseJSON = strParseJSON "ChunkReplicaId"
@@ -153,10 +152,6 @@ instance ToJSON ChunkReplicaId where
toJSON = strToJSON
toEncoding = strToJEncoding
instance FromField ChunkReplicaId where fromField f = ChunkReplicaId <$> fromField f
instance ToField ChunkReplicaId where toField (ChunkReplicaId s) = toField s
data YAMLFileDescription = YAMLFileDescription
{ party :: FileParty,
size :: String,
src/Simplex/FileTransfer/Protocol.hs
+4 -3
View File
@@ -41,6 +41,7 @@ import Simplex.Messaging.Protocol
ProtocolType (..),
RcvPublicAuthKey,
RcvPublicDhKey,
EntityId (..),
RecipientId,
SenderId,
SentRawTransmission,
@@ -170,7 +171,7 @@ data FileInfo = FileInfo
}
deriving (Show)
type XFTPFileId = ByteString
type XFTPFileId = EntityId
instance FilePartyI p => ProtocolEncoding XFTPVersion XFTPErrorType (FileCommand p) where
type Tag (FileCommand p) = FileCommandTag p
@@ -191,7 +192,7 @@ instance FilePartyI p => ProtocolEncoding XFTPVersion XFTPErrorType (FileCommand
fromProtocolError = fromProtocolError @XFTPVersion @XFTPErrorType @FileResponse
{-# INLINE fromProtocolError #-}
checkCredentials (auth, _, fileId, _) cmd = case cmd of
checkCredentials (auth, _, EntityId fileId, _) cmd = case cmd of
-- FNEW must not have signature and chunk ID
FNEW {}
| isNothing auth -> Left $ CMD NO_AUTH
@@ -301,7 +302,7 @@ instance ProtocolEncoding XFTPVersion XFTPErrorType FileResponse where
PEBlock -> BLOCK
{-# INLINE fromProtocolError #-}
checkCredentials (_, _, entId, _) cmd = case cmd of
checkCredentials (_, _, EntityId entId, _) cmd = case cmd of
FRSndIds {} -> noEntity
-- ERR response does not always have entity ID
FRErr _ -> Right cmd
src/Simplex/FileTransfer/Server.hs
+6 -7
View File
@@ -9,6 +9,7 @@
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
@@ -53,7 +54,7 @@ import qualified Simplex.Messaging.Crypto as C
import qualified Simplex.Messaging.Crypto.Lazy as LC
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (CorrId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TransmissionAuth)
import Simplex.Messaging.Protocol (CorrId (..), EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TransmissionAuth, pattern NoEntity)
import Simplex.Messaging.Server (dummyVerifyCmd, verifyCmdAuthorization)
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.Stats
@@ -310,7 +311,7 @@ data ServerFile = ServerFile
processRequest :: XFTPTransportRequest -> M ()
processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHead}, sendResponse}
| B.length bodyHead /= xftpBlockSize = sendXFTPResponse ("", "", FRErr BLOCK) Nothing
| B.length bodyHead /= xftpBlockSize = sendXFTPResponse ("", NoEntity, FRErr BLOCK) Nothing
| otherwise = do
case xftpDecodeTransmission thParams bodyHead of
Right (sig_, signed, (corrId, fId, cmdOrErr)) ->
@@ -323,7 +324,7 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea
Left e -> send (FRErr e) Nothing
where
send resp = sendXFTPResponse (corrId, fId, resp)
Left e -> sendXFTPResponse ("", "", FRErr e) Nothing
Left e -> sendXFTPResponse ("", NoEntity, FRErr e) Nothing
where
sendXFTPResponse (corrId, fId, resp) serverFile_ = do
let t_ = xftpEncodeTransmission thParams (corrId, fId, resp)
@@ -464,7 +465,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
\used -> let used' = used + fromIntegral size in if used' <= quota then (True, used') else (False, used)
receive = do
path <- asks $ filesPath . config
let fPath = path </> B.unpack (B64.encode senderId)
let fPath = path </> B.unpack (B64.encode $ unEntityId senderId)
receiveChunk (XFTPRcvChunkSpec fPath size digest) >>= \case
Right () -> do
stats <- asks serverStats
@@ -560,9 +561,7 @@ randomId :: Int -> M ByteString
randomId n = atomically . C.randomBytes n =<< asks random
getFileId :: M XFTPFileId
getFileId = do
size <- asks (fileIdSize . config)
atomically . C.randomBytes size =<< asks random
getFileId = fmap EntityId . randomId =<< asks (fileIdSize . config)
withFileLog :: (StoreLog 'WriteMode -> IO a) -> M ()
withFileLog action = liftIO . mapM_ action =<< asks storeLog
src/Simplex/FileTransfer/Server/Control.hs
+2 -2
View File
@@ -4,7 +4,7 @@
module Simplex.FileTransfer.Server.Control where
import qualified Data.Attoparsec.ByteString.Char8 as A
import Data.ByteString (ByteString)
import Simplex.FileTransfer.Protocol (XFTPFileId)
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (BasicAuth)
@@ -13,7 +13,7 @@ data CPClientRole = CPRNone | CPRUser | CPRAdmin
data ControlProtocol
= CPAuth BasicAuth
| CPStatsRTS
| CPDelete ByteString
| CPDelete XFTPFileId
| CPHelp
| CPQuit
| CPSkip
src/Simplex/FileTransfer/Types.hs
+2 -2
View File
@@ -25,9 +25,9 @@ import Simplex.Messaging.Parsers
import Simplex.Messaging.Protocol (XFTPServer)
import System.FilePath ((</>))
type RcvFileId = ByteString
type RcvFileId = ByteString -- Agent entity ID
type SndFileId = ByteString
type SndFileId = ByteString -- Agent entity ID
authTagSize :: Int64
authTagSize = fromIntegral C.authTagSize
src/Simplex/Messaging/Agent.hs
+14 -14
View File
@@ -177,7 +177,7 @@ import Simplex.Messaging.Notifications.Protocol (DeviceToken, NtfRegCode (NtfReg
import Simplex.Messaging.Notifications.Server.Push.APNS (PNMessageData (..))
import Simplex.Messaging.Notifications.Types
import Simplex.Messaging.Parsers (parse)
import Simplex.Messaging.Protocol (BrokerMsg, Cmd (..), EntityId, ErrorType (AUTH), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth, ProtocolType (..), ProtocolTypeI (..), SMPMsgMeta, SParty (..), SProtocolType (..), SndPublicAuthKey, SubscriptionMode (..), UserProtocol, VersionSMPC, sndAuthKeySMPClientVersion)
import Simplex.Messaging.Protocol (BrokerMsg, Cmd (..), ErrorType (AUTH), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth, ProtocolType (..), ProtocolTypeI (..), SMPMsgMeta, SParty (..), SProtocolType (..), SndPublicAuthKey, SubscriptionMode (..), UserProtocol, VersionSMPC, sndAuthKeySMPClientVersion)
import qualified Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
import qualified Simplex.Messaging.TMap as TM
@@ -891,7 +891,7 @@ joinConnSrv c userId connId hasNewConn enableNtfs cReqUri@CRContactUri {} cInfo
lift (compatibleContactUri cReqUri) >>= \case
Just (qInfo, vrsn) -> do
(connId', cReq) <- newConnSrv c userId connId hasNewConn enableNtfs SCMInvitation Nothing (CR.IKNoPQ pqSup) subMode srv
void $ sendInvitation c userId qInfo vrsn cReq cInfo
void $ sendInvitation c userId connId' qInfo vrsn cReq cInfo
pure (connId', False)
Nothing -> throwE $ AGENT A_VERSION
@@ -2208,7 +2208,7 @@ cleanupManager c@AgentClient {subQ} = do
deleteExpiredReplicasForDeletion = do
rcvFilesTTL <- asks $ rcvFilesTTL . config
withStore' c (`deleteDeletedSndChunkReplicasExpired` rcvFilesTTL)
notify :: forall e. AEntityI e => EntityId -> AEvent e -> AM ()
notify :: forall e. AEntityI e => AEntityId -> AEvent e -> AM ()
notify entId cmd = atomically $ writeTBQueue subQ ("", entId, AEvt (sAEntity @e) cmd)
data ACKd = ACKd | ACKPending
@@ -2345,7 +2345,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
HELLO -> helloMsg srvMsgId msgMeta conn'' >> 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>:" <> logSecret srvMsgId
logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret' srvMsgId
notify $ MSG msgMeta msgFlags body
pure ACKPending
A_RCVD rcpts -> qDuplex conn'' "RCVD" $ messagesRcvd rcpts msgMeta
@@ -2355,7 +2355,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
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>:" <> logSecret srvMsgId) >> 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 -> AM ()) -> AM ACKd
@@ -2378,7 +2378,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
| otherwise ->
liftEither (parse smpP (AGENT A_MESSAGE) agentMsgBody) >>= \case
AgentMessage _ (A_MSG body) -> do
logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret srvMsgId
logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret' srvMsgId
notify $ MSG msgMeta msgFlags body
pure ACKPending
_ -> ack
@@ -2500,7 +2500,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
smpConfirmation :: SMP.MsgId -> Connection c -> Maybe C.APublicAuthKey -> C.PublicKeyX25519 -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> ByteString -> VersionSMPC -> VersionSMPA -> AM ()
smpConfirmation srvMsgId conn' senderKey e2ePubKey e2eEncryption encConnInfo smpClientVersion agentVersion = do
logServer "<--" c srv rId $ "MSG <CONF>:" <> logSecret srvMsgId
logServer "<--" c srv rId $ "MSG <CONF>:" <> logSecret' srvMsgId
AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
let ConnData {pqSupport} = toConnData conn'
unless
@@ -2569,7 +2569,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
helloMsg :: SMP.MsgId -> MsgMeta -> Connection c -> AM ()
helloMsg srvMsgId MsgMeta {pqEncryption} conn' = do
logServer "<--" c srv rId $ "MSG <HELLO>:" <> logSecret srvMsgId
logServer "<--" c srv rId $ "MSG <HELLO>:" <> logSecret' srvMsgId
case status of
Active -> prohibited "hello: active"
_ ->
@@ -2593,7 +2593,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
continueSending srvMsgId addr (DuplexConnection _ _ sqs) =
case findQ addr sqs of
Just sq -> do
logServer "<--" c srv rId $ "MSG <QCONT>:" <> logSecret srvMsgId
logServer "<--" c srv rId $ "MSG <QCONT>:" <> logSecret' srvMsgId
atomically $
TM.lookup (qAddress sq) (smpDeliveryWorkers c)
>>= mapM_ (\(_, retryLock) -> tryPutTMVar retryLock ())
@@ -2602,7 +2602,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
messagesRcvd :: NonEmpty AMessageReceipt -> MsgMeta -> Connection 'CDuplex -> AM ACKd
messagesRcvd rcpts msgMeta@MsgMeta {broker = (srvMsgId, _)} _ = do
logServer "<--" c srv rId $ "MSG <RCPT>:" <> logSecret srvMsgId
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') $> ACKPending
@@ -2642,7 +2642,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
sq2 <- withStore c $ \db -> do
liftIO $ mapM_ (deleteConnSndQueue db connId) delSqs
addConnSndQueue db connId (sq_ :: NewSndQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
logServer "<--" c srv rId $ "MSG <QADD>:" <> logSecret srvMsgId <> " " <> 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', sndPublicKey)]
sq1 <- withStore' c $ \db -> setSndSwitchStatus db sq $ Just SSSendingQKEY
@@ -2663,7 +2663,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
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 srvMsgId <> " " <> 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
@@ -2684,7 +2684,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
case find ((replaceQId ==) . dbQId) sqs of
Just sq1 -> do
checkSQSwchStatus sq1 SSSendingQKEY
logServer "<--" c srv rId $ "MSG <QUSE>:" <> logSecret srvMsgId <> " " <> 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
@@ -2708,7 +2708,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
smpInvitation :: SMP.MsgId -> Connection c -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> AM ()
smpInvitation srvMsgId conn' connReq@(CRInvitationUri crData _) cInfo = do
logServer "<--" c srv rId $ "MSG <KEY>:" <> logSecret srvMsgId
logServer "<--" c srv rId $ "MSG <KEY>:" <> logSecret' srvMsgId
case conn' of
ContactConnection {} -> do
-- show connection request even if invitaion via contact address is not compatible.
src/Simplex/Messaging/Agent/Client.hs
+59 -49
View File
@@ -12,6 +12,7 @@
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StrictData #-}
@@ -82,6 +83,7 @@ module Simplex.Messaging.Agent.Client
deleteQueues,
logServer,
logSecret,
logSecret',
removeSubscription,
hasActiveSubscription,
hasPendingSubscription,
@@ -228,7 +230,7 @@ import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON, parse, sumT
import Simplex.Messaging.Protocol
( AProtocolType (..),
BrokerMsg,
EntityId,
EntityId (..),
ErrorType,
MsgFlags (..),
MsgId,
@@ -241,7 +243,6 @@ import Simplex.Messaging.Protocol
ProtocolServer (..),
ProtocolType (..),
ProtocolTypeI (..),
QueueId,
QueueIdsKeys (..),
RcvMessage (..),
RcvNtfPublicDhKey,
@@ -255,6 +256,7 @@ import Simplex.Messaging.Protocol
VersionSMPC,
XFTPServer,
XFTPServerWithAuth,
pattern NoEntity,
sameSrvAddr',
)
import qualified Simplex.Messaging.Protocol as SMP
@@ -996,7 +998,7 @@ withClient_ c tSess@(_, srv, _) action = do
where
logServerError :: AgentErrorType -> AM a
logServerError e = do
logServer "<--" c srv "" $ bshow e
logServer "<--" c srv NoEntity $ bshow e
throwE e
withProxySession :: AgentClient -> Maybe SMPServerWithAuth -> SMPTransportSession -> SMP.SenderId -> ByteString -> ((SMPConnectedClient, ProxiedRelay) -> AM a) -> AM a
@@ -1013,32 +1015,32 @@ withProxySession c proxySrv_ destSess@(_, destSrv, _) entId cmdStr action = do
proxySrv = showServer . protocolClientServer' . protocolClient
logServerError :: SMPConnectedClient -> AgentErrorType -> AM a
logServerError cl e = do
logServer ("<-- " <> proxySrv cl <> " <") c destSrv "" $ bshow e
logServer ("<-- " <> proxySrv cl <> " <") c destSrv NoEntity $ bshow e
throwE e
withLogClient_ :: ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> AM a) -> AM a
withLogClient_ :: ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> ByteString -> ByteString -> (Client msg -> AM a) -> AM a
withLogClient_ c tSess@(_, srv, _) entId cmdStr action = do
logServer "-->" c srv entId cmdStr
logServer' "-->" c srv entId cmdStr
res <- withClient_ c tSess action
logServer "<--" c srv entId "OK"
logServer' "<--" c srv entId "OK"
return res
withClient :: forall v err msg a. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
withClient c tSess action = withClient_ c tSess $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer $ protocolClient client) $ action client
{-# INLINE withClient #-}
withLogClient :: forall v err msg a. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
withLogClient :: forall v err msg a. ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> ByteString -> ByteString -> (Client msg -> ExceptT (ProtocolClientError err) IO a) -> AM a
withLogClient c tSess entId cmdStr action = withLogClient_ c tSess entId cmdStr $ \client -> liftClient (clientProtocolError @v @err @msg) (clientServer $ protocolClient client) $ action client
{-# INLINE withLogClient #-}
withSMPClient :: SMPQueueRec q => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT SMPClientError IO a) -> AM a
withSMPClient c q cmdStr action = do
tSess <- mkSMPTransportSession c q
withLogClient c tSess (queueId q) cmdStr $ action . connectedClient
withLogClient c tSess (unEntityId $ queueId q) cmdStr $ action . connectedClient
sendOrProxySMPMessage :: AgentClient -> UserId -> SMPServer -> ByteString -> Maybe SMP.SndPrivateAuthKey -> SMP.SenderId -> MsgFlags -> SMP.MsgBody -> AM (Maybe SMPServer)
sendOrProxySMPMessage c userId destSrv cmdStr spKey_ senderId msgFlags msg =
sendOrProxySMPCommand c userId destSrv cmdStr senderId sendViaProxy sendDirectly
sendOrProxySMPMessage :: AgentClient -> UserId -> SMPServer -> ConnId -> ByteString -> Maybe SMP.SndPrivateAuthKey -> SMP.SenderId -> MsgFlags -> SMP.MsgBody -> AM (Maybe SMPServer)
sendOrProxySMPMessage c userId destSrv connId cmdStr spKey_ senderId msgFlags msg =
sendOrProxySMPCommand c userId destSrv connId cmdStr senderId sendViaProxy sendDirectly
where
sendViaProxy smp proxySess = do
atomically $ incSMPServerStat c userId destSrv sentViaProxyAttempts
@@ -1052,14 +1054,15 @@ sendOrProxySMPCommand ::
AgentClient ->
UserId ->
SMPServer ->
ConnId ->
ByteString ->
SMP.SenderId ->
(SMPClient -> ProxiedRelay -> ExceptT SMPClientError IO (Either ProxyClientError ())) ->
(SMPClient -> ExceptT SMPClientError IO ()) ->
AM (Maybe SMPServer)
sendOrProxySMPCommand c userId destSrv cmdStr senderId sendCmdViaProxy sendCmdDirectly = do
sess <- mkTransportSession c userId destSrv senderId
ifM shouldUseProxy (sendViaProxy Nothing sess) (sendDirectly sess $> Nothing)
sendOrProxySMPCommand c userId destSrv connId cmdStr senderId sendCmdViaProxy sendCmdDirectly = do
tSess <- mkTransportSession c userId destSrv connId
ifM shouldUseProxy (sendViaProxy Nothing tSess) (sendDirectly tSess $> Nothing)
where
shouldUseProxy = do
cfg <- getNetworkConfig c
@@ -1078,7 +1081,7 @@ sendOrProxySMPCommand c userId destSrv cmdStr senderId sendCmdViaProxy sendCmdDi
SPFProhibit -> False
unknownServer = liftIO $ maybe True (notElem destSrv . knownSrvs) <$> TM.lookupIO userId (smpServers c)
sendViaProxy :: Maybe SMPServerWithAuth -> SMPTransportSession -> AM (Maybe SMPServer)
sendViaProxy proxySrv_ destSess@(_, _, qId) = do
sendViaProxy proxySrv_ destSess@(_, _, connId_) = do
r <- tryAgentError . withProxySession c proxySrv_ destSess senderId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess@ProxiedRelay {prBasicAuth}) -> do
r' <- liftClient SMP (clientServer smp) $ sendCmdViaProxy smp proxySess
let proxySrv = protocolClientServer' smp
@@ -1105,7 +1108,7 @@ sendOrProxySMPCommand c userId destSrv cmdStr senderId sendCmdViaProxy sendCmdDi
-- checks that the current proxied relay session is the same one that was used to send the message and removes it
deleteRelaySession =
( TM.lookup destSess (smpProxiedRelays c)
$>>= \(ProtoServerWithAuth srv _) -> tryReadSessVar (userId, srv, qId) (smpClients c)
$>>= \(ProtoServerWithAuth srv _) -> tryReadSessVar (userId, srv, connId_) (smpClients c)
)
>>= \case
Just (Right (SMPConnectedClient smp' prs))
@@ -1125,7 +1128,7 @@ sendOrProxySMPCommand c userId destSrv cmdStr senderId sendCmdViaProxy sendCmdDi
| serverHostError e -> ifM directAllowed (sendDirectly destSess $> Nothing) (throwE e)
| otherwise -> throwE e
sendDirectly tSess =
withLogClient_ c tSess senderId ("SEND " <> cmdStr) $ \(SMPConnectedClient smp _) -> do
withLogClient_ c tSess (unEntityId senderId) ("SEND " <> cmdStr) $ \(SMPConnectedClient smp _) -> do
r <- tryAgentError $ liftClient SMP (clientServer smp) $ sendCmdDirectly smp
case r of
Right () -> atomically $ incSMPServerStat c userId destSrv sentDirect
@@ -1138,18 +1141,18 @@ ipAddressProtected NetworkConfig {socksProxy, hostMode} (ProtocolServer _ hosts
isOnionHost = \case THOnionHost _ -> True; _ -> False
withNtfClient :: AgentClient -> NtfServer -> EntityId -> ByteString -> (NtfClient -> ExceptT NtfClientError IO a) -> AM a
withNtfClient c srv = withLogClient c (0, srv, Nothing)
withNtfClient c srv (EntityId entId) = withLogClient c (0, srv, Nothing) entId
withXFTPClient ::
ProtocolServerClient v err msg =>
AgentClient ->
(UserId, ProtoServer msg, EntityId) ->
(UserId, ProtoServer msg, ByteString) ->
ByteString ->
(Client msg -> ExceptT (ProtocolClientError err) IO b) ->
AM b
withXFTPClient c (userId, srv, entityId) cmdStr action = do
tSess <- mkTransportSession c userId srv entityId
withLogClient c tSess entityId cmdStr action
withXFTPClient c (userId, srv, sessEntId) cmdStr action = do
tSess <- mkTransportSession c userId srv sessEntId
withLogClient c tSess sessEntId cmdStr action
liftClient :: (Show err, Encoding err) => (HostName -> err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a
liftClient protocolError_ = liftError . protocolClientError protocolError_
@@ -1291,12 +1294,12 @@ getXFTPWorkPath = do
workDir <- readTVarIO =<< asks (xftpWorkDir . xftpAgent)
maybe getTemporaryDirectory pure workDir
mkTransportSession :: MonadIO m => AgentClient -> UserId -> ProtoServer msg -> EntityId -> m (TransportSession msg)
mkTransportSession c userId srv entityId = mkTSession userId srv entityId <$> getSessionMode c
mkTransportSession :: MonadIO m => AgentClient -> UserId -> ProtoServer msg -> ByteString -> m (TransportSession msg)
mkTransportSession c userId srv sessEntId = mkTSession userId srv sessEntId <$> getSessionMode c
{-# INLINE mkTransportSession #-}
mkTSession :: UserId -> ProtoServer msg -> EntityId -> TransportSessionMode -> TransportSession msg
mkTSession userId srv entityId mode = (userId, srv, if mode == TSMEntity then Just entityId else Nothing)
mkTSession :: UserId -> ProtoServer msg -> ByteString -> TransportSessionMode -> TransportSession msg
mkTSession userId srv sessEntId mode = (userId, srv, if mode == TSMEntity then Just sessEntId else Nothing)
{-# INLINE mkTSession #-}
mkSMPTransportSession :: (SMPQueueRec q, MonadIO m) => AgentClient -> q -> m SMPTransportSession
@@ -1318,12 +1321,12 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode sender
rKeys@(_, rcvPrivateKey) <- atomically $ C.generateAuthKeyPair a g
(dhKey, privDhKey) <- atomically $ C.generateKeyPair g
(e2eDhKey, e2ePrivKey) <- atomically $ C.generateKeyPair g
logServer "-->" c srv "" "NEW"
logServer "-->" c srv NoEntity "NEW"
tSess <- mkTransportSession c userId srv connId
(sessId, QIK {rcvId, sndId, rcvPublicDhKey, sndSecure}) <-
withClient c tSess $ \(SMPConnectedClient smp _) ->
(sessionId $ thParams smp,) <$> createSMPQueue smp rKeys dhKey auth subMode senderCanSecure
liftIO . logServer "<--" c srv "" $ B.unwords ["IDS", logSecret rcvId, logSecret sndId]
liftIO . logServer "<--" c srv NoEntity $ B.unwords ["IDS", logSecret rcvId, logSecret sndId]
let rq =
RcvQueue
{ userId,
@@ -1457,7 +1460,7 @@ sendTSessionBatches statCmd toRQ action c qs =
tryAgentError' (getSMPServerClient c tSess) >>= \case
Left e -> pure $ L.map ((,Left e) . toRQ) qs'
Right (SMPConnectedClient smp _) -> liftIO $ do
logServer "-->" c srv (bshow (length qs') <> " queues") statCmd
logServer' "-->" c srv (bshow (length qs') <> " queues") statCmd
L.map agentError <$> action smp qs'
where
agentError = second . first $ protocolClientError SMP $ clientServer smp
@@ -1511,32 +1514,39 @@ getSubscriptions :: AgentClient -> IO (Set ConnId)
getSubscriptions = readTVarIO . subscrConns
{-# INLINE getSubscriptions #-}
logServer :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> QueueId -> ByteString -> m ()
logServer dir AgentClient {clientId} srv qId cmdStr =
logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret qId, cmdStr]
logServer :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> EntityId -> ByteString -> m ()
logServer dir c srv = logServer' dir c srv . unEntityId
{-# INLINE logServer #-}
logServer' :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> ByteString -> ByteString -> m ()
logServer' dir AgentClient {clientId} srv qStr cmdStr =
logInfo . decodeUtf8 $ B.unwords ["A", "(" <> bshow clientId <> ")", dir, showServer srv, ":", logSecret' qStr, cmdStr]
showServer :: ProtocolServer s -> ByteString
showServer ProtocolServer {host, port} =
strEncode host <> B.pack (if null port then "" else ':' : port)
{-# INLINE showServer #-}
logSecret :: ByteString -> ByteString
logSecret bs = B64.encode $ B.take 3 bs
logSecret :: EntityId -> ByteString
logSecret = logSecret' . unEntityId
{-# INLINE logSecret #-}
logSecret' :: ByteString -> ByteString
logSecret' = B64.encode . B.take 3
{-# INLINE logSecret' #-}
sendConfirmation :: AgentClient -> SndQueue -> ByteString -> AM (Maybe SMPServer)
sendConfirmation c sq@SndQueue {userId, server, sndId, sndSecure, sndPublicKey, sndPrivateKey, e2ePubKey = e2ePubKey@Just {}} agentConfirmation = do
sendConfirmation c sq@SndQueue {userId, server, connId, sndId, sndSecure, sndPublicKey, sndPrivateKey, e2ePubKey = e2ePubKey@Just {}} agentConfirmation = do
let (privHdr, spKey) = if sndSecure then (SMP.PHEmpty, Just sndPrivateKey) else (SMP.PHConfirmation sndPublicKey, Nothing)
clientMsg = SMP.ClientMessage privHdr agentConfirmation
msg <- agentCbEncrypt sq e2ePubKey $ smpEncode clientMsg
sendOrProxySMPMessage c userId server "<CONF>" spKey sndId (MsgFlags {notification = True}) msg
sendOrProxySMPMessage c userId server connId "<CONF>" spKey sndId (MsgFlags {notification = True}) msg
sendConfirmation _ _ _ = throwE $ INTERNAL "sendConfirmation called without snd_queue public key(s) in the database"
sendInvitation :: AgentClient -> UserId -> Compatible SMPQueueInfo -> Compatible VersionSMPA -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> AM (Maybe SMPServer)
sendInvitation c userId (Compatible (SMPQueueInfo v SMPQueueAddress {smpServer, senderId, dhPublicKey})) (Compatible agentVersion) connReq connInfo = do
sendInvitation :: AgentClient -> UserId -> ConnId -> Compatible SMPQueueInfo -> Compatible VersionSMPA -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> AM (Maybe SMPServer)
sendInvitation c userId connId (Compatible (SMPQueueInfo v SMPQueueAddress {smpServer, senderId, dhPublicKey})) (Compatible agentVersion) connReq connInfo = do
msg <- mkInvitation
sendOrProxySMPMessage c userId smpServer "<INV>" Nothing senderId (MsgFlags {notification = True}) msg
sendOrProxySMPMessage c userId smpServer connId "<INV>" Nothing senderId (MsgFlags {notification = True}) msg
where
mkInvitation :: AM ByteString
-- this is only encrypted with per-queue E2E, not with double ratchet
@@ -1572,8 +1582,8 @@ secureQueue c rq@RcvQueue {rcvId, rcvPrivateKey} senderKey =
secureSMPQueue smp rcvPrivateKey rcvId senderKey
secureSndQueue :: AgentClient -> SndQueue -> AM ()
secureSndQueue c SndQueue {userId, server, sndId, sndPrivateKey, sndPublicKey} =
void $ sendOrProxySMPCommand c userId server "SKEY <key>" sndId secureViaProxy secureDirectly
secureSndQueue c SndQueue {userId, connId, server, sndId, sndPrivateKey, sndPublicKey} =
void $ sendOrProxySMPCommand c userId server connId "SKEY <key>" sndId secureViaProxy secureDirectly
where
-- TODO track statistics
secureViaProxy smp proxySess = proxySecureSndSMPQueue smp proxySess sndPrivateKey sndId sndPublicKey
@@ -1603,7 +1613,7 @@ disableQueuesNtfs = sendTSessionBatches "NDEL" id $ sendBatch disableSMPQueuesNt
sendAck :: AgentClient -> RcvQueue -> MsgId -> AM ()
sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = do
withSMPClient c rq ("ACK:" <> logSecret msgId) $ \smp ->
withSMPClient c rq ("ACK:" <> logSecret' msgId) $ \smp ->
ackSMPMessage smp rcvPrivateKey rcvId msgId
atomically $ releaseGetLock c rq
@@ -1637,10 +1647,10 @@ deleteQueues c = sendTSessionBatches "DEL" id deleteQueues_ c
pure rs
sendAgentMessage :: AgentClient -> SndQueue -> MsgFlags -> ByteString -> AM (Maybe SMPServer)
sendAgentMessage c sq@SndQueue {userId, server, sndId, sndPrivateKey} msgFlags agentMsg = do
sendAgentMessage c sq@SndQueue {userId, server, connId, sndId, sndPrivateKey} msgFlags agentMsg = do
let clientMsg = SMP.ClientMessage SMP.PHEmpty agentMsg
msg <- agentCbEncrypt sq Nothing $ smpEncode clientMsg
sendOrProxySMPMessage c userId server "<MSG>" (Just sndPrivateKey) sndId msgFlags msg
sendOrProxySMPMessage c userId server connId "<MSG>" (Just sndPrivateKey) sndId msgFlags msg
data ServerQueueInfo = ServerQueueInfo
{ server :: SMPServer,
@@ -1659,7 +1669,7 @@ getQueueInfo c rq@RcvQueue {server, rcvId, rcvPrivateKey, sndId, status, clientN
let ntfId = enc . (\ClientNtfCreds {notifierId} -> notifierId) <$> clientNtfCreds
pure ServerQueueInfo {server, rcvId = enc rcvId, sndId = enc sndId, ntfId, status = serializeQueueStatus status, info}
where
enc = decodeLatin1 . B64.encode
enc = decodeLatin1 . B64.encode . unEntityId
agentNtfRegisterToken :: AgentClient -> NtfToken -> NtfPublicAuthKey -> C.PublicKeyX25519 -> AM (NtfTokenId, C.PublicKeyX25519)
agentNtfRegisterToken c NtfToken {deviceToken, ntfServer, ntfPrivKey} ntfPubKey pubDhKey =
@@ -1707,10 +1717,10 @@ agentXFTPNewChunk c SndFileChunk {userId, chunkSpec = XFTPChunkSpec {chunkSize},
rKeys <- xftpRcvKeys n
(sndKey, replicaKey) <- atomically . C.generateAuthKeyPair C.SEd25519 =<< asks random
let fileInfo = FileInfo {sndKey, size = chunkSize, digest = chunkDigest}
logServer "-->" c srv "" "FNEW"
logServer "-->" c srv NoEntity "FNEW"
tSess <- mkTransportSession c userId srv chunkDigest
(sndId, rIds) <- withClient c tSess $ \xftp -> X.createXFTPChunk xftp replicaKey fileInfo (L.map fst rKeys) auth
logServer "<--" c srv "" $ B.unwords ["SIDS", logSecret sndId]
logServer "<--" c srv NoEntity $ B.unwords ["SIDS", logSecret sndId]
pure NewSndChunkReplica {server = srv, replicaId = ChunkReplicaId sndId, replicaKey, rcvIdsKeys = L.toList $ xftpRcvIdsKeys rIds rKeys}
agentXFTPUploadChunk :: AgentClient -> UserId -> FileDigest -> SndFileChunkReplica -> XFTPChunkSpec -> AM ()
@@ -1734,7 +1744,7 @@ xftpRcvKeys n = do
Just rKeys' -> pure rKeys'
_ -> throwE $ INTERNAL "non-positive number of recipients"
xftpRcvIdsKeys :: NonEmpty ByteString -> NonEmpty C.AAuthKeyPair -> NonEmpty (ChunkReplicaId, C.APrivateAuthKey)
xftpRcvIdsKeys :: NonEmpty EntityId -> NonEmpty C.AAuthKeyPair -> NonEmpty (ChunkReplicaId, C.APrivateAuthKey)
xftpRcvIdsKeys rIds rKeys = L.map ChunkReplicaId rIds `L.zip` L.map snd rKeys
agentCbEncrypt :: SndQueue -> Maybe C.PublicKeyX25519 -> ByteString -> AM ByteString
src/Simplex/Messaging/Agent/Protocol.hs
+4 -2
View File
@@ -51,6 +51,7 @@ module Simplex.Messaging.Agent.Protocol
-- * SMP agent protocol types
ConnInfo,
SndQueueSecured,
AEntityId,
ACommand (..),
AEvent (..),
AEvt (..),
@@ -190,7 +191,6 @@ import Simplex.Messaging.Parsers
import Simplex.Messaging.Protocol
( AProtocolType,
BrokerErrorType (..),
EntityId,
ErrorType,
MsgBody,
MsgFlags,
@@ -287,10 +287,12 @@ e2eEncAgentMsgLength v = \case
_ -> 15856
-- | SMP agent event
type ATransmission = (ACorrId, EntityId, AEvt)
type ATransmission = (ACorrId, AEntityId, AEvt)
type UserId = Int64
type AEntityId = ByteString
type ACorrId = ByteString
data AEntity = AEConn | AERcvFile | AESndFile | AENone
src/Simplex/Messaging/Agent/Store/SQLite.hs
+11
View File
@@ -1,10 +1,12 @@
{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE InstanceSigs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE MultiParamTypeClasses #-}
@@ -15,6 +17,7 @@
{-# LANGUAGE QuasiQuotes #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StandaloneDeriving #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeOperators #-}
@@ -1814,6 +1817,14 @@ instance ToField (Version v) where toField (Version v) = toField v
instance FromField (Version v) where fromField f = Version <$> fromField f
deriving newtype instance ToField EntityId
deriving newtype instance FromField EntityId
deriving newtype instance ToField ChunkReplicaId
deriving newtype instance FromField ChunkReplicaId
listToEither :: e -> [a] -> Either e a
listToEither _ (x : _) = Right x
listToEither e _ = Left e
src/Simplex/Messaging/Agent/TRcvQueues.hs
+3 -3
View File
@@ -31,7 +31,7 @@ import Simplex.Messaging.Transport
class Queue q where
connId' :: q -> ConnId
qKey :: q -> (UserId, SMPServer, ConnId)
qKey :: q -> (UserId, SMPServer, RecipientId)
-- the fields in this record have the same data with swapped keys for lookup efficiency,
-- and all methods must maintain this invariant.
@@ -97,7 +97,7 @@ getDelSessQueues tSess sessId' (TRcvQueues qs cs) = do
delQ acc@(removed, qs') (sessId, rq)
| rq `isSession` tSess && sessId == sessId' = (rq : removed, M.delete (qKey rq) qs')
| otherwise = acc
delConn :: ([ConnId], M.Map ConnId (NonEmpty (UserId, SMPServer, ConnId))) -> RcvQueue -> ([ConnId], M.Map ConnId (NonEmpty (UserId, SMPServer, ConnId)))
delConn :: ([ConnId], M.Map ConnId (NonEmpty (UserId, SMPServer, RecipientId))) -> RcvQueue -> ([ConnId], M.Map ConnId (NonEmpty (UserId, SMPServer, RecipientId)))
delConn (removed, cs') rq = M.alterF f cId cs'
where
cId = connId rq
@@ -113,7 +113,7 @@ isSession rq (uId, srv, connId_) =
instance Queue RcvQueue where
connId' = connId
qKey rq = (userId rq, server rq, connId rq)
qKey rq = (userId rq, server rq, rcvId rq)
instance Queue (SessionId, RcvQueue) where
connId' = connId . snd
src/Simplex/Messaging/Client.hs
+8 -7
View File
@@ -441,7 +441,8 @@ transportSession' = transportSession . client_
type UserId = Int64
-- | Transport session key - includes entity ID if `sessionMode = TSMEntity`.
type TransportSession msg = (UserId, ProtoServer msg, Maybe EntityId)
-- Please note that for SMP connection ID is used as entity ID, not queue ID.
type TransportSession msg = (UserId, ProtoServer msg, Maybe ByteString)
-- | Connects to 'ProtocolServer' using passed client configuration
-- and queue for messages and notifications.
@@ -544,7 +545,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
if remaining > 1_000_000 -- delay pings only for significant time
then loop remaining
else do
whenM (readTVarIO sendPings) $ void . runExceptT $ sendProtocolCommand c Nothing "" (protocolPing @v @err @msg)
whenM (readTVarIO sendPings) $ void . runExceptT $ sendProtocolCommand c Nothing NoEntity (protocolPing @v @err @msg)
-- sendProtocolCommand/getResponse updates counter for each command
cnt <- readTVarIO timeoutErrorCount
-- drop client when maxCnt of commands have timed out in sequence, but only after some time has passed after last received response
@@ -670,7 +671,7 @@ createSMPQueue ::
Bool ->
ExceptT SMPClientError IO QueueIdsKeys
createSMPQueue c (rKey, rpKey) dhKey auth subMode sndSecure =
sendSMPCommand c (Just rpKey) "" (NEW rKey dhKey auth subMode sndSecure) >>= \case
sendSMPCommand c (Just rpKey) NoEntity (NEW rKey dhKey auth subMode sndSecure) >>= \case
IDS qik -> pure qik
r -> throwE $ unexpectedResponse r
@@ -829,7 +830,7 @@ deleteSMPQueues = okSMPCommands DEL
connectSMPProxiedRelay :: SMPClient -> SMPServer -> Maybe BasicAuth -> ExceptT SMPClientError IO ProxiedRelay
connectSMPProxiedRelay c@ProtocolClient {client_ = PClient {tcpConnectTimeout, tcpTimeout}} relayServ@ProtocolServer {keyHash = C.KeyHash kh} proxyAuth
| thVersion (thParams c) >= sendingProxySMPVersion =
sendProtocolCommand_ c Nothing tOut Nothing "" (Cmd SProxiedClient (PRXY relayServ proxyAuth)) >>= \case
sendProtocolCommand_ c Nothing tOut Nothing NoEntity (Cmd SProxiedClient (PRXY relayServ proxyAuth)) >>= \case
PKEY sId vr (chain, key) ->
case supportedClientSMPRelayVRange `compatibleVersion` vr of
Nothing -> throwE $ transportErr TEVersion
@@ -931,7 +932,7 @@ proxySMPCommand c@ProtocolClient {thParams = proxyThParams, client_ = PClient {c
et <- liftEitherWith PCECryptoError $ EncTransmission <$> C.cbEncrypt cmdSecret nonce b paddedProxiedTLength
-- proxy interaction errors are wrapped
let tOut = Just $ 2 * tcpTimeout
tryE (sendProtocolCommand_ c (Just nonce) tOut Nothing sessionId (Cmd SProxiedClient (PFWD v cmdPubKey et))) >>= \case
tryE (sendProtocolCommand_ c (Just nonce) tOut Nothing (EntityId sessionId) (Cmd SProxiedClient (PFWD v cmdPubKey et))) >>= \case
Right r -> case r of
PRES (EncResponse er) -> do
-- server interaction errors are thrown directly
@@ -967,7 +968,7 @@ forwardSMPTransmission c@ProtocolClient {thParams, client_ = PClient {clientCorr
let fwdT = FwdTransmission {fwdCorrId, fwdVersion, fwdKey, fwdTransmission}
eft = EncFwdTransmission $ C.cbEncryptNoPad sessSecret nonce (smpEncode fwdT)
-- send
sendProtocolCommand_ c (Just nonce) Nothing Nothing "" (Cmd SSender (RFWD eft)) >>= \case
sendProtocolCommand_ c (Just nonce) Nothing Nothing NoEntity (Cmd SSender (RFWD eft)) >>= \case
RRES (EncFwdResponse efr) -> do
-- unwrap
r' <- liftEitherWith PCECryptoError $ C.cbDecryptNoPad sessSecret (C.reverseNonce nonce) efr
@@ -1015,7 +1016,7 @@ sendProtocolCommands c@ProtocolClient {thParams = THandleParams {batch, blockSiz
| diff == 0 = pure $ L.fromList rs
| diff > 0 = do
putStrLn "send error: fewer responses than expected"
pure $ L.fromList $ rs <> replicate diff (Response "" $ Left $ PCETransportError TEBadBlock)
pure $ L.fromList $ rs <> replicate diff (Response NoEntity $ Left $ PCETransportError TEBadBlock)
| otherwise = do
putStrLn "send error: more responses than expected"
pure $ L.fromList $ take (L.length cs) rs
src/Simplex/Messaging/Encoding.hs
+2
View File
@@ -42,10 +42,12 @@ class Encoding a where
-- | decoding of type (default implementation uses parser)
smpDecode :: ByteString -> Either String a
smpDecode = parseAll smpP
{-# INLINE smpDecode #-}
-- | protocol parser of type (default implementation parses protocol ByteString encoding)
smpP :: Parser a
smpP = smpDecode <$?> smpP
{-# INLINE smpP #-}
instance Encoding Char where
smpEncode = B.singleton
src/Simplex/Messaging/Encoding/String.hs
+6
View File
@@ -53,14 +53,20 @@ class StrEncoding a where
-- Please note - if you only specify strDecode, it will use base64urlP as default parser before decoding the string
strDecode :: ByteString -> Either String a
strDecode = parseAll strP
{-# INLINE strDecode #-}
strP :: Parser a
strP = strDecode <$?> base64urlP
{-# INLINE strP #-}
-- base64url encoding/decoding of ByteStrings - the parser only allows non-empty strings
instance StrEncoding ByteString where
strEncode = U.encode
{-# INLINE strEncode #-}
strDecode = U.decode
{-# INLINE strDecode #-}
strP = base64urlP
{-# INLINE strP #-}
base64urlP :: Parser ByteString
base64urlP = do
src/Simplex/Messaging/Notifications/Client.hs
+4 -3
View File
@@ -1,6 +1,7 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
module Simplex.Messaging.Notifications.Client where
@@ -11,7 +12,7 @@ import Simplex.Messaging.Client
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Notifications.Protocol
import Simplex.Messaging.Notifications.Transport (NTFVersion, supportedClientNTFVRange, supportedNTFHandshakes)
import Simplex.Messaging.Protocol (ErrorType)
import Simplex.Messaging.Protocol (ErrorType, pattern NoEntity)
type NtfClient = ProtocolClient NTFVersion ErrorType NtfResponse
@@ -22,7 +23,7 @@ defaultNTFClientConfig = defaultClientConfig (Just supportedNTFHandshakes) suppo
ntfRegisterToken :: NtfClient -> C.APrivateAuthKey -> NewNtfEntity 'Token -> ExceptT NtfClientError IO (NtfTokenId, C.PublicKeyX25519)
ntfRegisterToken c pKey newTkn =
sendNtfCommand c (Just pKey) "" (TNEW newTkn) >>= \case
sendNtfCommand c (Just pKey) NoEntity (TNEW newTkn) >>= \case
NRTknId tknId dhKey -> pure (tknId, dhKey)
r -> throwE $ unexpectedResponse r
@@ -46,7 +47,7 @@ ntfEnableCron c pKey tknId int = okNtfCommand (TCRN int) c pKey tknId
ntfCreateSubscription :: NtfClient -> C.APrivateAuthKey -> NewNtfEntity 'Subscription -> ExceptT NtfClientError IO NtfSubscriptionId
ntfCreateSubscription c pKey newSub =
sendNtfCommand c (Just pKey) "" (SNEW newSub) >>= \case
sendNtfCommand c (Just pKey) NoEntity (SNEW newSub) >>= \case
NRSubId subId -> pure subId
r -> throwE $ unexpectedResponse r
src/Simplex/Messaging/Notifications/Protocol.hs
+3 -3
View File
@@ -208,7 +208,7 @@ instance NtfEntityI e => ProtocolEncoding NTFVersion ErrorType (NtfCommand e) wh
fromProtocolError = fromProtocolError @NTFVersion @ErrorType @NtfResponse
{-# INLINE fromProtocolError #-}
checkCredentials (auth, _, entityId, _) cmd = case cmd of
checkCredentials (auth, _, EntityId entityId, _) cmd = case cmd of
-- TNEW and SNEW must have signature but NOT token/subscription IDs
TNEW {} -> sigNoEntity
SNEW {} -> sigNoEntity
@@ -322,7 +322,7 @@ instance ProtocolEncoding NTFVersion ErrorType NtfResponse where
PEBlock -> BLOCK
{-# INLINE fromProtocolError #-}
checkCredentials (_, _, entId, _) cmd = case cmd of
checkCredentials (_, _, EntityId entId, _) cmd = case cmd of
-- IDTKN response must not have queue ID
NRTknId {} -> noEntity
-- IDSUB response must not have queue ID
@@ -426,7 +426,7 @@ instance FromJSON DeviceToken where
t <- encodeUtf8 <$> o .: "token"
pure $ DeviceToken pp t
type NtfEntityId = ByteString
type NtfEntityId = EntityId
type NtfSubscriptionId = NtfEntityId
src/Simplex/Messaging/Notifications/Server.hs
+5 -4
View File
@@ -7,6 +7,7 @@
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
@@ -44,7 +45,7 @@ import Simplex.Messaging.Notifications.Server.Stats
import Simplex.Messaging.Notifications.Server.Store
import Simplex.Messaging.Notifications.Server.StoreLog
import Simplex.Messaging.Notifications.Transport
import Simplex.Messaging.Protocol (ErrorType (..), ProtocolServer (host), SMPServer, SignedTransmission, Transmission, encodeTransmission, tGet, tPut)
import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), ProtocolServer (host), SMPServer, SignedTransmission, Transmission, pattern NoEntity, encodeTransmission, tGet, tPut)
import qualified Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.Server
import Simplex.Messaging.Server.Stats
@@ -448,7 +449,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
atomically $ writeTBQueue pushQ (tkn, PNVerification regCode)
withNtfLog (`logCreateToken` tkn)
incNtfStatT token tknCreated
pure (corrId, "", NRTknId tknId srvDhPubKey)
pure (corrId, NoEntity, NRTknId tknId srvDhPubKey)
NtfReqCmd SToken (NtfTkn tkn@NtfTknData {token, ntfTknId, tknStatus, tknRegCode, tknDhSecret, tknDhKeys = (srvDhPubKey, srvDhPrivKey), tknCronInterval}) (corrId, tknId, cmd) -> do
status <- readTVarIO tknStatus
(corrId,tknId,) <$> case cmd of
@@ -539,7 +540,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
_ -> pure $ NRErr AUTH
withNtfLog (`logCreateSubscription` sub)
incNtfStat subCreated
pure (corrId, "", resp)
pure (corrId, NoEntity, resp)
NtfReqCmd SSubscription (NtfSub NtfSubData {smpQueue = SMPQueueNtf {smpServer, notifierId}, notifierKey = registeredNKey, subStatus}) (corrId, subId, cmd) -> do
status <- readTVarIO subStatus
(corrId,subId,) <$> case cmd of
@@ -564,7 +565,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
PING -> pure NRPong
NtfReqPing corrId entId -> pure (corrId, entId, NRPong)
getId :: M NtfEntityId
getId = randomBytes =<< asks (subIdBytes . config)
getId = fmap EntityId . randomBytes =<< asks (subIdBytes . config)
getRegCode :: M NtfRegCode
getRegCode = NtfRegCode <$> (randomBytes =<< asks (regCodeBytes . config))
randomBytes :: Int -> M ByteString
src/Simplex/Messaging/Protocol.hs
+24 -12
View File
@@ -2,11 +2,13 @@
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE FunctionalDependencies #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedLists #-}
@@ -99,7 +101,8 @@ module Simplex.Messaging.Protocol
BasicAuth (..),
SrvLoc (..),
CorrId (..),
EntityId,
EntityId (..),
pattern NoEntity,
QueueId,
RecipientId,
SenderId,
@@ -329,8 +332,8 @@ data RawTransmission = RawTransmission
{ authenticator :: ByteString, -- signature or encrypted transmission hash
authorized :: ByteString, -- authorized transmission
sessId :: SessionId,
corrId :: ByteString,
entityId :: ByteString,
corrId :: CorrId,
entityId :: EntityId,
command :: ByteString
}
deriving (Show)
@@ -357,7 +360,7 @@ instance IsString (Maybe TransmissionAuth) where
fromString = parseString $ B64.decode >=> C.decodeSignature >=> pure . fmap TASignature
-- | unparsed sent SMP transmission with signature, without session ID.
type SignedRawTransmission = (Maybe TransmissionAuth, SessionId, ByteString, ByteString)
type SignedRawTransmission = (Maybe TransmissionAuth, CorrId, EntityId, ByteString)
-- | unparsed sent SMP transmission with signature.
type SentRawTransmission = (Maybe TransmissionAuth, ByteString)
@@ -374,7 +377,13 @@ type NotifierId = QueueId
-- | SMP queue ID on the server.
type QueueId = EntityId
type EntityId = ByteString
-- this type is used for server entities only
newtype EntityId = EntityId {unEntityId :: ByteString}
deriving (Eq, Ord, Show)
deriving newtype (Encoding, StrEncoding)
pattern NoEntity :: EntityId
pattern NoEntity = EntityId ""
-- | Parameterized type for SMP protocol commands from all clients.
data Command (p :: Party) where
@@ -1097,10 +1106,13 @@ serverStrP = do
portP = show <$> (A.char ':' *> (A.decimal :: Parser Int))
-- | Transmission correlation ID.
newtype CorrId = CorrId {bs :: ByteString} deriving (Eq, Ord, Show)
newtype CorrId = CorrId {bs :: ByteString}
deriving (Eq, Ord, Show)
deriving newtype (Encoding)
instance IsString CorrId where
fromString = CorrId . fromString
{-# INLINE fromString #-}
instance StrEncoding CorrId where
strEncode (CorrId cId) = strEncode cId
@@ -1323,7 +1335,7 @@ instance PartyI p => ProtocolEncoding SMPVersion ErrorType (Command p) where
fromProtocolError = fromProtocolError @SMPVersion @ErrorType @BrokerMsg
{-# INLINE fromProtocolError #-}
checkCredentials (auth, _, entId, _) cmd = case cmd of
checkCredentials (auth, _, EntityId entId, _) cmd = case cmd of
-- NEW must have signature but NOT queue ID
NEW {}
| isNothing auth -> Left $ CMD NO_AUTH
@@ -1448,7 +1460,7 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
PEBlock -> BLOCK
{-# INLINE fromProtocolError #-}
checkCredentials (_, _, entId, _) cmd = case cmd of
checkCredentials (_, _, EntityId entId, _) cmd = case cmd of
-- IDS response should not have queue ID
IDS _ -> Right cmd
-- ERR response does not always have queue ID
@@ -1721,16 +1733,16 @@ tDecodeParseValidate THandleParams {sessionId, thVersion = v, implySessId} = \ca
| implySessId || sessId == sessionId ->
let decodedTransmission = (,corrId,entityId,command) <$> decodeTAuthBytes authenticator
in either (const $ tError corrId) (tParseValidate authorized) decodedTransmission
| otherwise -> (Nothing, "", (CorrId corrId, "", Left $ fromProtocolError @v @err @cmd PESession))
| otherwise -> (Nothing, "", (corrId, NoEntity, Left $ fromProtocolError @v @err @cmd PESession))
Left _ -> tError ""
where
tError :: ByteString -> SignedTransmission err cmd
tError corrId = (Nothing, "", (CorrId corrId, "", Left $ fromProtocolError @v @err @cmd PEBlock))
tError :: CorrId -> SignedTransmission err cmd
tError corrId = (Nothing, "", (corrId, NoEntity, Left $ fromProtocolError @v @err @cmd PEBlock))
tParseValidate :: ByteString -> SignedRawTransmission -> SignedTransmission err cmd
tParseValidate signed t@(sig, corrId, entityId, command) =
let cmd = parseProtocol @v @err @cmd v command >>= checkCredentials t
in (sig, signed, (CorrId corrId, entityId, cmd))
in (sig, signed, (corrId, entityId, cmd))
$(J.deriveJSON defaultJSON ''MsgFlags)
src/Simplex/Messaging/Server.hs
+12 -8
View File
@@ -898,7 +898,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
reply :: MonadIO m => NonEmpty (Transmission BrokerMsg) -> m ()
reply = atomically . writeTBQueue sndQ
processProxiedCmd :: Transmission (Command 'ProxiedClient) -> M (Maybe (Transmission BrokerMsg))
processProxiedCmd (corrId, sessId, command) = (corrId,sessId,) <$$> case command of
processProxiedCmd (corrId, EntityId sessId, command) = (corrId,EntityId sessId,) <$$> case command of
PRXY srv auth -> ifM allowProxy getRelay (pure $ Just $ ERR $ PROXY BASIC_AUTH)
where
allowProxy = do
@@ -961,7 +961,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
forkProxiedCmd cmdAction = do
bracket_ wait signal . forkClient clnt (B.unpack $ "client $" <> encode sessionId <> " proxy") $ do
-- commands MUST be processed under a reasonable timeout or the client would halt
cmdAction >>= \t -> reply [(corrId, sessId, t)]
cmdAction >>= \t -> reply [(corrId, EntityId sessId, t)]
pure Nothing
where
wait = do
@@ -987,8 +987,8 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
| otherwise -> pure $ ERR AUTH
Nothing -> pure $ ERR INTERNAL
SEND flags msgBody -> withQueue $ \qr -> sendMessage qr flags msgBody
PING -> pure (corrId, "", PONG)
RFWD encBlock -> (corrId, "",) <$> processForwardedCommand encBlock
PING -> pure (corrId, NoEntity, PONG)
RFWD encBlock -> (corrId, NoEntity,) <$> processForwardedCommand encBlock
Cmd SNotifier NSUB -> Just <$> subscribeNotifications
Cmd SRecipient command -> do
st <- asks queueStore
@@ -1265,7 +1265,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
THandleParams {thVersion} = thParams'
mkMessage :: C.MaxLenBS MaxMessageLen -> M Message
mkMessage body = do
msgId <- randomId =<< asks (msgIdBytes . config)
msgId <- randomId' =<< asks (msgIdBytes . config)
msgTs <- liftIO getSystemTime
pure $! Message msgId msgTs msgFlags body
@@ -1508,7 +1508,7 @@ withLog action = do
liftIO . mapM_ action $ storeLog (env :: Env)
timed :: T.Text -> RecipientId -> M a -> M a
timed name qId a = do
timed name (EntityId qId) a = do
t <- liftIO getSystemTime
r <- a
t' <- liftIO getSystemTime
@@ -1519,8 +1519,12 @@ timed name qId a = do
diff t t' = (systemSeconds t' - systemSeconds t) * sec + fromIntegral (systemNanoseconds t' - systemNanoseconds t)
sec = 1000_000000
randomId :: Int -> M ByteString
randomId n = atomically . C.randomBytes n =<< asks random
randomId' :: Int -> M ByteString
randomId' n = atomically . C.randomBytes n =<< asks random
randomId :: Int -> M EntityId
randomId = fmap EntityId . randomId'
{-# INLINE randomId #-}
saveServerMessages :: Bool -> M ()
saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessages
src/Simplex/Messaging/Server/Control.hs
+2 -3
View File
@@ -4,9 +4,8 @@
module Simplex.Messaging.Server.Control where
import qualified Data.Attoparsec.ByteString.Char8 as A
import Data.ByteString (ByteString)
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (BasicAuth)
import Simplex.Messaging.Protocol (BasicAuth, SenderId)
data CPClientRole = CPRNone | CPRUser | CPRAdmin
deriving (Eq)
@@ -22,7 +21,7 @@ data ControlProtocol
| CPSockets
| CPSocketThreads
| CPServerInfo
| CPDelete ByteString
| CPDelete SenderId
| CPSave
| CPHelp
| CPQuit
src/Simplex/Messaging/Transport.hs
-1
View File
@@ -116,7 +116,6 @@ import Simplex.Messaging.Version.Internal
import System.IO.Error (isEOFError)
import UnliftIO.Exception (Exception)
import qualified UnliftIO.Exception as E
import UnliftIO.STM
-- * Transport parameters
src/Simplex/Messaging/Util.hs
+2
View File
@@ -168,9 +168,11 @@ threadDelay' = loop
diffToMicroseconds :: NominalDiffTime -> Int64
diffToMicroseconds diff = truncate $ diff * 1000000
{-# INLINE diffToMicroseconds #-}
diffToMilliseconds :: NominalDiffTime -> Int64
diffToMilliseconds diff = truncate $ diff * 1000
{-# INLINE diffToMilliseconds #-}
labelMyThread :: MonadIO m => String -> m ()
labelMyThread label = liftIO $ myThreadId >>= (`labelThread` label)