mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 15:12:26 +00:00
remove SMP server signing responses (#226)
* remove SMP server signing responses * keep only one session ID
This commit is contained in:
1 parent
de01692ffd
commit
323fb1f03c
22 files changed
+86
-162
No files matched your search
@@ -671,7 +671,7 @@ newSndQueue_ ::
|
||||
SMPQueueUri ->
|
||||
C.APublicEncryptKey ->
|
||||
m (SndQueue, SndPublicVerifyKey, C.APublicVerifyKey)
|
||||
newSndQueue_ a (SMPQueueUri smpServer senderId _) encryptKey = do
|
||||
newSndQueue_ a (SMPQueueUri smpServer senderId) encryptKey = do
|
||||
size <- asks $ rsaKeySize . config
|
||||
(senderKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair size a
|
||||
(verifyKey, signKey) <- liftIO $ C.generateSignatureKeyPair size C.SRSA
|
||||
|
||||
@@ -240,9 +240,8 @@ newRcvQueue_ a c srv = do
|
||||
(recipientKey, rcvPrivateKey) <- liftIO $ C.generateSignatureKeyPair size a
|
||||
(dhKey, privDhKey) <- liftIO $ C.generateKeyPair' 0
|
||||
logServer "-->" c srv "" "NEW"
|
||||
QIK {rcvId, rcvSrvVerifyKey, rcvPublicDHKey, sndId, sndSrvVerifyKey} <-
|
||||
QIK {rcvId, sndId, rcvPublicDHKey} <-
|
||||
withSMP c srv $ \smp -> createSMPQueue smp rcvPrivateKey recipientKey dhKey
|
||||
let rcvDhSecret = C.dh' rcvPublicDHKey privDhKey
|
||||
logServer "<--" c srv "" $ B.unwords ["IDS", logSecret rcvId, logSecret sndId]
|
||||
(encryptKey, decryptKey) <- liftIO $ C.generateEncryptionKeyPair size C.SRSA
|
||||
let rq =
|
||||
@@ -250,15 +249,13 @@ newRcvQueue_ a c srv = do
|
||||
{ server = srv,
|
||||
rcvId,
|
||||
rcvPrivateKey,
|
||||
rcvSrvVerifyKey,
|
||||
rcvDhSecret,
|
||||
rcvDhSecret = C.dh' rcvPublicDHKey privDhKey,
|
||||
sndId = Just sndId,
|
||||
sndSrvVerifyKey,
|
||||
decryptKey,
|
||||
verifyKey = Nothing,
|
||||
status = New
|
||||
}
|
||||
pure (rq, SMPQueueUri srv sndId sndSrvVerifyKey, encryptKey)
|
||||
pure (rq, SMPQueueUri srv sndId, encryptKey)
|
||||
|
||||
subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
|
||||
subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
|
||||
|
||||
@@ -366,13 +366,13 @@ serializeAgentMessage = \case
|
||||
|
||||
-- | Serialize SMP queue information that is sent out-of-band.
|
||||
serializeSMPQueueUri :: SMPQueueUri -> ByteString
|
||||
serializeSMPQueueUri (SMPQueueUri srv qId _) =
|
||||
serializeSMPQueueUri (SMPQueueUri srv qId) =
|
||||
serializeServerUri srv <> "/" <> U.encode qId <> "#"
|
||||
|
||||
-- | SMP queue information parser.
|
||||
smpQueueUriP :: Parser SMPQueueUri
|
||||
smpQueueUriP =
|
||||
SMPQueueUri <$> smpServerUriP <* "/" <*> base64UriP <* "#" <*> pure reservedServerKey
|
||||
SMPQueueUri <$> smpServerUriP <* "/" <*> base64UriP <* optional "#"
|
||||
|
||||
reservedServerKey :: C.APublicVerifyKey
|
||||
reservedServerKey = C.APublicVerifyKey C.SRSA (C.PublicKeyRSA $ R.PublicKey 1 0 0)
|
||||
@@ -497,8 +497,7 @@ newtype AckMode = AckMode OnOff deriving (Eq, Show)
|
||||
-- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#out-of-band-messages
|
||||
data SMPQueueUri = SMPQueueUri
|
||||
{ smpServer :: SMPServer,
|
||||
senderId :: SMP.SenderId,
|
||||
serverVerifyKey :: C.APublicVerifyKey
|
||||
senderId :: SMP.SenderId
|
||||
}
|
||||
deriving (Eq, Show)
|
||||
|
||||
|
||||
@@ -24,7 +24,6 @@ import Simplex.Messaging.Protocol
|
||||
MsgId,
|
||||
RcvDhSecret,
|
||||
RcvPrivateSignKey,
|
||||
RcvPublicVerifyKey,
|
||||
SndPrivateSignKey,
|
||||
SndPublicVerifyKey,
|
||||
)
|
||||
@@ -80,14 +79,10 @@ data RcvQueue = RcvQueue
|
||||
rcvId :: SMP.RecipientId,
|
||||
-- | key used by the recipient to sign transmissions
|
||||
rcvPrivateKey :: RcvPrivateSignKey,
|
||||
-- | key used by the recipient to verify server transmissions
|
||||
rcvSrvVerifyKey :: RcvPublicVerifyKey,
|
||||
-- | shared DH secret used to encrypt/decrypt message bodies from server to recipient
|
||||
rcvDhSecret :: RcvDhSecret,
|
||||
-- | sender queue ID
|
||||
sndId :: Maybe SMP.SenderId,
|
||||
-- | key used by the sender to sign transmissions
|
||||
sndSrvVerifyKey :: SndPublicVerifyKey,
|
||||
-- | TODO keys used for E2E encryption - these will change with double ratchet
|
||||
decryptKey :: C.APrivateDecryptKey,
|
||||
verifyKey :: Maybe C.APublicVerifyKey,
|
||||
|
||||
@@ -625,19 +625,17 @@ insertRcvQueue_ dbConn connId RcvQueue {..} = do
|
||||
dbConn
|
||||
[sql|
|
||||
INSERT INTO rcv_queues
|
||||
( host, port, rcv_id, conn_alias, rcv_private_key, rcv_srv_verify_key, rcv_dh_secret, snd_id, snd_srv_verify_key, decrypt_key, verify_key, status)
|
||||
( host, port, rcv_id, conn_alias, rcv_private_key, rcv_dh_secret, snd_id, decrypt_key, verify_key, status)
|
||||
VALUES
|
||||
(:host,:port,:rcv_id,:conn_alias,:rcv_private_key,:rcv_srv_verify_key,:rcv_dh_secret,:snd_id,:snd_srv_verify_key,:decrypt_key,:verify_key,:status);
|
||||
(:host,:port,:rcv_id,:conn_alias,:rcv_private_key,:rcv_dh_secret,:snd_id,:decrypt_key,:verify_key,:status);
|
||||
|]
|
||||
[ ":host" := host server,
|
||||
":port" := port_,
|
||||
":rcv_id" := rcvId,
|
||||
":conn_alias" := connId,
|
||||
":rcv_private_key" := rcvPrivateKey,
|
||||
":rcv_srv_verify_key" := rcvSrvVerifyKey,
|
||||
":rcv_dh_secret" := rcvDhSecret,
|
||||
":snd_id" := sndId,
|
||||
":snd_srv_verify_key" := sndSrvVerifyKey,
|
||||
":decrypt_key" := decryptKey,
|
||||
":verify_key" := verifyKey,
|
||||
":status" := status
|
||||
@@ -733,25 +731,23 @@ getRcvQueueByConnAlias_ dbConn connId =
|
||||
<$> DB.query
|
||||
dbConn
|
||||
[sql|
|
||||
SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_srv_verify_key, q.rcv_dh_secret,
|
||||
q.snd_id, q.snd_srv_verify_key, q.decrypt_key, q.verify_key, q.status
|
||||
SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
|
||||
q.snd_id, q.decrypt_key, q.verify_key, q.status
|
||||
FROM rcv_queues q
|
||||
INNER JOIN servers s ON q.host = s.host AND q.port = s.port
|
||||
WHERE q.conn_alias = ?;
|
||||
|]
|
||||
(Only connId)
|
||||
where
|
||||
rcvQueue [(keyHash, host, port, rcvId, rcvPrivateKey, rcvSrvVerifyKey, rcvDhSecret, sndId, sndSrvVerifyKey, decryptKey, verifyKey, status)] =
|
||||
rcvQueue [(keyHash, host, port, rcvId, rcvPrivateKey, rcvDhSecret, sndId, decryptKey, verifyKey, status)] =
|
||||
let server = SMPServer host (deserializePort_ port) keyHash
|
||||
in Just $
|
||||
RcvQueue
|
||||
{ server,
|
||||
rcvId,
|
||||
rcvPrivateKey,
|
||||
rcvSrvVerifyKey,
|
||||
rcvDhSecret,
|
||||
sndId,
|
||||
sndSrvVerifyKey,
|
||||
decryptKey,
|
||||
verifyKey,
|
||||
status
|
||||
|
||||
@@ -64,7 +64,7 @@ import Numeric.Natural
|
||||
import Simplex.Messaging.Agent.Protocol (SMPServer (..))
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Protocol
|
||||
import Simplex.Messaging.Transport (ATransport (..), SessionId (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)
|
||||
import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)
|
||||
import Simplex.Messaging.Transport.WebSockets (WS)
|
||||
import Simplex.Messaging.Util (bshow, liftError, raceAny_)
|
||||
import System.Timeout (timeout)
|
||||
@@ -78,8 +78,7 @@ import System.Timeout (timeout)
|
||||
data SMPClient = SMPClient
|
||||
{ action :: Async (),
|
||||
connected :: TVar Bool,
|
||||
sndSessionId :: SessionId,
|
||||
rcvSessionId :: SessionId,
|
||||
sessionId :: ByteString,
|
||||
smpServer :: SMPServer,
|
||||
tcpTimeout :: Int,
|
||||
clientCorrId :: TVar Natural,
|
||||
@@ -149,8 +148,7 @@ getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing, smpBlock
|
||||
return
|
||||
SMPClient
|
||||
{ action = undefined,
|
||||
sndSessionId = undefined,
|
||||
rcvSessionId = undefined,
|
||||
sessionId = undefined,
|
||||
blockSize = undefined,
|
||||
connected,
|
||||
smpServer,
|
||||
@@ -171,8 +169,8 @@ getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing, smpBlock
|
||||
`finally` atomically (putTMVar thVar $ Left SMPNetworkError)
|
||||
bSize <- tcpTimeout `timeout` atomically (takeTMVar thVar)
|
||||
pure $ case bSize of
|
||||
Just (Right THandle {sndSessionId, rcvSessionId, blockSize}) ->
|
||||
Right c {action, sndSessionId, rcvSessionId, blockSize}
|
||||
Just (Right THandle {sessionId, blockSize}) ->
|
||||
Right c {action, sessionId, blockSize}
|
||||
Just (Left e) -> Left e
|
||||
Nothing -> Left SMPNetworkError
|
||||
|
||||
@@ -347,9 +345,9 @@ okSMPCommand cmd c pKey qId =
|
||||
|
||||
-- | Send any SMP command ('ClientCmd' type).
|
||||
sendSMPCommand :: SMPClient -> Maybe C.APrivateSignKey -> QueueId -> ClientCmd -> ExceptT SMPClientError IO (Command 'Broker)
|
||||
sendSMPCommand SMPClient {sndQ, sentCommands, clientCorrId, sndSessionId, tcpTimeout} pKey qId cmd = do
|
||||
sendSMPCommand SMPClient {sndQ, sentCommands, clientCorrId, sessionId, tcpTimeout} pKey qId cmd = do
|
||||
corrId <- lift_ getNextCorrId
|
||||
t <- signTransmission $ serializeTransmission sndSessionId (corrId, qId, cmd)
|
||||
t <- signTransmission $ serializeTransmission sessionId (corrId, qId, cmd)
|
||||
ExceptT $ sendRecv corrId t
|
||||
where
|
||||
lift_ :: STM a -> ExceptT SMPClientError IO a
|
||||
|
||||
@@ -96,7 +96,7 @@ import GHC.TypeLits (ErrorMessage (..), TypeError)
|
||||
import Generic.Random (genericArbitraryU)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Parsers
|
||||
import Simplex.Messaging.Transport (SessionId (..), THandle (..), Transport, TransportError (..), tGetEncrypted, tPutEncrypted)
|
||||
import Simplex.Messaging.Transport (THandle (..), Transport, TransportError (..), tGetEncrypted, tPutEncrypted)
|
||||
import Simplex.Messaging.Util
|
||||
import Test.QuickCheck (Arbitrary (..))
|
||||
|
||||
@@ -132,6 +132,8 @@ instance PartyI Notifier where sParty = SNotifier
|
||||
|
||||
data ClientParty = forall p. IsClient p => CP (SParty p)
|
||||
|
||||
deriving instance Show ClientParty
|
||||
|
||||
-- | Type for command or response of any participant.
|
||||
data Cmd = forall p. PartyI p => Cmd (SParty p) (Command p)
|
||||
|
||||
@@ -237,10 +239,8 @@ instance IsString CorrId where
|
||||
-- | Queue IDs and keys
|
||||
data QueueIdsKeys = QIK
|
||||
{ rcvId :: RecipientId,
|
||||
rcvSrvVerifyKey :: RcvPublicVerifyKey,
|
||||
rcvPublicDHKey :: RcvPublicDhKey,
|
||||
sndId :: SenderId,
|
||||
sndSrvVerifyKey :: SndPublicVerifyKey
|
||||
rcvPublicDHKey :: RcvPublicDhKey
|
||||
}
|
||||
deriving (Eq, Show)
|
||||
|
||||
@@ -358,13 +358,7 @@ instance CommandI Cmd where
|
||||
where
|
||||
newCmd = Cmd SRecipient <$> (NEW <$> C.strKeyP <* A.space <*> C.strKeyP)
|
||||
idsResp = Cmd SBroker . IDS <$> qik
|
||||
qik = do
|
||||
rcvId <- base64P <* A.space
|
||||
rcvSrvVerifyKey <- C.strKeyP <* A.space
|
||||
rcvPublicDHKey <- C.strKeyP <* A.space
|
||||
sndId <- base64P <* A.space
|
||||
sndSrvVerifyKey <- C.strKeyP
|
||||
pure QIK {rcvId, rcvSrvVerifyKey, rcvPublicDHKey, sndId, sndSrvVerifyKey}
|
||||
qik = QIK <$> base64P <* A.space <*> base64P <* A.space <*> C.strKeyP
|
||||
nIdsResp = Cmd SBroker . NID <$> base64P
|
||||
keyCmd = Cmd SRecipient . KEY <$> C.strKeyP
|
||||
nKeyCmd = Cmd SRecipient . NKEY <$> C.strKeyP
|
||||
@@ -411,8 +405,8 @@ instance PartyI p => CommandI (Command p) where
|
||||
NSUB -> "NSUB"
|
||||
MSG msgId ts msgBody ->
|
||||
B.unwords ["MSG", encode msgId, B.pack $ formatISO8601Millis ts, serializeBody msgBody]
|
||||
IDS QIK {rcvId, rcvSrvVerifyKey = rsKey, rcvPublicDHKey = dhKey, sndId, sndSrvVerifyKey = ssKey} ->
|
||||
B.unwords ["IDS", encode rcvId, C.serializeKey rsKey, C.serializeKey dhKey, encode sndId, C.serializeKey ssKey]
|
||||
IDS (QIK rcvId sndId srvDh) ->
|
||||
B.unwords ["IDS", encode rcvId, encode sndId, C.serializeKey srvDh]
|
||||
NID nId -> "NID " <> encode nId
|
||||
ERR err -> "ERR " <> serializeErrorType err
|
||||
NMSG -> "NMSG"
|
||||
@@ -435,9 +429,9 @@ serializeErrorType = bshow
|
||||
tPut :: Transport c => THandle c -> SentRawTransmission -> IO (Either TransportError ())
|
||||
tPut th (sig, t) = tPutEncrypted th $ C.serializeSignature sig <> " " <> serializeBody t
|
||||
|
||||
serializeTransmission :: CommandI c => SessionId -> Transmission c -> ByteString
|
||||
serializeTransmission (SessionId sessId) (CorrId corrId, queueId, command) =
|
||||
B.unwords [sessId, corrId, encode queueId, serializeCommand command]
|
||||
serializeTransmission :: CommandI c => ByteString -> Transmission c -> ByteString
|
||||
serializeTransmission sessionId (CorrId corrId, queueId, command) =
|
||||
B.unwords [sessionId, corrId, encode queueId, serializeCommand command]
|
||||
|
||||
-- | Validate that it is an SMP client command, used with 'tGet' by 'Simplex.Messaging.Server'.
|
||||
fromClient :: Cmd -> Either ErrorType ClientCmd
|
||||
@@ -460,12 +454,12 @@ tGetParse th = (first (const TEBadBlock) . A.parseOnly transmissionP =<<) <$> tG
|
||||
-- The first argument is used to limit allowed senders.
|
||||
-- 'fromClient' or 'fromServer' should be used here.
|
||||
tGet :: forall c m cmd. (Transport c, MonadIO m) => (Cmd -> Either ErrorType cmd) -> THandle c -> m (SignedTransmission cmd)
|
||||
tGet fromParty th@THandle {rcvSessionId} = liftIO (tGetParse th) >>= decodeParseValidate
|
||||
tGet fromParty th@THandle {sessionId} = liftIO (tGetParse th) >>= decodeParseValidate
|
||||
where
|
||||
decodeParseValidate :: Either TransportError RawTransmission -> m (SignedTransmission cmd)
|
||||
decodeParseValidate = \case
|
||||
Right RawTransmission {signature, signed, sessId, corrId, queueId, command}
|
||||
| SessionId sessId == rcvSessionId ->
|
||||
| sessId == sessionId ->
|
||||
let decodedTransmission = liftM2 (,corrId,,command) (C.decodeSignature =<< decode signature) (decode queueId)
|
||||
in either (const $ tError corrId) (tParseValidate signed) decodedTransmission
|
||||
| otherwise -> pure (Nothing, "", (CorrId corrId, "", Left SESSION))
|
||||
|
||||
@@ -106,7 +106,7 @@ runSMPServerBlocking started cfg@ServerConfig {transports} = do
|
||||
join <$> mapM (endPreviousSubscriptions qId) (M.lookup qId serverSubs)
|
||||
endPreviousSubscriptions :: QueueId -> Client -> STM (Maybe s)
|
||||
endPreviousSubscriptions qId c = do
|
||||
writeTBQueue (sndQ c) (Just $ CP SRecipient, (CorrId "", qId, END))
|
||||
writeTBQueue (sndQ c) (CorrId "", qId, END)
|
||||
stateTVar (clientSubs c) $ \ss -> (M.lookup qId ss, M.delete qId ss)
|
||||
|
||||
runClient :: (Transport c, MonadUnliftIO m, MonadReader Env m) => TProxy c -> c -> m ()
|
||||
@@ -118,9 +118,9 @@ runClient _ h = do
|
||||
Left _ -> pure ()
|
||||
|
||||
runClientTransport :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> m ()
|
||||
runClientTransport th@THandle {sndSessionId} = do
|
||||
runClientTransport th@THandle {sessionId} = do
|
||||
q <- asks $ tbqSize . config
|
||||
c <- atomically $ newClient q sndSessionId
|
||||
c <- atomically $ newClient q sessionId
|
||||
s <- asks server
|
||||
raceAny_ [send th c, client c s, receive th c]
|
||||
`finally` cancelSubscribers c
|
||||
@@ -138,41 +138,19 @@ receive :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> Cli
|
||||
receive th Client {rcvQ, sndQ} = forever $ do
|
||||
(sig, signed, (corrId, queueId, cmdOrError)) <- tGet fromClient th
|
||||
case cmdOrError of
|
||||
Left e -> write sndQ (Nothing, (corrId, queueId, ERR e))
|
||||
Left e -> write sndQ (corrId, queueId, ERR e)
|
||||
Right cmd -> do
|
||||
verified <- verifyTransmission sig signed queueId cmd
|
||||
if verified
|
||||
then write rcvQ (corrId, queueId, cmd)
|
||||
else write sndQ (Nothing, (corrId, queueId, ERR AUTH))
|
||||
else write sndQ (corrId, queueId, ERR AUTH)
|
||||
where
|
||||
write q t = atomically $ writeTBQueue q t
|
||||
|
||||
send :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> Client -> m ()
|
||||
send h Client {sndQ, sndSessionId} = forever $ do
|
||||
atomically (readTBQueue sndQ)
|
||||
>>= signTransmission sndSessionId
|
||||
>>= liftIO . tPut h
|
||||
|
||||
signTransmission ::
|
||||
forall m. (MonadUnliftIO m, MonadReader Env m) => SessionId -> (Maybe ClientParty, BrokerTransmission) -> m SentRawTransmission
|
||||
signTransmission sndSessionId (party_, t@(_, queueId, cmd)) =
|
||||
case party_ of
|
||||
Nothing -> unsigned
|
||||
Just (CP SNotifier) -> unsigned
|
||||
Just party ->
|
||||
case cmd of
|
||||
ERR QUOTA -> signed party
|
||||
ERR _ -> unsigned
|
||||
PONG -> unsigned
|
||||
_ -> signed party
|
||||
where
|
||||
s = serializeTransmission sndSessionId t
|
||||
unsigned = pure (Nothing, s)
|
||||
signed :: ClientParty -> m SentRawTransmission
|
||||
signed party = do
|
||||
st <- asks queueStore
|
||||
q <- atomically $ getQueue st party queueId
|
||||
pure (Nothing, s)
|
||||
send :: (Transport c, MonadUnliftIO m) => THandle c -> Client -> m ()
|
||||
send h Client {sndQ, sessionId} = forever $ do
|
||||
t <- atomically $ readTBQueue sndQ
|
||||
liftIO $ tPut h (Nothing, serializeTransmission sessionId t)
|
||||
|
||||
verifyTransmission ::
|
||||
forall m. (MonadUnliftIO m, MonadReader Env m) => Maybe C.ASignature -> ByteString -> QueueId -> ClientCmd -> m Bool
|
||||
@@ -239,16 +217,17 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
|
||||
>>= processCommand
|
||||
>>= atomically . writeTBQueue sndQ
|
||||
where
|
||||
processCommand :: Transmission ClientCmd -> m (Maybe ClientParty, BrokerTransmission)
|
||||
processCommand :: Transmission ClientCmd -> m BrokerTransmission
|
||||
processCommand (corrId, queueId, cmd) = do
|
||||
st <- asks queueStore
|
||||
case cmd of
|
||||
ClientCmd SSender command -> case command of
|
||||
SEND msgBody -> (Just $ CP SSender,) <$> sendMessage st msgBody
|
||||
PING -> pure (Nothing, (corrId, queueId, PONG))
|
||||
ClientCmd SNotifier NSUB -> (Just $ CP SNotifier,) <$> subscribeNotifications
|
||||
ClientCmd SSender command ->
|
||||
case command of
|
||||
SEND msgBody -> sendMessage st msgBody
|
||||
PING -> pure (corrId, "", PONG)
|
||||
ClientCmd SNotifier NSUB -> subscribeNotifications
|
||||
ClientCmd SRecipient command ->
|
||||
(Just $ CP SRecipient,) <$> case command of
|
||||
case command of
|
||||
NEW rKey dhKey -> createQueue st rKey dhKey
|
||||
SUB -> subscribeQueue queueId
|
||||
ACK -> acknowledgeMsg
|
||||
@@ -259,21 +238,16 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
|
||||
where
|
||||
createQueue :: QueueStore -> RcvPublicVerifyKey -> RcvPublicDhKey -> m BrokerTransmission
|
||||
createQueue st recipientKey dhKey = checkKeySize recipientKey $ do
|
||||
C.SignAlg a <- asks $ trnSignAlg . config
|
||||
(rcvPublicDHKey, privDhKey) <- liftIO $ C.generateKeyPair' 0
|
||||
(rcvSrvVerifyKey, rcvSrvSignKey) <- liftIO $ C.generateSignatureKeyPair 0 a
|
||||
(sndSrvVerifyKey, sndSrvSignKey) <- liftIO $ C.generateSignatureKeyPair 0 a
|
||||
let rcvDhSecret = C.dh' dhKey privDhKey
|
||||
qik (rcvId, sndId) = QIK {rcvId, rcvSrvVerifyKey, rcvPublicDHKey, sndId, sndSrvVerifyKey}
|
||||
qik (rcvId, sndId) = QIK {rcvId, sndId, rcvPublicDHKey}
|
||||
qRec (recipientId, senderId) =
|
||||
QueueRec
|
||||
{ recipientId,
|
||||
senderId,
|
||||
recipientKey,
|
||||
rcvSrvSignKey,
|
||||
rcvDhSecret,
|
||||
senderKey = Nothing,
|
||||
sndSrvSignKey,
|
||||
notifier = Nothing,
|
||||
status = QueueActive
|
||||
}
|
||||
@@ -401,7 +375,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
|
||||
writeNtf :: NotifierId -> Client -> STM ()
|
||||
writeNtf nId Client {sndQ = q} =
|
||||
unlessM (isFullTBQueue sndQ) $
|
||||
writeTBQueue q (Just $ CP SNotifier, (CorrId "", nId, NMSG))
|
||||
writeTBQueue q (CorrId "", nId, NMSG)
|
||||
|
||||
deliverMessage :: (MsgQueue -> STM (Maybe Message)) -> RecipientId -> Sub -> m BrokerTransmission
|
||||
deliverMessage tryPeek rId = \case
|
||||
@@ -425,7 +399,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
|
||||
subscriber :: MsgQueue -> m ()
|
||||
subscriber q = atomically $ do
|
||||
msg <- peekMsg q
|
||||
writeTBQueue sndQ (Just $ CP SRecipient, (CorrId "", rId, msgCmd msg))
|
||||
writeTBQueue sndQ (CorrId "", rId, msgCmd msg)
|
||||
setSub (\s -> s {subThread = NoSub})
|
||||
void setDelivered
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ module Simplex.Messaging.Server.Env.STM where
|
||||
import Control.Concurrent (ThreadId)
|
||||
import Control.Monad.IO.Unlift
|
||||
import Crypto.Random
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import Data.Map.Strict (Map)
|
||||
import qualified Data.Map.Strict as M
|
||||
import Network.Socket (ServiceName)
|
||||
@@ -19,7 +20,7 @@ import Simplex.Messaging.Server.MsgStore.STM
|
||||
import Simplex.Messaging.Server.QueueStore (QueueRec (..))
|
||||
import Simplex.Messaging.Server.QueueStore.STM
|
||||
import Simplex.Messaging.Server.StoreLog
|
||||
import Simplex.Messaging.Transport (ATransport, SessionId, loadTLSServerParams)
|
||||
import Simplex.Messaging.Transport (ATransport, loadTLSServerParams)
|
||||
import System.IO (IOMode (..))
|
||||
import UnliftIO.STM
|
||||
|
||||
@@ -33,8 +34,7 @@ data ServerConfig = ServerConfig
|
||||
blockSize :: Int,
|
||||
serverPrivateKey :: C.PrivateKey 'C.RSA, -- TODO delete
|
||||
serverPrivateKeyFile :: FilePath,
|
||||
serverCertificateFile :: FilePath,
|
||||
trnSignAlg :: C.SignAlg
|
||||
serverCertificateFile :: FilePath
|
||||
}
|
||||
|
||||
data Env = Env
|
||||
@@ -59,8 +59,8 @@ data Client = Client
|
||||
{ subscriptions :: TVar (Map RecipientId Sub),
|
||||
ntfSubscriptions :: TVar (Map NotifierId ()),
|
||||
rcvQ :: TBQueue (Transmission ClientCmd),
|
||||
sndQ :: TBQueue (Maybe ClientParty, BrokerTransmission),
|
||||
sndSessionId :: SessionId
|
||||
sndQ :: TBQueue BrokerTransmission,
|
||||
sessionId :: ByteString
|
||||
}
|
||||
|
||||
data SubscriptionThread = NoSub | SubPending | SubThread ThreadId
|
||||
@@ -78,13 +78,13 @@ newServer qSize = do
|
||||
notifiers <- newTVar M.empty
|
||||
return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers}
|
||||
|
||||
newClient :: Natural -> SessionId -> STM Client
|
||||
newClient qSize sndSessionId = do
|
||||
newClient :: Natural -> ByteString -> STM Client
|
||||
newClient qSize sessionId = do
|
||||
subscriptions <- newTVar M.empty
|
||||
ntfSubscriptions <- newTVar M.empty
|
||||
rcvQ <- newTBQueue qSize
|
||||
sndQ <- newTBQueue qSize
|
||||
return Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sndSessionId}
|
||||
return Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId}
|
||||
|
||||
newSubscription :: STM Sub
|
||||
newSubscription = do
|
||||
|
||||
@@ -8,22 +8,20 @@ import Simplex.Messaging.Protocol
|
||||
|
||||
data QueueRec = QueueRec
|
||||
{ recipientId :: RecipientId,
|
||||
senderId :: SenderId,
|
||||
recipientKey :: RcvPublicVerifyKey,
|
||||
rcvSrvSignKey :: RcvPrivateSignKey,
|
||||
rcvDhSecret :: RcvDhSecret,
|
||||
senderId :: SenderId,
|
||||
senderKey :: Maybe SndPublicVerifyKey,
|
||||
sndSrvSignKey :: SndPrivateSignKey,
|
||||
notifier :: Maybe (NotifierId, NtfPublicVerifyKey),
|
||||
status :: QueueStatus
|
||||
}
|
||||
|
||||
data QueueStatus = QueueActive | QueueOff deriving (Eq)
|
||||
data QueueStatus = QueueActive | QueueOff deriving (Eq, Show)
|
||||
|
||||
class MonadQueueStore s m where
|
||||
addQueue :: s -> QueueRec -> m (Either ErrorType ())
|
||||
getQueue :: s -> ClientParty -> QueueId -> m (Either ErrorType QueueRec)
|
||||
secureQueue :: s -> RecipientId -> SndPublicVerifyKey -> m (Either ErrorType ())
|
||||
addQueueNotifier :: s -> RecipientId -> NotifierId -> NtfPublicVerifyKey -> m (Either ErrorType ())
|
||||
secureQueue :: s -> RecipientId -> SndPublicVerifyKey -> m (Either ErrorType QueueRec)
|
||||
addQueueNotifier :: s -> RecipientId -> NotifierId -> NtfPublicVerifyKey -> m (Either ErrorType QueueRec)
|
||||
suspendQueue :: s -> RecipientId -> m (Either ErrorType ())
|
||||
deleteQueue :: s -> RecipientId -> m (Either ErrorType ())
|
||||
@@ -59,14 +59,14 @@ instance MonadQueueStore QueueStore STM where
|
||||
Just rId -> getRcpQueue cs rId
|
||||
Nothing -> Left AUTH
|
||||
|
||||
secureQueue :: QueueStore -> RecipientId -> SndPublicVerifyKey -> STM (Either ErrorType ())
|
||||
secureQueue :: QueueStore -> RecipientId -> SndPublicVerifyKey -> STM (Either ErrorType QueueRec)
|
||||
secureQueue store rId sKey =
|
||||
updateQueues store rId $ \cs c ->
|
||||
case senderKey c of
|
||||
Just _ -> (Left AUTH, cs)
|
||||
_ -> (Right (), cs {queues = M.insert rId c {senderKey = Just sKey} (queues cs)})
|
||||
_ -> (Right c, cs {queues = M.insert rId c {senderKey = Just sKey} (queues cs)})
|
||||
|
||||
addQueueNotifier :: QueueStore -> RecipientId -> NotifierId -> NtfPublicVerifyKey -> STM (Either ErrorType ())
|
||||
addQueueNotifier :: QueueStore -> RecipientId -> NotifierId -> NtfPublicVerifyKey -> STM (Either ErrorType QueueRec)
|
||||
addQueueNotifier store rId nId nKey = do
|
||||
cs@QueueStoreData {queues, notifiers} <- readTVar store
|
||||
if M.member nId notifiers
|
||||
@@ -81,7 +81,7 @@ instance MonadQueueStore QueueStore STM where
|
||||
{ queues = M.insert rId q {notifier = Just (nId, nKey)} queues,
|
||||
notifiers = M.insert nId rId notifiers
|
||||
}
|
||||
pure $ Right ()
|
||||
pure $ Right q
|
||||
|
||||
suspendQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
|
||||
suspendQueue store rId =
|
||||
@@ -101,8 +101,8 @@ instance MonadQueueStore QueueStore STM where
|
||||
updateQueues ::
|
||||
QueueStore ->
|
||||
RecipientId ->
|
||||
(QueueStoreData -> QueueRec -> (Either ErrorType (), QueueStoreData)) ->
|
||||
STM (Either ErrorType ())
|
||||
(QueueStoreData -> QueueRec -> (Either ErrorType a, QueueStoreData)) ->
|
||||
STM (Either ErrorType a)
|
||||
updateQueues store rId update = do
|
||||
cs <- readTVar store
|
||||
let conn = getRcpQueue cs rId
|
||||
|
||||
@@ -68,13 +68,11 @@ storeLogRecordP =
|
||||
queueRecP = do
|
||||
recipientId <- "rid=" *> base64P
|
||||
recipientKey <- " rk=" *> C.strKeyP
|
||||
rcvSrvSignKey <- " rsk=" *> C.strKeyP
|
||||
rcvDhSecret <- " rdh=" *> C.strDhSecretP
|
||||
senderId <- " sid=" *> base64P
|
||||
senderKey <- " sk=" *> optional C.strKeyP
|
||||
sndSrvSignKey <- " ssk=" *> C.strKeyP
|
||||
notifier <- optional $ (,) <$> (" nid=" *> base64P) <*> (" nk=" *> C.strKeyP)
|
||||
pure QueueRec {recipientId, recipientKey, rcvSrvSignKey, rcvDhSecret, senderId, senderKey, sndSrvSignKey, notifier, status = QueueActive}
|
||||
pure QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, notifier, status = QueueActive}
|
||||
|
||||
serializeStoreLogRecord :: StoreLogRecord -> ByteString
|
||||
serializeStoreLogRecord = \case
|
||||
@@ -84,15 +82,13 @@ serializeStoreLogRecord = \case
|
||||
DeleteQueue rId -> "DELETE " <> encode rId
|
||||
where
|
||||
serializeQueue
|
||||
QueueRec {recipientId, recipientKey, rcvSrvSignKey, rcvDhSecret, senderId, senderKey, sndSrvSignKey, notifier} =
|
||||
QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, notifier} =
|
||||
B.unwords
|
||||
[ "rid=" <> encode recipientId,
|
||||
"rk=" <> C.serializeKey recipientKey,
|
||||
"rsk=" <> C.serializeKey rcvSrvSignKey,
|
||||
"rdh=" <> C.serializeDhSecret rcvDhSecret,
|
||||
"sid=" <> encode senderId,
|
||||
"sk=" <> maybe "" C.serializeKey senderKey,
|
||||
"ssk=" <> C.serializeKey sndSrvSignKey
|
||||
"sk=" <> maybe "" C.serializeKey senderKey
|
||||
]
|
||||
<> maybe "" serializeNotifier notifier
|
||||
serializeNotifier (nId, nKey) = " nid=" <> encode nId <> " nk=" <> C.serializeKey nKey
|
||||
|
||||
@@ -41,7 +41,6 @@ module Simplex.Messaging.Transport
|
||||
|
||||
-- * SMP encrypted transport
|
||||
THandle (..),
|
||||
SessionId (..),
|
||||
TransportError (..),
|
||||
serverHandshake,
|
||||
clientHandshake,
|
||||
@@ -328,16 +327,10 @@ smpVersionP =
|
||||
let ver = A.decimal <* A.char '.'
|
||||
in SMPVersion <$> ver <*> ver <*> ver <*> A.decimal
|
||||
|
||||
-- | Session identifier (base64 encoded here, to avoid encoding every time it is sent)
|
||||
-- It should be set from TLS finished and passed in the initial handshake
|
||||
newtype SessionId = SessionId {unSessionId :: ByteString}
|
||||
deriving (Eq, Show)
|
||||
|
||||
-- | The handle for SMP encrypted transport connection over Transport .
|
||||
data THandle c = THandle
|
||||
{ connection :: c,
|
||||
sndSessionId :: SessionId,
|
||||
rcvSessionId :: SessionId,
|
||||
sessionId :: ByteString,
|
||||
sndKey :: SessionKey,
|
||||
rcvKey :: SessionKey,
|
||||
blockSize :: Int
|
||||
@@ -596,8 +589,7 @@ transportHandle c sk rk blockSize = do
|
||||
pure
|
||||
THandle
|
||||
{ connection = c,
|
||||
sndSessionId = SessionId "",
|
||||
rcvSessionId = SessionId "",
|
||||
sessionId = "",
|
||||
sndKey = sk {counter = sndCounter},
|
||||
rcvKey = rk {counter = rcvCounter},
|
||||
blockSize
|
||||
|
||||
Reference in new issue
Block a user