expect DOWN

* xftp: negotiate protocol with ALPN

* add RFC

* add handshake implementation

* implement extended handshake

* enable authentication

* update rfc

* Apply suggestions from code review

Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>

* cleanup

* discard trailing data

* cleanup diff

* use find

* rename

* refactor

* add x509 tests

---------

Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>

cabal.project

@@ -14,12 +14,6 @@ source-repository-package
     location: https://github.com/simplex-chat/aeson.git
     tag: aab7b5a14d6c5ea64c64dcaee418de1bb00dcc2b
 
--- old bs/text compat for 8.10
-source-repository-package
-    type: git
-    location: https://github.com/simplex-chat/base64.git
-    tag: 2d77b6dbcaffc00570a70be8694049f3710e7c94
-
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/hs-socks.git

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.6.2.0
+version: 5.6.2.1
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -31,7 +31,7 @@ dependencies:
   - async == 2.2.*
   - attoparsec == 0.14.*
   - base >= 4.14 && < 5
-  - base64 == 1.0.*
+  - base64-bytestring >= 1.0 && < 1.3
   - case-insensitive == 1.2.*
   - composition == 1.0.*
   - constraints >= 0.12 && < 0.14
@@ -114,45 +114,30 @@ executables:
     main: Main.hs
     dependencies:
       - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
 
   ntf-server:
     source-dirs: apps/ntf-server
     main: Main.hs
     dependencies:
       - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
 
   xftp-server:
     source-dirs: apps/xftp-server
     main: Main.hs
     dependencies:
       - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
 
   smp-agent:
     source-dirs: apps/smp-agent
     main: Main.hs
     dependencies:
       - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
 
   xftp:
     source-dirs: apps/xftp
     main: Main.hs
     dependencies:
       - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
 
 tests:
   simplexmq-test:
@@ -180,6 +165,8 @@ ghc-options:
   - -Wincomplete-uni-patterns
   - -Wunused-type-patterns
   - -O2
+  - -threaded
+  - -rtsopts
 
 default-extensions:
   - StrictData

rfcs/2024-03-28-xftp-version.md

@@ -0,0 +1,120 @@
+# XFTP version agreement
+
+## Problem
+
+XFTP is using HTTP2 protocol for encoding requests and responses.
+Unlike SMP which has a connection handshake initiated by a server and signals available versions XFTP/HTTP2 is almost entirely client-driven.
+So, a client can only try to guess which protocol versions are supported by a server by sending a probe/hello request first.
+Determining the endpoint for such a request is an implicit version agreement by itself.
+Sending such a request to an old server would error out and requiring it from old clients would break them.
+
+## Solution
+
+The TLS layer used by the XFTP server has an optional [ALPN](https://datatracker.ietf.org/doc/html/rfc7301) extension which allows the client and server to negotiate protocols and store the decision in TLS session context.
+Unless a client and a server run ALPN-aware versions, they would default to the old "unversioned" protocol.
+
+TLS extension content is a 65kb chunk, but ALPN standard breaks it into 254b-sized chunks making it unusable for things like key exchange.
+The exchange is still client-driven: the client proposes a list, and then a server callback picks one.
+In effect, this makes it usable only to signal that some application-level handshake is desired and supported.
+
+## Implementation
+
+ALPN can be used to negotiate for any TLS-based protocol, but the description will focus on XFTP.
+
+TransportClientConfig gets a new `alpn :: Maybe [ALPN]` field so a TLS transport can use it during TLS client creation.
+XFTP client sets it to `Just ["xftp/1"]`.
+The exact value is not important as long it is in agreement with the server side, but ALPN RFC insists on it being an IANA-registered identifier.
+
+XFTP server sets `onALPNClientSuggest` TLS hook to pick the protocol when it is provided.
+The `tls` library treats SHOULD from the RFC as MUST and does a client-side check that the server responded with one of the client-proposed protocols.
+
+Upon connection, transport implementation invokes `getNegotiatedProtocol` and stores it in `tlsALPN :: Maybe ALPN` field of transport context.
+HTTP2 transport implementation using `withHTTP2` passes negotiated "protocol" to client and server setup callbacks where they store it in their respective wrappers along with TLS session ID.
+A server request handler then knows by looking at the `sessionALPN` if it should require a "handshake" request first.
+A client code that got HTTP2Client with `sessionALPN` set knows if it has to proceed with handshake request.
+A handshake request still has to be initiated by a client, so it should be kept minimal, just enough data to pass the initiative to a server.
+A reply to that initial request should contain a server version range for the client to pick.
+A client then commits to a version, sending its part of a handshake.
+
+In the future ALPN negotiation can be dropped in favor of mandatory handshakes or used to signal further handshake schemes.
+
+The XFTP handshake data types and validation code are cloned from SMP.
+Currently they carry version information and session authentication parameters.
+Authentication parameters made mandatory as this exchange is guarded by the handshake version.
+
+### Server side
+
+`runHTTP2Server` callback used by `xftpServer` should get access to the session state to track handshakes.
+A local `TMap SessionId Handshake` is enough to switch request handlers.
+The HTTP2 server framework is extended with a way to signal client disconnection to remove sessions from this map.
+
+The `Handshake` type mimics implicit state in stream based handshakes of SMP and NTF.
+
+```haskell
+data Handshake
+  = HandshakeSent C.PrivateKeyX25519 -- server private key that will be merged with client public in `THandleAuth`
+  | HandshakeAccepted THandleAuth VersionXFTP -- session steady state after handshakes
+```
+
+An HTTP2 request without ALPN is treated as legacy and requires no session entry.
+Its `Request`s are marked with `THandleParams {thVersion = VersionXFTP 1, ..}`.
+
+An HTTP2 request with ALPN requires a session lookup.
+- A lack of entry indicates that a client must send an empty request, to which the server replies with its "server handshake" block and stores its private state in `HandshakeSent`.
+- If the session entry contains `HandshakeSent`, then the only valid request content is the "client handshake" block.
+  The server validates client handshake (in the same way as SMP) and stores authentication and version in `HandshakeAccepted`
+- If the session entry contains `HandshakeAccepted`, then the server just passes it to `THandleParams`.
+
+### Client side
+
+`getXFTPClient` tweaks its transport config to include the ALPN marker and then checks if the client got its `sessionALPN` value.
+If there's a value set, it then sends an initial block and checks out the server handshake in response.
+After validation, it sends "client handshake" request to finish version negotiation.
+
+```haskell
+let tcConfig = (transportClientConfig xftpNetworkConfig) {alpn = Just ["xftp/1"]}
+-- ...
+http2Client <- liftEitherError xftpClientError $ getVerifiedHTTP2Client -- ...
+thVersion <- case sessionALPN http2Client of
+  Nothing -> pure $ VersionXFTP 1
+  Just proto -> negotiate http2Client proto
+```
+
+The resulting `XFTPClient` then contains a negotiated version and can be used to send transmissions with a more recent encoding.
+
+## Block encoding
+
+### Client Hello (request)
+
+A request with an empty body and no padding.
+
+### Server handshake (response)
+
+SMP-encoded and padded to `xftpBlockSize` (~16kb).
+
+```haskell
+data XFTPServerHandshake = XFTPServerHandshake
+  { xftpVersionRange :: VersionRangeXFTP,
+    sessionId :: SessionId, -- validated by client against TLS unique
+    authPubKey ::
+      ( X.CertificateChain, -- fingerprint validated by client against pre-shared hash
+        X.SignedExact X.PubKey -- signature validated by client against server key from TLS
+      )
+  }
+```
+
+### Client handshake (request)
+
+SMP-encoded and padded to `xftpBlockSize` (~16kb).
+
+```haskell
+data XFTPClientHandshake = XFTPClientHandshake
+  { xftpVersion :: VersionXFTP,
+    keyHash :: C.KeyHash, -- validated by server against its own cert fingerprint
+    authPubKey :: C.PublicKeyX25519
+  }
+```
+
+### Server confirmation (response)
+
+A response with an empty body and no padding.

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.6.2.0
+version:        5.6.2.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -119,8 +119,6 @@ library
       Simplex.Messaging.Crypto.SNTRUP761.Bindings.FFI
       Simplex.Messaging.Crypto.SNTRUP761.Bindings.RNG
       Simplex.Messaging.Encoding
-      Simplex.Messaging.Encoding.Base64
-      Simplex.Messaging.Encoding.Base64.URL
       Simplex.Messaging.Encoding.String
       Simplex.Messaging.Notifications.Client
       Simplex.Messaging.Notifications.Protocol
@@ -175,7 +173,7 @@ library
       src
   default-extensions:
       StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
   include-dirs:
       cbits
   c-sources:
@@ -191,7 +189,7 @@ library
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
-    , base64 ==1.0.*
+    , base64-bytestring >=1.0 && <1.3
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
@@ -265,7 +263,7 @@ executable ntf-server
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
-    , base64 ==1.0.*
+    , base64-bytestring >=1.0 && <1.3
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
@@ -340,7 +338,7 @@ executable smp-agent
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
-    , base64 ==1.0.*
+    , base64-bytestring >=1.0 && <1.3
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
@@ -415,7 +413,7 @@ executable smp-server
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
-    , base64 ==1.0.*
+    , base64-bytestring >=1.0 && <1.3
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
@@ -490,7 +488,7 @@ executable xftp
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
-    , base64 ==1.0.*
+    , base64-bytestring >=1.0 && <1.3
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
@@ -565,7 +563,7 @@ executable xftp-server
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
-    , base64 ==1.0.*
+    , base64-bytestring >=1.0 && <1.3
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
@@ -663,7 +661,7 @@ test-suite simplexmq-test
       tests
   default-extensions:
       StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
       HUnit ==1.6.*
     , QuickCheck ==2.14.*
@@ -674,7 +672,7 @@ test-suite simplexmq-test
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
-    , base64 ==1.0.*
+    , base64-bytestring >=1.0 && <1.3
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14

src/Simplex/FileTransfer/Agent.hs

@@ -70,7 +70,7 @@ import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String (strDecode, strEncode)
 import Simplex.Messaging.Protocol (EntityId, XFTPServer)
-import Simplex.Messaging.Util (catchAll_, liftError, tshow, unlessM, whenM)
+import Simplex.Messaging.Util (catchAll_, liftError, tshow, unlessM, whenM, atomically')
 import System.FilePath (takeFileName, (</>))
 import UnliftIO
 import UnliftIO.Directory
@@ -110,7 +110,7 @@ closeXFTPAgent a = do
   stopWorkers $ xftpSndWorkers a
   stopWorkers $ xftpDelWorkers a
   where
-    stopWorkers workers = atomically (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
+    stopWorkers workers = atomically' (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
 
 xftpReceiveFile' :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> AM RcvFileId
 xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redirect}) cfArgs = do
@@ -131,7 +131,7 @@ xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redi
           relSavePathRedirect = relPrefixPath </> "xftp.redirect-decrypted"
       lift $ createDirectory =<< toFSFilePath relTmpPathRedirect
       lift $ createEmptyFile =<< toFSFilePath relSavePathRedirect
-      cfArgsRedirect <- atomically $ CF.randomArgs g
+      cfArgsRedirect <- atomically' $ CF.randomArgs g
       let saveFileRedirect = CryptoFile relSavePathRedirect $ Just cfArgsRedirect
       -- create download tasks
       withStore c $ \db -> createRcvFileRedirect db g userId fd relPrefixPath relTmpPathRedirect saveFileRedirect relTmpPath saveFile
@@ -170,7 +170,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
   cfg <- asks config
   forever $ do
     lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
     runXFTPOperation cfg
   where
     runXFTPOperation :: AgentConfig -> AM ()
@@ -188,7 +188,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
                 when notifyOnRetry $ notify c rcvFileEntityId $ RFERR e
                 liftIO $ closeXFTPServerClient c userId server digest
                 withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay
-              atomically $ assertAgentForeground c
+              atomically' $ assertAgentForeground c
               loop
             retryDone e = rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) (show e)
     downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> AM ()
@@ -198,7 +198,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
       let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
           relChunkPath = fileTmpPath </> takeFileName chunkPath
       agentXFTPDownloadChunk c userId digest replica chunkSpec
-      atomically $ waitUntilForeground c
+      atomically' $ waitUntilForeground c
       (entityId, complete, progress) <- withStore c $ \db -> runExceptT $ do
         liftIO $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId relChunkPath
         RcvFile {size = FileSize currentSize, chunks, redirect} <- ExceptT $ getRcvFile db rcvFileId
@@ -244,7 +244,7 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
   cfg <- asks config
   forever $ do
     lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
     runXFTPOperation cfg
   where
     runXFTPOperation :: AgentConfig -> AM ()
@@ -270,12 +270,12 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
         Nothing -> do
           notify c rcvFileEntityId $ RFDONE fsSavePath
           lift $ forM_ tmpPath (removePath <=< toFSFilePath)
-          atomically $ waitUntilForeground c
+          atomically' $ waitUntilForeground c
           withStore' c (`updateRcvFileComplete` rcvFileId)
         Just RcvFileRedirect {redirectFileInfo, redirectDbId} -> do
           let RedirectFileInfo {size = redirectSize, digest = redirectDigest} = redirectFileInfo
           lift $ forM_ tmpPath (removePath <=< toFSFilePath)
-          atomically $ waitUntilForeground c
+          atomically' $ waitUntilForeground c
           withStore' c (`updateRcvFileComplete` rcvFileId)
           -- proceed with redirect
           yaml <- liftError (INTERNAL . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `agentFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
@@ -339,7 +339,7 @@ xftpSendDescription' c userId (ValidFileDescription fdDirect@FileDescription {si
   createDirectory prefixPath
   let relPrefixPath = takeFileName prefixPath
   let directYaml = prefixPath </> "direct.yaml"
-  cfArgs <- atomically $ CF.randomArgs g
+  cfArgs <- atomically' $ CF.randomArgs g
   let file = CryptoFile directYaml (Just cfArgs)
   liftError (INTERNAL . show) $ CF.writeFile file (LB.fromStrict $ strEncode fdDirect)
   key <- atomically $ C.randomSbKey g
@@ -362,7 +362,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
   cfg <- asks config
   forever $ do
     lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
     runXFTPOperation cfg
   where
     runXFTPOperation :: AgentConfig -> AM ()
@@ -415,7 +415,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
           pure (FileDigest digest, zip chunkSpecs $ coerce chunkDigests)
         createChunk :: Int -> SndFileChunk -> AM ()
         createChunk numRecipients' ch = do
-          atomically $ assertAgentForeground c
+          atomically' $ assertAgentForeground c
           (replica, ProtoServerWithAuth srv _) <- tryCreate
           withStore' c $ \db -> createSndFileReplica db ch replica
           lift . void $ getXFTPSndWorker True c (Just srv)
@@ -426,7 +426,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
                 createWithNextSrv usedSrvs
                   `catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
               where
-                retryLoop loop = atomically (assertAgentForeground c) >> loop
+                retryLoop loop = atomically' (assertAgentForeground c) >> loop
             createWithNextSrv usedSrvs = do
               deleted <- withStore' c $ \db -> getSndFileDeleted db sndFileId
               when deleted $ throwError $ INTERNAL "file deleted, aborting chunk creation"
@@ -445,7 +445,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
   cfg <- asks config
   forever $ do
     lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
     runXFTPOperation cfg
   where
     runXFTPOperation :: AgentConfig -> AM ()
@@ -463,7 +463,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
                 when notifyOnRetry $ notify c sndFileEntityId $ SFERR e
                 liftIO $ closeXFTPServerClient c userId server digest
                 withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay
-              atomically $ assertAgentForeground c
+              atomically' $ assertAgentForeground c
               loop
             retryDone e = sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) (show e)
     uploadFileChunk :: AgentConfig -> SndFileChunk -> SndFileChunkReplica -> AM ()
@@ -472,9 +472,9 @@ runXFTPSndWorker c srv Worker {doWork} = do
       fsFilePath <- lift $ toFSFilePath filePath
       unlessM (doesFileExist fsFilePath) $ throwError $ INTERNAL "encrypted file doesn't exist on upload"
       let chunkSpec' = chunkSpec {filePath = fsFilePath} :: XFTPChunkSpec
-      atomically $ assertAgentForeground c
+      atomically' $ assertAgentForeground c
       agentXFTPUploadChunk c userId chunkDigest replica' chunkSpec'
-      atomically $ waitUntilForeground c
+      atomically' $ waitUntilForeground c
       sf@SndFile {sndFileEntityId, prefixPath, chunks} <- withStore c $ \db -> do
         updateSndChunkReplicaStatus db sndChunkReplicaId SFRSUploaded
         getSndFile db sndFileId
@@ -610,7 +610,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
   cfg <- asks config
   forever $ do
     lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
     runXFTPOperation cfg
   where
     runXFTPOperation :: AgentConfig -> AM ()
@@ -629,7 +629,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
                 when notifyOnRetry $ notify c "" $ SFERR e
                 liftIO $ closeXFTPServerClient c userId server chunkDigest
                 withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay
-              atomically $ assertAgentForeground c
+              atomically' $ assertAgentForeground c
               loop
             retryDone = delWorkerInternalError c deletedSndChunkReplicaId
             deleteChunkReplica = do

src/Simplex/FileTransfer/Client.hs

@@ -4,11 +4,14 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 
 module Simplex.FileTransfer.Client where
 
+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Crypto.Random (ChaChaDRG)
@@ -20,6 +23,8 @@ import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty (..))
 import Data.Time (UTCTime)
 import Data.Word (Word32)
+import qualified Data.X509 as X
+import qualified Data.X509.Validation as XV
 import qualified Network.HTTP.Types as N
 import qualified Network.HTTP2.Client as H
 import Simplex.FileTransfer.Description (mb)
@@ -37,6 +42,7 @@ import Simplex.Messaging.Client
 import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
+import Simplex.Messaging.Encoding (smpDecode, smpEncode)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
   ( BasicAuth,
@@ -45,12 +51,13 @@ import Simplex.Messaging.Protocol
     RecipientId,
     SenderId,
   )
-import Simplex.Messaging.Transport (THandleParams (..), supportedParameters)
-import Simplex.Messaging.Transport.Client (TransportClientConfig, TransportHost)
+import Simplex.Messaging.Transport (HandshakeError (VERSION), THandleAuth (..), THandleParams (..), TransportError (..), supportedParameters)
+import Simplex.Messaging.Transport.Client (TransportClientConfig, TransportHost, alpn)
 import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Transport.HTTP2.Client
 import Simplex.Messaging.Transport.HTTP2.File
-import Simplex.Messaging.Util (bshow, whenM)
+import Simplex.Messaging.Util (bshow, liftEitherWith, liftError', tshow, whenM)
+import Simplex.Messaging.Version (compatibleVersion, pattern Compatible)
 import UnliftIO
 import UnliftIO.Directory
 
@@ -63,7 +70,8 @@ data XFTPClient = XFTPClient
 
 data XFTPClientConfig = XFTPClientConfig
   { xftpNetworkConfig :: NetworkConfig,
-    uploadTimeoutPerMb :: Int64
+    uploadTimeoutPerMb :: Int64,
+    serverVRange :: VersionRangeXFTP
   }
 
 data XFTPChunkBody = XFTPChunkBody
@@ -85,12 +93,13 @@ defaultXFTPClientConfig :: XFTPClientConfig
 defaultXFTPClientConfig =
   XFTPClientConfig
     { xftpNetworkConfig = defaultNetworkConfig,
-      uploadTimeoutPerMb = 10000000 -- 10 seconds
+      uploadTimeoutPerMb = 10000000, -- 10 seconds
+      serverVRange = supportedFileServerVRange
     }
 
-getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
-getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {xftpNetworkConfig} disconnected = runExceptT $ do
-  let tcConfig = transportClientConfig xftpNetworkConfig
+getXFTPClient :: TVar ChaChaDRG -> TransportSession FileResponse -> XFTPClientConfig -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
+getXFTPClient g transportSession@(_, srv, _) config@XFTPClientConfig {xftpNetworkConfig, serverVRange} disconnected = runExceptT $ do
+  let tcConfig = (transportClientConfig xftpNetworkConfig) {alpn = Just ["xftp/1"]}
       http2Config = xftpHTTP2Config tcConfig config
       username = proxyUsername transportSession
       ProtocolServer _ host port keyHash = srv
@@ -98,13 +107,50 @@ getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {xftpNetworkC
   clientVar <- newTVarIO Nothing
   let usePort = if null port then "443" else port
       clientDisconnected = readTVarIO clientVar >>= mapM_ disconnected
-  http2Client <- withExceptT xftpClientError . ExceptT $ getVerifiedHTTP2Client (Just username) useHost usePort (Just keyHash) Nothing http2Config clientDisconnected
-  let HTTP2Client {sessionId} = http2Client
-      thParams = THandleParams {sessionId, blockSize = xftpBlockSize, thVersion = currentXFTPVersion, thAuth = Nothing, implySessId = False, batch = True}
-      c = XFTPClient {http2Client, thParams, transportSession, config}
+  http2Client <- liftError' xftpClientError $ getVerifiedHTTP2Client (Just username) useHost usePort (Just keyHash) Nothing http2Config clientDisconnected
+  let HTTP2Client {sessionId, sessionALPN} = http2Client
+      thParams0 = THandleParams {sessionId, blockSize = xftpBlockSize, thVersion = VersionXFTP 1, thAuth = Nothing, implySessId = False, batch = True}
+  logDebug $ "Client negotiated handshake protocol: " <> tshow sessionALPN
+  thParams <- case sessionALPN of
+    Just "xftp/1" -> xftpClientHandshakeV1 g serverVRange keyHash http2Client thParams0
+    Nothing -> pure thParams0
+    _ -> throwError $ PCETransportError (TEHandshake VERSION)
+  let c = XFTPClient {http2Client, thParams, transportSession, config}
   atomically $ writeTVar clientVar $ Just c
   pure c
 
+xftpClientHandshakeV1 :: TVar ChaChaDRG -> VersionRangeXFTP -> C.KeyHash -> HTTP2Client -> THandleParamsXFTP -> ExceptT XFTPClientError IO THandleParamsXFTP
+xftpClientHandshakeV1 g serverVRange keyHash@(C.KeyHash kh) c@HTTP2Client {sessionId, serverKey} thParams0 = do
+  shs <- getServerHandshake
+  (v, sk) <- processServerHandshake shs
+  (k, pk) <- atomically $ C.generateKeyPair g
+  sendClientHandshake XFTPClientHandshake {xftpVersion = v, keyHash, authPubKey = k}
+  pure thParams0 {thAuth = Just THandleAuth {peerPubKey = sk, privKey = pk}, thVersion = v}
+  where
+    getServerHandshake = do
+      let helloReq = H.requestNoBody "POST" "/" []
+      HTTP2Response {respBody = HTTP2Body {bodyHead = shsBody}} <-
+        liftError' (const $ PCEResponseError HANDSHAKE) $ sendRequestDirect c helloReq Nothing
+      liftHS . smpDecode =<< liftHS (C.unPad shsBody)
+    processServerHandshake XFTPServerHandshake {xftpVersionRange, sessionId = serverSessId, authPubKey = serverAuth} = do
+      unless (sessionId == serverSessId) $ throwError $ PCEResponseError SESSION
+      case xftpVersionRange `compatibleVersion` serverVRange of
+        Nothing -> throwError $ PCEResponseError HANDSHAKE
+        Just (Compatible v) ->
+          fmap (v,) . liftHS $ do
+            let (X.CertificateChain cert, exact) = serverAuth
+            case cert of
+              [_leaf, ca] | XV.Fingerprint kh == XV.getFingerprint ca X.HashSHA256 -> pure ()
+              _ -> throwError "bad certificate"
+            pubKey <- C.verifyX509 serverKey exact
+            C.x509ToPublic (pubKey, []) >>= C.pubKey
+    sendClientHandshake chs = do
+      chs' <- liftHS $ C.pad (smpEncode chs) xftpBlockSize
+      let chsReq = H.requestBuilder "POST" "/" [] $ byteString chs'
+      HTTP2Response {respBody = HTTP2Body {bodyHead}} <- liftError' (const $ PCEResponseError HANDSHAKE) $ sendRequestDirect c chsReq Nothing
+      unless (B.null bodyHead) $ throwError $ PCEResponseError HANDSHAKE
+    liftHS = liftEitherWith (const $ PCEResponseError HANDSHAKE)
+
 closeXFTPClient :: XFTPClient -> IO ()
 closeXFTPClient XFTPClient {http2Client} = closeHTTP2Client http2Client
 
@@ -198,8 +244,8 @@ downloadXFTPChunk g c@XFTPClient {config} rpKey fId chunkSpec@XFTPRcvChunkSpec {
         let t = chunkTimeout config chunkSize
         ExceptT (sequence <$> (t `timeout` download cbState)) >>= maybe (throwError PCEResponseTimeout) pure
         where
-          download cbState = runExceptT $
-            withExceptT PCEResponseError $
+          download cbState =
+            runExceptT . withExceptT PCEResponseError $
               receiveEncFile chunkPart cbState chunkSpec `catchError` \e ->
                 whenM (doesFileExist filePath) (removeFile filePath) >> throwError e
       _ -> throwError $ PCEResponseError NO_FILE

src/Simplex/FileTransfer/Client/Agent.hs

@@ -11,6 +11,7 @@ import Control.Logger.Simple (logInfo)
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Trans (lift)
+import Crypto.Random (ChaChaDRG)
 import Data.Bifunctor (first)
 import qualified Data.ByteString.Char8 as B
 import Data.Text (Text)
@@ -23,7 +24,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtocolServer (..), XFTPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (catchAll_)
+import Simplex.Messaging.Util (catchAll_, atomically')
 import UnliftIO
 
 type XFTPClientVar = TMVar (Either XFTPClientAgentError XFTPClient)
@@ -60,15 +61,15 @@ newXFTPAgent config = do
 
 type ME a = ExceptT XFTPClientAgentError IO a
 
-getXFTPServerClient :: XFTPClientAgent -> XFTPServer -> ME XFTPClient
-getXFTPServerClient XFTPClientAgent {xftpClients, config} srv = do
-  atomically getClientVar >>= either newXFTPClient waitForXFTPClient
+getXFTPServerClient :: TVar ChaChaDRG -> XFTPClientAgent -> XFTPServer -> ME XFTPClient
+getXFTPServerClient g XFTPClientAgent {xftpClients, config} srv = do
+  atomically' getClientVar >>= either newXFTPClient waitForXFTPClient
   where
     connectClient :: ME XFTPClient
     connectClient =
       ExceptT $
         first (XFTPClientAgentError srv)
-          <$> getXFTPClient (1, srv, Nothing) (xftpConfig config) clientDisconnected
+          <$> getXFTPClient g (1, srv, Nothing) (xftpConfig config) clientDisconnected
 
     clientDisconnected :: XFTPClient -> IO ()
     clientDisconnected _ = do
@@ -87,7 +88,7 @@ getXFTPServerClient XFTPClientAgent {xftpClients, config} srv = do
     waitForXFTPClient :: XFTPClientVar -> ME XFTPClient
     waitForXFTPClient clientVar = do
       let XFTPClientConfig {xftpNetworkConfig = NetworkConfig {tcpConnectTimeout}} = xftpConfig config
-      client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar clientVar)
+      client_ <- liftIO $ tcpConnectTimeout `timeout` atomically' (readTMVar clientVar)
       liftEither $ case client_ of
         Just (Right c) -> Right c
         Just (Left e) -> Left e
@@ -101,12 +102,12 @@ getXFTPServerClient XFTPClientAgent {xftpClients, config} srv = do
           tryError connectClient >>= \r -> case r of
             Right client -> do
               logInfo $ "connected to " <> showServer srv
-              atomically $ putTMVar clientVar r
+              atomically' $ putTMVar clientVar r
               pure client
             Left e@(XFTPClientAgentError _ e') -> do
               if temporaryClientError e'
                 then retryAction
-                else atomically $ do
+                else atomically' $ do
                   putTMVar clientVar r
                   TM.delete srv xftpClients
               throwError e
@@ -124,6 +125,6 @@ closeXFTPServerClient XFTPClientAgent {xftpClients, config} srv =
   where
     closeClient cVar = do
       let NetworkConfig {tcpConnectTimeout} = xftpNetworkConfig $ xftpConfig config
-      tcpConnectTimeout `timeout` atomically (readTMVar cVar) >>= \case
+      tcpConnectTimeout `timeout` atomically' (readTMVar cVar) >>= \case
         Just (Right client) -> closeXFTPClient client `catchAll_` pure ()
         _ -> pure ()

src/Simplex/FileTransfer/Client/Main.hs

@@ -333,9 +333,9 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
           rKeys <- atomically $ L.fromList <$> replicateM numRecipients (C.generateAuthKeyPair C.SEd25519 g)
           digest <- liftIO $ getChunkDigest chunkSpec
           let ch = FileInfo {sndKey, size = fromIntegral chunkSize, digest}
-          c <- withRetry retryCount $ getXFTPServerClient a xftpServer
+          c <- withRetry retryCount $ getXFTPServerClient g a xftpServer
           (sndId, rIds) <- withRetry retryCount $ createXFTPChunk c spKey ch (L.map fst rKeys) auth
-          withReconnect a xftpServer retryCount $ \c' -> uploadXFTPChunk c' spKey sndId chunkSpec
+          withReconnect g a xftpServer retryCount $ \c' -> uploadXFTPChunk c' spKey sndId chunkSpec
           logInfo $ "uploaded chunk " <> tshow chunkNo
           uploaded <- atomically . stateTVar uploadedChunks $ \cs ->
             let cs' = fromIntegral chunkSize : cs in (sum cs', cs')
@@ -445,7 +445,7 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
       when (FileSize encSize /= size) $ throwError $ CLIError "File size mismatch"
       liftIO $ printNoNewLine "Decrypting file..."
       CryptoFile path _ <- withExceptT cliCryptoError $ decryptChunks encSize chunkPaths key nonce $ fmap CF.plain . getFilePath
-      forM_ chunks $ acknowledgeFileChunk a
+      forM_ chunks $ acknowledgeFileChunk g a
       whenM (doesPathExist encPath) $ removeDirectoryRecursive encPath
       liftIO $ do
         printNoNewLine $ "File downloaded: " <> path
@@ -456,7 +456,7 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
       logInfo $ "downloading chunk " <> tshow chunkNo <> " from " <> showServer server <> "..."
       chunkPath <- uniqueCombine encPath $ show chunkNo
       let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
-      withReconnect a server retryCount $ \c -> downloadXFTPChunk g c replicaKey (unChunkReplicaId replicaId) chunkSpec
+      withReconnect g a server retryCount $ \c -> downloadXFTPChunk g c replicaKey (unChunkReplicaId replicaId) chunkSpec
       logInfo $ "downloaded chunk " <> tshow chunkNo <> " to " <> T.pack chunkPath
       downloaded <- atomically . stateTVar downloadedChunks $ \cs ->
         let cs' = fromIntegral (unFileSize chunkSize) : cs in (sum cs', cs')
@@ -472,12 +472,12 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
           ifM (doesDirectoryExist path) (uniqueCombine path name) $
             ifM (doesFileExist path) (throwError "File already exists") (pure path)
         _ -> (`uniqueCombine` name) . (</> "Downloads") =<< getHomeDirectory
-    acknowledgeFileChunk :: XFTPClientAgent -> FileChunk -> ExceptT CLIError IO ()
-    acknowledgeFileChunk a FileChunk {replicas = replica : _} = do
+    acknowledgeFileChunk :: TVar ChaChaDRG -> XFTPClientAgent -> FileChunk -> ExceptT CLIError IO ()
+    acknowledgeFileChunk g a FileChunk {replicas = replica : _} = do
       let FileChunkReplica {server, replicaId, replicaKey} = replica
-      c <- withRetry retryCount $ getXFTPServerClient a server
+      c <- withRetry retryCount $ getXFTPServerClient g a server
       withRetry retryCount $ ackXFTPChunk c replicaKey (unChunkReplicaId replicaId)
-    acknowledgeFileChunk _ _ = throwError $ CLIError "chunk has no replicas"
+    acknowledgeFileChunk _ _ _ = throwError $ CLIError "chunk has no replicas"
 
 printProgress :: String -> Int64 -> Int64 -> IO ()
 printProgress s part total = printNoNewLine $ s <> " " <> show ((part * 100) `div` total) <> "%"
@@ -501,7 +501,8 @@ cliDeleteFile DeleteOptions {fileDescription, retryCount, yes} = do
     deleteFileChunk :: XFTPClientAgent -> FileChunk -> ExceptT CLIError IO ()
     deleteFileChunk a FileChunk {chunkNo, replicas = replica : _} = do
       let FileChunkReplica {server, replicaId, replicaKey} = replica
-      withReconnect a server retryCount $ \c -> deleteXFTPChunk c replicaKey (unChunkReplicaId replicaId)
+      g <- liftIO C.newRandom
+      withReconnect g a server retryCount $ \c -> deleteXFTPChunk c replicaKey (unChunkReplicaId replicaId)
       logInfo $ "deleted chunk " <> tshow chunkNo <> " from " <> showServer server
     deleteFileChunk _ _ = throwError $ CLIError "chunk has no replicas"
 
@@ -569,9 +570,9 @@ prepareChunkSpecs filePath chunkSizes = reverse . snd $ foldl' addSpec (0, []) c
 getEncPath :: MonadIO m => Maybe FilePath -> String -> m FilePath
 getEncPath path name = (`uniqueCombine` (name <> ".encrypted")) =<< maybe (liftIO getCanonicalTemporaryDirectory) pure path
 
-withReconnect :: Show e => XFTPClientAgent -> XFTPServer -> Int -> (XFTPClient -> ExceptT e IO a) -> ExceptT CLIError IO a
-withReconnect a srv n run = withRetry n $ do
-  c <- withRetry n $ getXFTPServerClient a srv
+withReconnect :: Show e => TVar ChaChaDRG -> XFTPClientAgent -> XFTPServer -> Int -> (XFTPClient -> ExceptT e IO a) -> ExceptT CLIError IO a
+withReconnect g a srv n run = withRetry n $ do
+  c <- withRetry n $ getXFTPServerClient g a srv
   withExceptT (CLIError . show) (run c) `catchError` \e -> do
     liftIO $ closeXFTPServerClient a srv
     throwError e

src/Simplex/FileTransfer/Protocol.hs

@@ -25,7 +25,7 @@ import Data.List.NonEmpty (NonEmpty (..))
 import Data.Maybe (isNothing)
 import Data.Type.Equality
 import Data.Word (Word32)
-import Simplex.FileTransfer.Transport (VersionXFTP, XFTPErrorType (..), XFTPVersion, pattern VersionXFTP, xftpClientHandshake)
+import Simplex.FileTransfer.Transport (VersionXFTP, XFTPErrorType (..), XFTPVersion, xftpClientHandshakeStub, pattern VersionXFTP)
 import Simplex.Messaging.Client (authTransmission)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
@@ -144,7 +144,7 @@ instance FilePartyI p => ProtocolMsgTag (FileCommandTag p) where
 instance Protocol XFTPVersion XFTPErrorType FileResponse where
   type ProtoCommand FileResponse = FileCmd
   type ProtoType FileResponse = 'PXFTP
-  protocolClientHandshake = xftpClientHandshake
+  protocolClientHandshake = xftpClientHandshakeStub
   protocolPing = FileCmd SFRecipient PING
   protocolError = \case
     FRErr e -> Just e
@@ -329,9 +329,9 @@ checkParty' c = case testEquality (sFileParty @p) (sFileParty @p') of
   _ -> Nothing
 
 xftpEncodeAuthTransmission :: ProtocolEncoding XFTPVersion e c => THandleParams XFTPVersion -> C.APrivateAuthKey -> Transmission c -> Either TransportError ByteString
-xftpEncodeAuthTransmission thParams pKey (corrId, fId, msg) = do
+xftpEncodeAuthTransmission thParams@THandleParams {thAuth} pKey (corrId, fId, msg) = do
   let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, fId, msg)
-  xftpEncodeBatch1 . (,tToSend) =<< authTransmission Nothing (Just pKey) corrId tForAuth
+  xftpEncodeBatch1 . (,tToSend) =<< authTransmission thAuth (Just pKey) corrId tForAuth
 
 xftpEncodeTransmission :: ProtocolEncoding XFTPVersion e c => THandleParams XFTPVersion -> Transmission c -> Either TransportError ByteString
 xftpEncodeTransmission thParams (corrId, fId, msg) = do

src/Simplex/FileTransfer/Server.hs

@@ -9,6 +9,7 @@
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 
 module Simplex.FileTransfer.Server where
 
@@ -17,7 +18,8 @@ import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
 import Data.Bifunctor (first)
-import Data.ByteString.Builder (byteString)
+import qualified Data.ByteString.Base64.URL as B64
+import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
@@ -31,6 +33,7 @@ import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Data.Word (Word32)
+import qualified Data.X509 as X
 import GHC.IO.Handle (hSetNewlineMode)
 import GHC.Stats (getRTSStats)
 import qualified Network.HTTP.Types as N
@@ -45,19 +48,22 @@ import Simplex.FileTransfer.Server.StoreLog
 import Simplex.FileTransfer.Transport
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
+import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (CorrId, RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TransmissionAuth)
+import Simplex.Messaging.Protocol (CorrId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TransmissionAuth)
 import Simplex.Messaging.Server (dummyVerifyCmd, verifyCmdAuthorization)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Stats
-import Simplex.Messaging.Transport (THandleParams (..))
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Transport (SessionId, THandleAuth (..), THandleParams (..))
 import Simplex.Messaging.Transport.Buffer (trimCR)
 import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Transport.HTTP2.File (fileBlockSize)
 import Simplex.Messaging.Transport.HTTP2.Server
-import Simplex.Messaging.Transport.Server (runTCPServer)
+import Simplex.Messaging.Transport.Server (runTCPServer, tlsServerCredentials)
 import Simplex.Messaging.Util
+import Simplex.Messaging.Version (isCompatible)
 import System.Exit (exitFailure)
 import System.FilePath ((</>))
 import System.IO (hPrint, hPutStrLn, universalNewlineMode)
@@ -69,7 +75,7 @@ import qualified UnliftIO.Exception as E
 type M a = ReaderT XFTPEnv IO a
 
 data XFTPTransportRequest = XFTPTransportRequest
-  { thParams :: THandleParams XFTPVersion,
+  { thParams :: THandleParamsXFTP,
     reqBody :: HTTP2Body,
     request :: H.Request,
     sendResponse :: H.Response -> IO ()
@@ -83,6 +89,10 @@ runXFTPServer cfg = do
 runXFTPServerBlocking :: TMVar Bool -> XFTPServerConfig -> IO ()
 runXFTPServerBlocking started cfg = newXFTPServerEnv cfg >>= runReaderT (xftpServer cfg started)
 
+data Handshake
+  = HandshakeSent C.PrivateKeyX25519
+  | HandshakeAccepted THandleAuth VersionXFTP
+
 xftpServer :: XFTPServerConfig -> TMVar Bool -> M ()
 xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpiration, fileExpiration} started = do
   mapM_ (expireServerFiles Nothing) fileExpiration
@@ -92,12 +102,62 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
     runServer :: M ()
     runServer = do
       serverParams <- asks tlsServerParams
+      let (chain, pk) = tlsServerCredentials serverParams
+      signKey <- liftIO $ case C.x509ToPrivate (pk, []) >>= C.privKey of
+        Right pk' -> pure pk'
+        Left e -> putStrLn ("servers has no valid key: " <> show e) >> exitFailure
       env <- ask
-      liftIO $
-        runHTTP2Server started xftpPort defaultHTTP2BufferSize serverParams transportConfig inactiveClientExpiration $ \sessionId r sendResponse -> do
-          reqBody <- getHTTP2Body r xftpBlockSize
-          let thParams = THandleParams {sessionId, blockSize = xftpBlockSize, thVersion = currentXFTPVersion, thAuth = Nothing, implySessId = False, batch = True}
-          processRequest XFTPTransportRequest {thParams, request = r, reqBody, sendResponse} `runReaderT` env
+      sessions <- atomically' TM.empty
+      let cleanup sessionId = atomically $ TM.delete sessionId sessions
+      liftIO . runHTTP2Server started xftpPort defaultHTTP2BufferSize serverParams transportConfig inactiveClientExpiration cleanup $ \sessionId sessionALPN r sendResponse -> do
+        reqBody <- getHTTP2Body r xftpBlockSize
+        let thParams0 = THandleParams {sessionId, blockSize = xftpBlockSize, thVersion = VersionXFTP 1, thAuth = Nothing, implySessId = False, batch = True}
+            req0 = XFTPTransportRequest {thParams = thParams0, request = r, reqBody, sendResponse}
+        flip runReaderT env $ case sessionALPN of
+          Nothing -> processRequest req0
+          Just "xftp/1" ->
+            xftpServerHandshakeV1 chain signKey sessions req0 >>= \case
+              Nothing -> pure () -- handshake response sent
+              Just thParams -> processRequest req0 {thParams} -- proceed with new version (XXX: may as well switch the request handler here)
+          _ -> liftIO . sendResponse $ H.responseNoBody N.ok200 [] -- shouldn't happen: means server picked handshake protocol it doesn't know about
+    xftpServerHandshakeV1 :: X.CertificateChain -> C.APrivateSignKey -> TMap SessionId Handshake -> XFTPTransportRequest -> M (Maybe (THandleParams XFTPVersion))
+    xftpServerHandshakeV1 chain serverSignKey sessions XFTPTransportRequest {thParams = thParams@THandleParams {sessionId}, reqBody = HTTP2Body {bodyHead}, sendResponse} = do
+      s <- atomically $ TM.lookup sessionId sessions
+      r <- runExceptT $ case s of
+        Nothing -> processHello
+        Just (HandshakeSent pk) -> processClientHandshake pk
+        Just (HandshakeAccepted auth v) -> pure $ Just thParams {thAuth = Just auth, thVersion = v}
+      either sendError pure r
+      where
+        processHello = do
+          unless (B.null bodyHead) $ throwError HANDSHAKE
+          (k, pk) <- atomically . C.generateKeyPair =<< asks random
+          atomically $ TM.insert sessionId (HandshakeSent pk) sessions
+          let authPubKey = (chain, C.signX509 serverSignKey $ C.publicToX509 k)
+          let hs = XFTPServerHandshake {xftpVersionRange = supportedFileServerVRange, sessionId, authPubKey}
+          shs <- encodeXftp hs
+          liftIO . sendResponse $ H.responseBuilder N.ok200 [] shs
+          pure Nothing
+        processClientHandshake privKey = do
+          unless (B.length bodyHead == xftpBlockSize) $ throwError HANDSHAKE
+          body <- liftHS $ C.unPad bodyHead
+          XFTPClientHandshake {xftpVersion, keyHash, authPubKey} <- liftHS $ smpDecode body
+          kh <- asks serverIdentity
+          unless (keyHash == kh) $ throwError HANDSHAKE
+          unless (xftpVersion `isCompatible` supportedFileServerVRange) $ throwError HANDSHAKE
+          let auth = THandleAuth {peerPubKey = authPubKey, privKey}
+          atomically $ TM.insert sessionId (HandshakeAccepted auth xftpVersion) sessions
+          liftIO . sendResponse $ H.responseNoBody N.ok200 []
+          pure Nothing
+        sendError :: XFTPErrorType -> M (Maybe (THandleParams XFTPVersion))
+        sendError err = do
+          runExceptT (encodeXftp err) >>= \case
+            Right bs -> liftIO . sendResponse $ H.responseBuilder N.ok200 [] bs
+            Left _ -> logError $ "Error encoding handshake error: " <> tshow err
+          pure Nothing
+        encodeXftp :: Encoding a => a -> ExceptT XFTPErrorType (ReaderT XFTPEnv IO) Builder
+        encodeXftp a = byteString <$> liftHS (C.pad (smpEncode a) xftpBlockSize)
+        liftHS = liftEitherWith (const HANDSHAKE)
 
     stopServer :: M ()
     stopServer = do
@@ -131,15 +191,15 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
         withFile statsFilePath AppendMode $ \h -> liftIO $ do
           hSetBuffering h LineBuffering
           ts <- getCurrentTime
-          fromTime' <- atomically $ swapTVar fromTime ts
-          filesCreated' <- atomically $ swapTVar filesCreated 0
-          fileRecipients' <- atomically $ swapTVar fileRecipients 0
-          filesUploaded' <- atomically $ swapTVar filesUploaded 0
-          filesExpired' <- atomically $ swapTVar filesExpired 0
-          filesDeleted' <- atomically $ swapTVar filesDeleted 0
-          files <- atomically $ periodStatCounts filesDownloaded ts
-          fileDownloads' <- atomically $ swapTVar fileDownloads 0
-          fileDownloadAcks' <- atomically $ swapTVar fileDownloadAcks 0
+          fromTime' <- atomically' $ swapTVar fromTime ts
+          filesCreated' <- atomically' $ swapTVar filesCreated 0
+          fileRecipients' <- atomically' $ swapTVar fileRecipients 0
+          filesUploaded' <- atomically' $ swapTVar filesUploaded 0
+          filesExpired' <- atomically' $ swapTVar filesExpired 0
+          filesDeleted' <- atomically' $ swapTVar filesDeleted 0
+          files <- atomically' $ periodStatCounts filesDownloaded ts
+          fileDownloads' <- atomically' $ swapTVar fileDownloads 0
+          fileDownloadAcks' <- atomically' $ swapTVar fileDownloadAcks 0
           filesCount' <- readTVarIO filesCount
           filesSize' <- readTVarIO filesSize
           hPutStrLn h $
@@ -183,7 +243,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
           role <- newTVarIO CPRNone
           cpLoop h role
           where
-            cpLoop h role  = do
+            cpLoop h role = do
               s <- trimCR <$> B.hGetLine h
               case strDecode s of
                 Right CPQuit -> hClose h
@@ -208,8 +268,8 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
               CPDelete fileId -> withUserRole $ unliftIO u $ do
                 fs <- asks store
                 r <- runExceptT $ do
-                  let asSender = ExceptT . atomically $ getFile fs SFSender fileId
-                  let asRecipient = ExceptT . atomically $ getFile fs SFRecipient fileId
+                  let asSender = ExceptT . atomically' $ getFile fs SFSender fileId
+                  let asRecipient = ExceptT . atomically' $ getFile fs SFRecipient fileId
                   (fr, _) <- asSender `catchError` const asRecipient
                   ExceptT $ deleteServerFile_ fr
                 liftIO . hPutStrLn h $ either (\e -> "error: " <> show e) (\() -> "ok") r
@@ -217,12 +277,13 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
               CPQuit -> pure ()
               CPSkip -> pure ()
               where
-                withUserRole action = readTVarIO role >>= \case
-                  CPRAdmin -> action
-                  CPRUser -> action
-                  _ -> do
-                    logError "Unauthorized control port command"
-                    hPutStrLn h "AUTH"
+                withUserRole action =
+                  readTVarIO role >>= \case
+                    CPRAdmin -> action
+                    CPRUser -> action
+                    _ -> do
+                      logError "Unauthorized control port command"
+                      hPutStrLn h "AUTH"
 
 data ServerFile = ServerFile
   { filePath :: FilePath,
@@ -235,10 +296,11 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea
   | B.length bodyHead /= xftpBlockSize = sendXFTPResponse ("", "", FRErr BLOCK) Nothing
   | otherwise = do
       case xftpDecodeTransmission thParams bodyHead of
-        Right (sig_, signed, (corrId, fId, cmdOrErr)) -> do
+        Right (sig_, signed, (corrId, fId, cmdOrErr)) ->
           case cmdOrErr of
             Right cmd -> do
-              verifyXFTPTransmission sig_ signed fId cmd >>= \case
+              let THandleParams {thAuth} = thParams
+              verifyXFTPTransmission ((,C.cbNonce (bs corrId)) <$> thAuth) sig_ signed fId cmd >>= \case
                 VRVerified req -> uncurry send =<< processXFTPRequest body req
                 VRFailed -> send (FRErr AUTH) Nothing
             Left e -> send (FRErr e) Nothing
@@ -246,7 +308,6 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea
             send resp = sendXFTPResponse (corrId, fId, resp)
         Left e -> sendXFTPResponse ("", "", FRErr e) Nothing
   where
-    sendXFTPResponse :: (CorrId, XFTPFileId, FileResponse) -> Maybe ServerFile -> M ()
     sendXFTPResponse (corrId, fId, resp) serverFile_ = do
       let t_ = xftpEncodeTransmission thParams (corrId, fId, resp)
       liftIO $ sendResponse $ H.responseStreaming N.ok200 [] $ streamBody t_
@@ -265,8 +326,8 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea
 
 data VerificationResult = VRVerified XFTPRequest | VRFailed
 
-verifyXFTPTransmission :: Maybe TransmissionAuth -> ByteString -> XFTPFileId -> FileCmd -> M VerificationResult
-verifyXFTPTransmission tAuth authorized fId cmd =
+verifyXFTPTransmission :: Maybe (THandleAuth, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> XFTPFileId -> FileCmd -> M VerificationResult
+verifyXFTPTransmission auth_ tAuth authorized fId cmd =
   case cmd of
     FileCmd SFSender (FNEW file rcps auth') -> pure $ XFTPReqNew file rcps auth' `verifyWith` sndKey file
     FileCmd SFRecipient PING -> pure $ VRVerified XFTPReqPing
@@ -275,13 +336,13 @@ verifyXFTPTransmission tAuth authorized fId cmd =
     verifyCmd :: SFileParty p -> M VerificationResult
     verifyCmd party = do
       st <- asks store
-      atomically $ verify <$> getFile st party fId
+      atomically' $ verify <$> getFile st party fId
       where
         verify = \case
           Right (fr, k) -> XFTPReqCmd fId fr cmd `verifyWith` k
           _ -> maybe False (dummyVerifyCmd Nothing authorized) tAuth `seq` VRFailed
     -- TODO verify with DH authorization
-    req `verifyWith` k = if verifyCmdAuthorization Nothing tAuth authorized k then VRVerified req else VRFailed
+    req `verifyWith` k = if verifyCmdAuthorization auth_ tAuth authorized k then VRVerified req else VRFailed
 
 processXFTPRequest :: HTTP2Body -> XFTPRequest -> M (FileResponse, Maybe ServerFile)
 processXFTPRequest HTTP2Body {bodyPart} = \case
@@ -336,7 +397,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
     retryAdd 0 _ = pure $ Left INTERNAL
     retryAdd n add = do
       fId <- getFileId
-      atomically (add fId) >>= \case
+      atomically' (add fId) >>= \case
         Left DUPLICATE_ -> retryAdd (n - 1) add
         r -> pure r
     addRecipients :: XFTPFileId -> NonEmpty RcvPublicAuthKey -> M FileResponse
@@ -374,7 +435,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 (U.encode senderId)
+            let fPath = path </> B.unpack (B64.encode senderId)
             receiveChunk (XFTPRcvChunkSpec fPath size digest) >>= \case
               Right () -> do
                 stats <- asks serverStats
@@ -386,7 +447,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
                 pure FROk
               Left e -> do
                 us <- asks $ usedStorage . store
-                atomically . modifyTVar' us $ subtract (fromIntegral size)
+                atomically' . modifyTVar' us $ subtract (fromIntegral size)
                 liftIO $ whenM (doesFileExist fPath) (removeFile fPath) `catch` logFileError
                 pure $ FRErr e
           receiveChunk spec = do
@@ -404,7 +465,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
             Right sbState -> do
               stats <- asks serverStats
               atomically $ modifyTVar' (fileDownloads stats) (+ 1)
-              atomically $ updatePeriodStats (filesDownloaded stats) senderId
+              atomically' $ updatePeriodStats (filesDownloaded stats) senderId
               pure (FRFile sDhKey cbNonce, Just ServerFile {filePath = path, fileSize = size, sbState})
             _ -> pure (FRErr INTERNAL, Nothing)
         _ -> pure (FRErr NO_FILE, Nothing)
@@ -419,7 +480,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
     ackFileReception rId fr = do
       withFileLog (`logAckFile` rId)
       st <- asks store
-      atomically $ deleteRecipient st rId fr
+      atomically' $ deleteRecipient st rId fr
       stats <- asks serverStats
       atomically $ modifyTVar' (fileDownloadAcks stats) (+ 1)
       pure FROk
@@ -432,7 +493,7 @@ deleteServerFile_ FileRec {senderId, fileInfo, filePath} = do
     stats <- asks serverStats
     ExceptT $ first (\(_ :: SomeException) -> FILE_IO) <$> try (forM_ path $ \p -> whenM (doesFileExist p) (removeFile p >> deletedStats stats))
     st <- asks store
-    void $ atomically $ deleteFile st senderId
+    void $ atomically' $ deleteFile st senderId
     atomically $ modifyTVar' (filesDeleted stats) (+ 1)
   where
     deletedStats stats = do
@@ -448,7 +509,7 @@ expireServerFiles itemDelay expCfg = do
   logInfo $ "Expiration check: " <> tshow (M.size files') <> " files"
   forM_ (M.keys files') $ \sId -> do
     mapM_ threadDelay itemDelay
-    atomically (expiredFilePath st sId old)
+    atomically' (expiredFilePath st sId old)
       >>= mapM_ (maybeRemove $ delete st sId)
   usedEnd <- readTVarIO $ usedStorage st
   logInfo $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed."
@@ -462,7 +523,7 @@ expireServerFiles itemDelay expCfg = do
         del
     delete st sId = do
       withFileLog (`logDeleteFile` sId)
-      void . atomically $ deleteFile st sId -- will not update usedStorage if sId isn't in store
+      void . atomically' $ deleteFile st sId -- will not update usedStorage if sId isn't in store
       FileServerStats {filesExpired} <- asks serverStats
       atomically $ modifyTVar' filesExpired (+ 1)
 
@@ -485,7 +546,7 @@ incFileStat statSel = do
 saveServerStats :: M ()
 saveServerStats =
   asks (serverStatsBackupFile . config)
-    >>= mapM_ (\f -> asks serverStats >>= atomically . getFileServerStatsData >>= liftIO . saveStats f)
+    >>= mapM_ (\f -> asks serverStats >>= atomically' . getFileServerStatsData >>= liftIO . saveStats f)
   where
     saveStats f stats = do
       logInfo $ "saving server stats to file " <> T.pack f
@@ -503,7 +564,7 @@ restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStat
           FileStore {files, usedStorage} <- asks store
           _filesCount <- M.size <$> readTVarIO files
           _filesSize <- readTVarIO usedStorage
-          atomically $ setFileServerStats s d {_filesCount, _filesSize}
+          atomically' $ setFileServerStats s d {_filesCount, _filesSize}
           renameFile f $ f <> ".bak"
           logInfo "server stats restored"
           when (statsFilesCount /= _filesCount) $ logWarn $ "Files count differs: stats: " <> tshow statsFilesCount <> ", store: " <> tshow _filesCount

src/Simplex/FileTransfer/Server/Env.hs

@@ -2,19 +2,23 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE StrictData #-}
 
 module Simplex.FileTransfer.Server.Env where
 
-import Control.Logger.Simple (logInfo)
+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.IO.Unlift
 import Crypto.Random
+import Data.Default (def)
 import Data.Int (Int64)
+import Data.List (find)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map.Strict as M
+import Data.Maybe (fromMaybe)
 import Data.Time.Clock (getCurrentTime)
 import Data.Word (Word32)
 import Data.X509.Validation (Fingerprint (..))
@@ -27,6 +31,7 @@ import Simplex.FileTransfer.Server.StoreLog
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (BasicAuth, RcvPublicAuthKey)
 import Simplex.Messaging.Server.Expiration
+import Simplex.Messaging.Transport (ALPN)
 import Simplex.Messaging.Transport.Server (TransportServerConfig, loadFingerprint, loadTLSServerParams)
 import Simplex.Messaging.Util (tshow)
 import System.IO (IOMode (..))
@@ -94,6 +99,9 @@ defaultFileExpiration =
       checkInterval = 2 * 3600 -- seconds, 2 hours
     }
 
+supportedXFTPhandshakes :: [ALPN]
+supportedXFTPhandshakes = ["xftp/1"]
+
 newXFTPServerEnv :: XFTPServerConfig -> IO XFTPEnv
 newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertificateFile, certificateFile, privateKeyFile} = do
   random <- liftIO C.newRandom
@@ -104,7 +112,14 @@ newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertifi
   forM_ fileSizeQuota $ \quota -> do
     logInfo $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used)
     when (quota < used) $ logInfo "WARNING: storage quota is less than used storage, no files can be uploaded!"
-  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+  tlsServerParams' <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+  let tlsServerParams =
+        tlsServerParams'
+          { T.serverHooks =
+              def
+                { T.onALPNClientSuggest = Just $ pure . fromMaybe "" . find (`elem` supportedXFTPhandshakes)
+                }
+          }
   Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
   serverStats <- atomically . newFileServerStats =<< liftIO getCurrentTime
   pure XFTPEnv {config, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats}

src/Simplex/FileTransfer/Server/StoreLog.hs

@@ -34,7 +34,7 @@ import Simplex.FileTransfer.Server.Store
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (RcvPublicAuthKey, RecipientId, SenderId)
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (bshow, whenM)
+import Simplex.Messaging.Util (bshow, whenM, atomically')
 import System.Directory (doesFileExist, renameFile)
 import System.IO
 
@@ -94,7 +94,7 @@ readFileStore f st = mapM_ (addFileLogRecord . LB.toStrict) . LB.lines =<< LB.re
     addFileLogRecord s = case strDecode s of
       Left e -> B.putStrLn $ "Log parsing error (" <> B.pack e <> "): " <> B.take 100 s
       Right lr ->
-        atomically (addToStore lr) >>= \case
+        atomically' (addToStore lr) >>= \case
           Left e -> B.putStrLn $ "Log processing error (" <> bshow e <> "): " <> B.take 100 s
           _ -> pure ()
     addToStore = \case

src/Simplex/FileTransfer/Transport.hs

@@ -9,9 +9,16 @@
 
 module Simplex.FileTransfer.Transport
   ( supportedFileServerVRange,
-    xftpClientHandshake, -- stub
-    XFTPVersion,
+    xftpClientHandshakeStub,
+    XFTPClientHandshake (..),
+    -- xftpClientHandshake,
+    XFTPServerHandshake (..),
+    -- xftpServerHandshake,
+    THandleXFTP,
+    THandleParamsXFTP,
     VersionXFTP,
+    VersionRangeXFTP,
+    XFTPVersion,
     pattern VersionXFTP,
     XFTPErrorType (..),
     XFTPRcvChunkSpec (..),
@@ -30,20 +37,21 @@ import Control.Monad.Except
 import Control.Monad.IO.Class
 import qualified Data.Aeson.TH as J
 import qualified Data.Attoparsec.ByteString.Char8 as A
-import Data.Bifunctor (first)
+import Data.Bifunctor (bimap, first)
 import qualified Data.ByteArray as BA
 import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Word (Word16, Word32)
+import qualified Data.X509 as X
 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.Parsers
 import Simplex.Messaging.Protocol (CommandError)
-import Simplex.Messaging.Transport (HandshakeError (..), THandle, TransportError (..))
+import Simplex.Messaging.Transport (HandshakeError (..), SessionId, THandle (..), THandleParams (..), TransportError (..))
 import Simplex.Messaging.Transport.HTTP2.File
 import Simplex.Messaging.Util (bshow)
 import Simplex.Messaging.Version
@@ -68,6 +76,9 @@ type VersionRangeXFTP = VersionRange XFTPVersion
 pattern VersionXFTP :: Word16 -> VersionXFTP
 pattern VersionXFTP v = Version v
 
+type THandleXFTP c = THandle XFTPVersion c
+type THandleParamsXFTP = THandleParams XFTPVersion
+
 initialXFTPVersion :: VersionXFTP
 initialXFTPVersion = VersionXFTP 1
 
@@ -75,8 +86,45 @@ supportedFileServerVRange :: VersionRangeXFTP
 supportedFileServerVRange = mkVersionRange initialXFTPVersion initialXFTPVersion
 
 -- XFTP protocol does not support handshake
-xftpClientHandshake :: c -> C.KeyPairX25519 -> C.KeyHash -> VersionRangeXFTP -> ExceptT TransportError IO (THandle XFTPVersion c)
-xftpClientHandshake _c _ks _keyHash _xftpVRange = throwError $ TEHandshake VERSION
+xftpClientHandshakeStub :: c -> C.KeyPairX25519 -> C.KeyHash -> VersionRangeXFTP -> ExceptT TransportError IO (THandle XFTPVersion c)
+xftpClientHandshakeStub _c _ks _keyHash _xftpVRange = throwError $ TEHandshake VERSION
+
+data XFTPServerHandshake = XFTPServerHandshake
+  { xftpVersionRange :: VersionRangeXFTP,
+    sessionId :: SessionId,
+    -- | pub key to agree shared secrets for command authorization and entity ID encryption.
+    authPubKey :: (X.CertificateChain, X.SignedExact X.PubKey)
+  }
+
+data XFTPClientHandshake = XFTPClientHandshake
+  { -- | agreed XFTP server protocol version
+    xftpVersion :: VersionXFTP,
+    -- | server identity - CA certificate fingerprint
+    keyHash :: C.KeyHash,
+    -- | pub key to agree shared secret for entity ID encryption, shared secret for command authorization is agreed using per-queue keys.
+    authPubKey :: C.PublicKeyX25519
+  }
+
+instance Encoding XFTPClientHandshake where
+  smpEncode XFTPClientHandshake {xftpVersion, keyHash, authPubKey} =
+    smpEncode (xftpVersion, keyHash, authPubKey)
+  smpP = do
+    (xftpVersion, keyHash) <- smpP
+    authPubKey <- smpP
+    Tail _compat <- smpP
+    pure XFTPClientHandshake {xftpVersion, keyHash, authPubKey}
+
+instance Encoding XFTPServerHandshake where
+  smpEncode XFTPServerHandshake {xftpVersionRange, sessionId, authPubKey} =
+    smpEncode (xftpVersionRange, sessionId, auth)
+    where
+      auth = bimap C.encodeCertChain C.SignedObject authPubKey
+  smpP = do
+    (xftpVersionRange, sessionId) <- smpP
+    cert <- C.certChainP
+    C.SignedObject key <- smpP
+    Tail _compat <- smpP
+    pure XFTPServerHandshake {xftpVersionRange, sessionId, authPubKey = (cert, key)}
 
 sendEncFile :: Handle -> (Builder -> IO ()) -> LC.SbState -> Word32 -> IO ()
 sendEncFile h send = go
@@ -139,6 +187,8 @@ data XFTPErrorType
     BLOCK
   | -- | incorrect SMP session ID (TLS Finished message / tls-unique binding RFC5929)
     SESSION
+  | -- | incorrect handshake command
+    HANDSHAKE
   | -- | SMP command is unknown or has invalid syntax
     CMD {cmdErr :: CommandError}
   | -- | command authorization error - bad signature or non-existing SMP queue
@@ -181,6 +231,7 @@ instance Encoding XFTPErrorType where
   smpEncode = \case
     BLOCK -> "BLOCK"
     SESSION -> "SESSION"
+    HANDSHAKE -> "HANDSHAKE"
     CMD err -> "CMD " <> smpEncode err
     AUTH -> "AUTH"
     SIZE -> "SIZE"
@@ -199,6 +250,7 @@ instance Encoding XFTPErrorType where
     A.takeTill (== ' ') >>= \case
       "BLOCK" -> pure BLOCK
       "SESSION" -> pure SESSION
+      "HANDSHAKE" -> pure HANDSHAKE
       "CMD" -> CMD <$> _smpP
       "AUTH" -> pure AUTH
       "SIZE" -> pure SIZE

src/Simplex/Messaging/Agent.hs

@@ -197,7 +197,7 @@ getSMPAgentClient_ clientId cfg initServers store backgroundMode =
   liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent
   where
     runAgent = do
-      c@AgentClient {acThread}  <- atomically . newAgentClient clientId initServers =<< ask
+      c@AgentClient {acThread}  <- atomically' . newAgentClient clientId initServers =<< ask
       t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c)
       atomically . writeTVar acThread . Just =<< mkWeakThreadId t
       pure c
@@ -224,7 +224,7 @@ disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAge
 -- only used in the tests
 disposeAgentClient :: AgentClient -> IO ()
 disposeAgentClient c@AgentClient {acThread, agentEnv = Env {store}} = do
-  t_ <- atomically (swapTVar acThread Nothing) $>>= (liftIO . deRefWeak)
+  t_ <- atomically' (swapTVar acThread Nothing) $>>= (liftIO . deRefWeak)
   disconnectAgentClient c
   mapM_ killThread t_
   liftIO $ closeSQLiteStore store
@@ -405,7 +405,7 @@ testProtocolServer c userId srv = withAgentEnv' c $ case protocolTypeI @p of
 -- | set SOCKS5 proxy on/off and optionally set TCP timeout
 setNetworkConfig :: AgentClient -> NetworkConfig -> IO ()
 setNetworkConfig c cfg' = do
-  cfg <- atomically $ do
+  cfg <- atomically' $ do
     swapTVar (useNetworkConfig c) cfg'
   when (cfg /= cfg') $ reconnectAllServers c
 
@@ -522,7 +522,7 @@ getAgentStats :: AgentClient -> IO [(AgentStatsKey, Int)]
 getAgentStats c = readTVarIO (agentStats c) >>= mapM (\(k, cnt) -> (k,) <$> readTVarIO cnt) . M.assocs
 
 resetAgentStats :: AgentClient -> IO ()
-resetAgentStats = atomically . TM.clear . agentStats
+resetAgentStats = atomically' . TM.clear . agentStats
 {-# INLINE resetAgentStats #-}
 
 withAgentEnv' :: AgentClient -> AM' a -> IO a
@@ -545,9 +545,9 @@ runAgentClient c = race_ (subscriber c) (client c)
 
 client :: AgentClient -> AM' ()
 client c@AgentClient {rcvQ, subQ} = forever $ do
-  (corrId, entId, cmd) <- atomically $ readTBQueue rcvQ
+  (corrId, entId, cmd) <- atomically' $ readTBQueue rcvQ
   runExceptT (processCommand c (entId, cmd))
-    >>= atomically . writeTBQueue subQ . \case
+    >>= atomically' . writeTBQueue subQ . \case
       Left e -> (corrId, entId, APC SAEConn $ ERR e)
       Right (entId', resp) -> (corrId, entId', resp)
 
@@ -587,7 +587,7 @@ deleteUser' c userId delSMPQueues = do
   atomically $ TM.delete userId $ smpServers c
   where
     delUser =
-      whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically $
+      whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically' $
         writeTBQueue (subQ c) ("", "", APC SAENone $ DEL_USER userId)
 
 newConnAsync :: ConnectionModeI c => AgentClient -> UserId -> ACorrId -> Bool -> SConnectionMode c -> CR.InitialKeys -> SubscriptionMode -> AM ConnId
@@ -709,7 +709,7 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srv
     SMSubscribe -> addSubscription c rq'
   when enableNtfs $ do
     ns <- asks ntfSupervisor
-    atomically $ sendNtfSubCommand ns (connId, NSCCreate)
+    atomically' $ sendNtfSubCommand ns (connId, NSCCreate)
   let pqEnc = CR.connPQEncryption pqInitKeys
       crData = ConnReqUriData SSSimplex (smpAgentVRange pqEnc) [qUri] clientData
       e2eVRange = e2eEncryptVRange pqEnc
@@ -769,7 +769,7 @@ compatibleContactUri (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues =
   AgentConfig {smpClientVRange, smpAgentVRange} <- asks config
   pure $
     (,)
-      <$> (qUri `compatibleVersion` smpClientVRange) 
+      <$> (qUri `compatibleVersion` smpClientVRange)
       <*> (crAgentVRange `compatibleVersion` smpAgentVRange pqSup)
 
 versionPQSupport_ :: VersionSMPA -> Maybe CR.VersionE2E -> PQSupport
@@ -820,7 +820,7 @@ createReplyQueue c ConnData {userId, connId, enableNtfs} SndQueue {smpClientVers
     SMSubscribe -> addSubscription c rq'
   when enableNtfs $ do
     ns <- asks ntfSupervisor
-    atomically $ sendNtfSubCommand ns (connId, NSCCreate)
+    atomically' $ sendNtfSubCommand ns (connId, NSCCreate)
   pure qInfo
 
 -- | Approve confirmation (LET command) in Reader monad
@@ -930,7 +930,7 @@ subscribeConnections' c connIds = do
     notifyResultError rs = do
       let actual = M.size rs
           expected = length connIds
-      when (actual /= expected) . atomically $
+      when (actual /= expected) . atomically' $
         writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)
 
 resubscribeConnection' :: AgentClient -> ConnId -> AM ()
@@ -941,13 +941,13 @@ resubscribeConnections' :: AgentClient -> [ConnId] -> AM (Map ConnId (Either Age
 resubscribeConnections' _ [] = pure M.empty
 resubscribeConnections' c connIds = do
   let r = M.fromList . zip connIds . repeat $ Right ()
-  connIds' <- filterM (fmap not . atomically . hasActiveSubscription c) connIds
+  connIds' <- filterM (fmap not . atomically' . hasActiveSubscription c) connIds
   -- union is left-biased, so results returned by subscribeConnections' take precedence
   (`M.union` r) <$> subscribeConnections' c connIds'
 
 getConnectionMessage' :: AgentClient -> ConnId -> AM (Maybe SMPMsgMeta)
 getConnectionMessage' c connId = do
-  whenM (atomically $ hasActiveSubscription c connId) . throwError $ CMD PROHIBITED
+  whenM (atomically' $ hasActiveSubscription c connId) . throwError $ CMD PROHIBITED
   SomeConn _ conn <- withStore c (`getConn` connId)
   case conn of
     DuplexConnection _ (rq :| _) _ -> getQueueMessage c rq
@@ -1033,7 +1033,7 @@ resumeConnCmds c connId =
     withStore' c (`getPendingCommandServers` connId)
       >>= mapM_ (lift . resumeSrvCmds c)
   where
-    connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)
+    connQueued = atomically' $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)
 
 getAsyncCmdWorker :: Bool -> AgentClient -> Maybe SMPServer -> AM' Worker
 getAsyncCmdWorker hasWork c server =
@@ -1043,10 +1043,10 @@ runCommandProcessing :: AgentClient -> Maybe SMPServer -> Worker -> AM ()
 runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
   ri <- asks $ messageRetryInterval . config -- different retry interval?
   forever $ do
-    atomically $ endAgentOperation c AOSndNetwork
+    atomically' $ endAgentOperation c AOSndNetwork
     lift $ waitForWork doWork
-    atomically $ throwWhenInactive c
-    atomically $ beginAgentOperation c AOSndNetwork
+    atomically' $ throwWhenInactive c
+    atomically' $ beginAgentOperation c AOSndNetwork
     withWork c doWork (`getPendingServerCommand` server_) $ processCmd (riFast ri)
   where
     processCmd :: RetryInterval -> PendingCommand -> AM ()
@@ -1126,7 +1126,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
                     withStore' c $ \db -> deleteConnRcvQueue db rq'
                     when (enableNtfs cData) $ do
                       ns <- asks ntfSupervisor
-                      atomically $ sendNtfSubCommand ns (connId, NSCCreate)
+                      atomically' $ sendNtfSubCommand ns (connId, NSCCreate)
                     let conn' = DuplexConnection cData (rq'' :| rqs') sqs
                     notify $ SWITCH QDRcv SPCompleted $ connectionStats conn'
               _ -> internalErr "ICQDelete: cannot delete the only queue in connection"
@@ -1256,14 +1256,14 @@ runSmpQueueMsgDelivery :: AgentClient -> ConnData -> SndQueue -> (Worker, TMVar
 runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork}, qLock) = do
   AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
   forever $ do
-    atomically $ endAgentOperation c AOSndNetwork
+    atomically' $ endAgentOperation c AOSndNetwork
     lift $ waitForWork doWork
-    atomically $ throwWhenInactive c
-    atomically $ throwWhenNoDelivery c sq
-    atomically $ beginAgentOperation c AOSndNetwork
+    atomically' $ throwWhenInactive c
+    atomically' $ throwWhenNoDelivery c sq
+    atomically' $ beginAgentOperation c AOSndNetwork
     withWork c doWork (\db -> getPendingQueueMsg db connId sq) $
       \(rq_, PendingMsgData {msgId, msgType, msgBody, pqEncryption, msgFlags, msgRetryState, internalTs}) -> do
-        atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg
+        atomically' $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg
         let mId = unId msgId
             ri' = maybe id updateRetryInterval2 msgRetryState ri
         withRetryLock2 ri' qLock $ \riState loop -> do
@@ -1399,10 +1399,10 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
 
 retrySndOp :: AgentClient -> AM () -> AM ()
 retrySndOp c loop = do
-  -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
-  atomically $ endAgentOperation c AOSndNetwork
-  atomically $ throwWhenInactive c
-  atomically $ beginAgentOperation c AOSndNetwork
+  -- end... is in a separate atomically' because if begin... blocks, SUSPENDED won't be sent
+  atomically' $ endAgentOperation c AOSndNetwork
+  atomically' $ throwWhenInactive c
+  atomically' $ beginAgentOperation c AOSndNetwork
   loop
 
 ackMessage' :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AM ()
@@ -1545,7 +1545,7 @@ connRcvQueues = \case
 
 disableConn :: AgentClient -> ConnId -> AM' ()
 disableConn c connId = do
-  atomically $ removeSubscription c connId
+  atomically' $ removeSubscription c connId
   ns <- asks ntfSupervisor
   atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCDelete)
 
@@ -1589,7 +1589,7 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do
     rcvQueues (SomeConn _ conn) = case connRcvQueues conn of
       [] -> Left $ Right ()
       rqs -> Right rqs
-    notify = atomically . writeTBQueue (subQ c)
+    notify = atomically' . writeTBQueue (subQ c)
 
 deleteConnQueues :: AgentClient -> Bool -> Bool -> [RcvQueue] -> AM' (Map ConnId (Either AgentErrorType ()))
 deleteConnQueues c waitDelivery ntf rqs = do
@@ -1618,7 +1618,7 @@ deleteConnQueues c waitDelivery ntf rqs = do
             | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> incRcvDeleteErrors db rq $> ((rq, r), Nothing)
             | otherwise -> deleteConnRcvQueue db rq $> ((rq, Right ()), Just (notifyRQ rq (Just e)))
     notifyRQ rq e_ = notify ("", qConnId rq, APC SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
-    notify = when ntf . atomically . writeTBQueue (subQ c)
+    notify = when ntf . atomically' . writeTBQueue (subQ c)
     connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
     connResults = M.map snd . foldl' addResult M.empty
       where
@@ -1653,7 +1653,7 @@ deleteConnections_ getConnections ntf waitDelivery c connIds = do
     notifyResultError rs = do
       let actual = M.size rs
           expected = length connIds
-      when (actual /= expected) . atomically $
+      when (actual /= expected) . atomically' $
         writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)
 
 getConnectionServers' :: AgentClient -> ConnId -> AM ConnectionStats
@@ -1713,7 +1713,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
         (Just tknId, Just NTACheck)
           | savedDeviceToken == suppliedDeviceToken -> do
               ns <- asks ntfSupervisor
-              atomically $ nsUpdateToken ns tkn {ntfMode = suppliedNtfMode}
+              atomically' $ nsUpdateToken ns tkn {ntfMode = suppliedNtfMode}
               when (ntfTknStatus == NTActive) $ do
                 cron <- asks $ ntfCron . config
                 agentNtfEnableCron c tknId tkn cron
@@ -1726,7 +1726,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
           agentNtfDeleteToken c tknId tkn
           withStore' c (`removeNtfToken` tkn)
           ns <- asks ntfSupervisor
-          atomically $ nsRemoveNtfToken ns
+          atomically' $ nsRemoveNtfToken ns
           pure NTExpired
         _ -> pure ntfTknStatus
       withStore' c $ \db -> updateNtfMode db tkn suppliedNtfMode
@@ -1740,13 +1740,13 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
               then throwError e
               else do
                 withStore' c $ \db -> removeNtfToken db tkn
-                atomically $ nsRemoveNtfToken ns
+                atomically' $ nsRemoveNtfToken ns
                 createToken
           where
             tryReplace ns = do
               agentNtfReplaceToken c tknId tkn suppliedDeviceToken
               withStore' c $ \db -> updateDeviceToken db tkn suppliedDeviceToken
-              atomically $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
+              atomically' $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
               pure NTRegistered
     _ -> createToken
   where
@@ -1771,7 +1771,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
       let dhSecret = C.dh' srvPubDhKey privDhKey
       withStore' c $ \db -> updateNtfTokenRegistration db tkn tknId dhSecret
       ns <- asks ntfSupervisor
-      atomically $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
+      atomically' $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
 
 verifyNtfToken' :: AgentClient -> DeviceToken -> C.CbNonce -> ByteString -> AM ()
 verifyNtfToken' c deviceToken nonce code =
@@ -1834,7 +1834,7 @@ toggleConnectionNtfs' c connId enable = do
           withStore' c $ \db -> setConnectionNtfs db connId enable
           ns <- asks ntfSupervisor
           let cmd = if enable then NSCCreate else NSCDelete
-          atomically $ sendNtfSubCommand ns (connId, cmd)
+          atomically' $ sendNtfSubCommand ns (connId, cmd)
 
 deleteToken_ :: AgentClient -> NtfToken -> AM ()
 deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
@@ -1842,28 +1842,28 @@ deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
   forM_ ntfTokenId $ \tknId -> do
     let ntfTknAction = Just NTADelete
     withStore' c $ \db -> updateNtfToken db tkn ntfTknStatus ntfTknAction
-    atomically $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
+    atomically' $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
     agentNtfDeleteToken c tknId tkn `catchAgentError` \case
       NTF AUTH -> pure ()
       e -> throwError e
   withStore' c $ \db -> removeNtfToken db tkn
-  atomically $ nsRemoveNtfToken ns
+  atomically' $ nsRemoveNtfToken ns
 
 withToken :: AgentClient -> NtfToken -> Maybe (NtfTknStatus, NtfTknAction) -> (NtfTknStatus, Maybe NtfTknAction) -> AM a -> AM NtfTknStatus
 withToken c tkn@NtfToken {deviceToken, ntfMode} from_ (toStatus, toAction_) f = do
   ns <- asks ntfSupervisor
   forM_ from_ $ \(status, action) -> do
     withStore' c $ \db -> updateNtfToken db tkn status (Just action)
-    atomically $ nsUpdateToken ns tkn {ntfTknStatus = status, ntfTknAction = Just action}
+    atomically' $ nsUpdateToken ns tkn {ntfTknStatus = status, ntfTknAction = Just action}
   tryError f >>= \case
     Right _ -> do
       withStore' c $ \db -> updateNtfToken db tkn toStatus toAction_
       let updatedToken = tkn {ntfTknStatus = toStatus, ntfTknAction = toAction_}
-      atomically $ nsUpdateToken ns updatedToken
+      atomically' $ nsUpdateToken ns updatedToken
       pure toStatus
     Left e@(NTF AUTH) -> do
       withStore' c $ \db -> removeNtfToken db tkn
-      atomically $ nsRemoveNtfToken ns
+      atomically' $ nsRemoveNtfToken ns
       void $ registerNtfToken' c deviceToken ntfMode
       throwError e
     Left e -> throwError e
@@ -1875,13 +1875,13 @@ initializeNtfSubs c = sendNtfConnCommands c NSCCreate
 deleteNtfSubs :: AgentClient -> NtfSupervisorCommand -> AM ()
 deleteNtfSubs c deleteCmd = do
   ns <- asks ntfSupervisor
-  void . atomically . flushTBQueue $ ntfSubQ ns
+  void . atomically' . flushTBQueue $ ntfSubQ ns
   sendNtfConnCommands c deleteCmd
 
 sendNtfConnCommands :: AgentClient -> NtfSupervisorCommand -> AM ()
 sendNtfConnCommands c cmd = do
   ns <- asks ntfSupervisor
-  connIds <- atomically $ getSubscriptions c
+  connIds <- atomically' $ getSubscriptions c
   forM_ connIds $ \connId -> do
     withStore' c (`getConnData` connId) >>= \case
       Just (ConnData {enableNtfs}, _) ->
@@ -1910,7 +1910,7 @@ suspendAgent c 0 = do
     suspend opSel = atomically $ modifyTVar' (opSel c) $ \s -> s {opSuspended = True}
 suspendAgent c@AgentClient {agentState = as} maxDelay = do
   state <-
-    atomically $ do
+    atomically' $ do
       writeTVar as ASSuspending
       suspendOperation c AONtfNetwork $ pure ()
       suspendOperation c AORcvNetwork $
@@ -1920,7 +1920,7 @@ suspendAgent c@AgentClient {agentState = as} maxDelay = do
   when (state == ASSuspending) . void . forkIO $ do
     threadDelay maxDelay
     -- liftIO $ putStrLn "suspendAgent after timeout"
-    atomically . whenSuspending c $ do
+    atomically' . whenSuspending c $ do
       -- unsafeIOToSTM $ putStrLn $ "in timeout: suspendSendingAndDatabase"
       suspendSendingAndDatabase c
 
@@ -1934,10 +1934,10 @@ debugAgentLocks :: AgentClient -> IO AgentLocks
 debugAgentLocks AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do
   connLocks <- getLocks cs
   invLocks <- getLocks is
-  delLock <- atomically $ tryReadTMVar d
+  delLock <- atomically' $ tryReadTMVar d
   pure AgentLocks {connLocks, invLocks, delLock}
   where
-    getLocks ls = atomically $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)
+    getLocks ls = atomically' $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)
 
 getSMPServer :: AgentClient -> UserId -> AM SMPServerWithAuth
 getSMPServer c userId = withUserServers c userId pickServer
@@ -1945,7 +1945,7 @@ getSMPServer c userId = withUserServers c userId pickServer
 
 subscriber :: AgentClient -> AM' ()
 subscriber c@AgentClient {msgQ} = forever $ do
-  t <- atomically $ readTBQueue msgQ
+  t <- atomically' $ readTBQueue msgQ
   agentOperationBracket c AORcvNetwork waitUntilActive $
     runExceptT (processSMPTransmission c t) >>= \case
       Left e -> liftIO $ print e
@@ -1977,7 +1977,7 @@ cleanupManager c@AgentClient {subQ} = do
       step <- asks $ cleanupStepInterval . config
       liftIO $ threadDelay step
     -- we are catching it to avoid CRITICAL errors in tests when this is the only remaining handle to active
-    waitActive a = liftIO (E.tryAny . atomically $ waitUntilActive c) >>= either (\_ -> pure ()) (\_ -> void a)
+    waitActive a = liftIO (E.tryAny . atomically' $ waitUntilActive c) >>= either (\_ -> pure ()) (\_ -> void a)
     deleteConns =
       withLock (deleteLock c) "cleanupManager" $ do
         void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c
@@ -2042,7 +2042,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
               ack' <- handleNotifyAck $ case msg' of
                 SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} -> processClientMsg srvTs msgFlags msgBody
                 SMP.ClientRcvMsgQuota {} -> queueDrained >> ack
-              whenM (atomically $ hasGetLock c rq) $
+              whenM (atomically' $ hasGetLock c rq) $
                 notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg')
               pure ack'
             where
@@ -2204,7 +2204,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
             notify . ERR $ BROKER (B.unpack $ strEncode srv) UNEXPECTED
         where
           notify :: forall e m. MonadIO m => AEntityI e => ACommand 'Agent e -> m ()
-          notify = atomically . notify'
+          notify = atomically' . notify'
 
           notify' :: forall e. AEntityI e => ACommand 'Agent e -> STM ()
           notify' msg = writeTBQueue subQ ("", connId, APC (sAEntity @e) msg)
@@ -2307,7 +2307,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
             case findQ addr sqs of
               Just sq -> do
                 logServer "<--" c srv rId $ "MSG <QCONT>:" <> logSecret srvMsgId
-                atomically $
+                atomically' $
                   TM.lookup (qAddress sq) (smpDeliveryWorkers c)
                     >>= mapM_ (\(_, retryLock) -> tryPutTMVar retryLock ())
               Nothing -> qError "QCONT: queue address not found"

src/Simplex/Messaging/Agent/Client.hs

@@ -27,6 +27,7 @@ module Simplex.Messaging.Agent.Client
     withConnLock,
     withConnLocks,
     withInvLock,
+    withLockMap,
     closeAgentClient,
     closeProtocolServerClients,
     reconnectServerClients,
@@ -142,6 +143,7 @@ import Crypto.Random (ChaChaDRG)
 import qualified Data.Aeson as J
 import qualified Data.Aeson.TH as J
 import Data.Bifunctor (bimap, first, second)
+import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Composition ((.:.))
@@ -160,12 +162,13 @@ import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
+import GHC.Stack (HasCallStack, withFrozenCallStack)
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
 import qualified Simplex.FileTransfer.Client as X
 import Simplex.FileTransfer.Description (ChunkReplicaId (..), FileDigest (..), kb)
 import Simplex.FileTransfer.Protocol (FileInfo (..), FileResponse)
-import Simplex.FileTransfer.Transport (XFTPRcvChunkSpec (..), XFTPErrorType (DIGEST), XFTPVersion)
+import Simplex.FileTransfer.Transport (XFTPErrorType (DIGEST), XFTPRcvChunkSpec (..), XFTPVersion)
 import Simplex.FileTransfer.Types (DeletedSndChunkReplica (..), NewSndChunkReplica (..), RcvFileChunkReplica (..), SndFileChunk (..), SndFileChunkReplica (..))
 import Simplex.FileTransfer.Util (uniqueCombine)
 import Simplex.Messaging.Agent.Env.SQLite
@@ -181,7 +184,6 @@ import Simplex.Messaging.Client
 import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Encoding.Base64 (encode)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Client
 import Simplex.Messaging.Notifications.Protocol
@@ -195,6 +197,7 @@ import Simplex.Messaging.Protocol
     ErrorType,
     MsgFlags (..),
     MsgId,
+    NtfPublicAuthKey,
     NtfServer,
     NtfServerWithAuth,
     ProtoServer,
@@ -206,22 +209,21 @@ import Simplex.Messaging.Protocol
     QueueIdsKeys (..),
     RcvMessage (..),
     RcvNtfPublicDhKey,
-    NtfPublicAuthKey,
     SMPMsgMeta (..),
     SProtocolType (..),
     SndPublicAuthKey,
     SubscriptionMode (..),
     UserProtocol,
+    VersionRangeSMPC,
+    VersionSMPC,
     XFTPServer,
     XFTPServerWithAuth,
-    VersionSMPC,
-    VersionRangeSMPC,
     sameSrvAddr',
   )
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Transport (SMPVersion)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Transport (SMPVersion)
 import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
@@ -299,7 +301,7 @@ getAgentWorker = getAgentWorker' id pure
 
 getAgentWorker' :: forall a k. (Ord k, Show k) => (a -> Worker) -> (Worker -> STM a) -> String -> Bool -> AgentClient -> k -> TMap k a -> (a -> AM ()) -> AM' a
 getAgentWorker' toW fromW name hasWork c key ws work = do
-  atomically (getWorker >>= maybe createWorker whenExists) >>= \w -> runWorker w $> w
+  atomically' (getWorker >>= maybe createWorker whenExists) >>= \w -> runWorker w $> w
   where
     getWorker = TM.lookup key ws
     createWorker = do
@@ -318,7 +320,7 @@ getAgentWorker' toW fromW name hasWork c key ws work = do
           t <- liftIO getSystemTime
           maxRestarts <- asks $ maxWorkerRestartsPerMin . config
           -- worker may terminate because it was deleted from the map (getWorker returns Nothing), then it won't restart
-          restart <- atomically $ getWorker >>= maybe (pure False) (shouldRestart e_ (toW w) t maxRestarts)
+          restart <- atomically' $ getWorker >>= maybe (pure False) (shouldRestart e_ (toW w) t maxRestarts)
           when restart runWork
         shouldRestart e_ Worker {workerId = wId, doWork, action, restarts} t maxRestarts w'
           | wId == workerId (toW w') =
@@ -354,11 +356,11 @@ newWorker c = do
 runWorkerAsync :: Worker -> AM' () -> AM' ()
 runWorkerAsync Worker {action} work =
   E.bracket
-    (atomically $ takeTMVar action) -- get current action, locking to avoid race conditions
-    (atomically . tryPutTMVar action) -- if it was running (or if start crashes), put it back and unlock (don't lock if it was just started)
+    (atomically' $ takeTMVar action) -- get current action, locking to avoid race conditions
+    (atomically' . tryPutTMVar action) -- if it was running (or if start crashes), put it back and unlock (don't lock if it was just started)
     (\a -> when (isNothing a) start) -- start worker if it's not running
   where
-    start = atomically . putTMVar action . Just =<< async work
+    start = atomically' . putTMVar action . Just =<< async work
 
 data AgentOperation = AONtfNetwork | AORcvNetwork | AOMsgDelivery | AOSndNetwork | AODatabase
   deriving (Eq, Show)
@@ -516,7 +518,7 @@ instance ProtocolServerClient XFTPVersion XFTPErrorType FileResponse where
 getSMPServerClient :: AgentClient -> SMPTransportSession -> AM SMPClient
 getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getTSessVar c tSess smpClients)
+  atomically' (getTSessVar c tSess smpClients)
     >>= either newClient (waitForProtocolClient c tSess)
   where
     -- we resubscribe only on newClient error, but not on waitForProtocolClient error,
@@ -531,7 +533,8 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
       g <- asks random
       env <- ask
       liftError' (protocolClientError SMP $ B.unpack $ strEncode srv) $
-        getProtocolClient g tSess cfg (Just msgQ) $ clientDisconnected env v
+        getProtocolClient g tSess cfg (Just msgQ) $
+          clientDisconnected env v
 
     clientDisconnected :: Env -> SMPClientVar -> SMPClient -> IO ()
     clientDisconnected env v client = do
@@ -542,7 +545,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
         -- because we can have a race condition when a new current client could have already
         -- made subscriptions active, and the old client would be processing diconnection later.
         removeClientAndSubs :: IO ([RcvQueue], [ConnId])
-        removeClientAndSubs = atomically $ ifM currentActiveClient removeSubs $ pure ([], [])
+        removeClientAndSubs = atomically' $ ifM currentActiveClient removeSubs $ pure ([], [])
           where
             currentActiveClient = (&&) <$> removeTSessVar' v tSess smpClients <*> readTVar active
             removeSubs = do
@@ -556,7 +559,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
           notifySub "" $ hostEvent DISCONNECT client
           unless (null conns) $ notifySub "" $ DOWN srv conns
           unless (null qs) $ do
-            atomically $ mapM_ (releaseGetLock c) qs
+            atomically' $ mapM_ (releaseGetLock c) qs
             runReaderT (resubscribeSMPSession c tSess) env
 
         notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
@@ -564,7 +567,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
 
 resubscribeSMPSession :: AgentClient -> SMPTransportSession -> AM' ()
 resubscribeSMPSession c@AgentClient {smpSubWorkers} tSess =
-  atomically getWorkerVar >>= mapM_ (either newSubWorker (\_ -> pure ()))
+  atomically' getWorkerVar >>= mapM_ (either newSubWorker (\_ -> pure ()))
   where
     getWorkerVar =
       ifM
@@ -572,13 +575,13 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers} tSess =
         (pure Nothing) -- prevent race with cleanup and adding pending queues in another call
         (Just <$> getTSessVar c tSess smpSubWorkers)
     newSubWorker v = do
-      a <- async $ void (E.tryAny runSubWorker) >> atomically (cleanup v)
-      atomically $ putTMVar (sessionVar v) a
+      a <- async $ void (E.tryAny runSubWorker) >> atomically' (cleanup v)
+      atomically' $ putTMVar (sessionVar v) a
     runSubWorker = do
       ri <- asks $ reconnectInterval . config
       timeoutCounts <- newTVarIO 0
       withRetryInterval ri $ \_ loop -> do
-        pending <- atomically getPending
+        pending <- atomically' getPending
         forM_ (L.nonEmpty pending) $ \qs -> do
           void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
           loop
@@ -624,7 +627,7 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
 getNtfServerClient :: AgentClient -> NtfTransportSession -> AM NtfClient
 getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getTSessVar c tSess ntfClients)
+  atomically' (getTSessVar c tSess ntfClients)
     >>= either
       (newProtocolClient c tSess ntfClients connectClient)
       (waitForProtocolClient c tSess)
@@ -634,11 +637,12 @@ getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = d
       cfg <- lift $ getClientConfig c ntfCfg
       g <- asks random
       liftError' (protocolClientError NTF $ B.unpack $ strEncode srv) $
-        getProtocolClient g tSess cfg Nothing $ clientDisconnected v
+        getProtocolClient g tSess cfg Nothing $
+          clientDisconnected v
 
     clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
     clientDisconnected v client = do
-      atomically $ removeTSessVar v tSess ntfClients
+      atomically' $ removeTSessVar v tSess ntfClients
       incClientStat c userId client "DISCONNECT" ""
       atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
@@ -646,7 +650,7 @@ getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = d
 getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
 getXFTPServerClient c@AgentClient {active, xftpClients, useNetworkConfig} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getTSessVar c tSess xftpClients)
+  atomically' (getTSessVar c tSess xftpClients)
     >>= either
       (newProtocolClient c tSess xftpClients connectClient)
       (waitForProtocolClient c tSess)
@@ -654,13 +658,15 @@ getXFTPServerClient c@AgentClient {active, xftpClients, useNetworkConfig} tSess@
     connectClient :: XFTPClientVar -> AM XFTPClient
     connectClient v = do
       cfg <- asks $ xftpCfg . config
+      g <- asks random
       xftpNetworkConfig <- readTVarIO useNetworkConfig
       liftError' (protocolClientError XFTP $ B.unpack $ strEncode srv) $
-        X.getXFTPClient tSess cfg {xftpNetworkConfig} $ clientDisconnected v
+        X.getXFTPClient g tSess cfg {xftpNetworkConfig} $
+          clientDisconnected v
 
     clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
     clientDisconnected v client = do
-      atomically $ removeTSessVar v tSess xftpClients
+      atomically' $ removeTSessVar v tSess xftpClients
       incClientStat c userId client "DISCONNECT" ""
       atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
@@ -689,7 +695,7 @@ removeTSessVar' v tSess vs =
 waitForProtocolClient :: ProtocolTypeI (ProtoType msg) => AgentClient -> TransportSession msg -> ClientVar msg -> AM (Client msg)
 waitForProtocolClient c (_, srv, _) v = do
   NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v)
+  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically' (readTMVar $ sessionVar v)
   liftEither $ case client_ of
     Just (Right smpClient) -> Right smpClient
     Just (Left e) -> Left e
@@ -709,13 +715,13 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
   tryAgentError (connectClient v) >>= \case
     Right client -> do
       logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
-      atomically $ putTMVar (sessionVar v) (Right client)
+      atomically' $ putTMVar (sessionVar v) (Right client)
       liftIO $ incClientStat c userId client "CLIENT" "OK"
       atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
       pure client
     Left e -> do
       liftIO $ incServerStat c userId srv "CLIENT" $ strEncode e
-      atomically $ do
+      atomically' $ do
         removeTSessVar v tSess clients
         putTMVar (sessionVar v) (Left e)
       throwError e -- signal error to caller
@@ -735,26 +741,26 @@ closeAgentClient c = do
   closeProtocolServerClients c smpClients
   closeProtocolServerClients c ntfClients
   closeProtocolServerClients c xftpClients
-  atomically (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
+  atomically' (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
   clearWorkers smpDeliveryWorkers >>= mapM_ (cancelWorker . fst)
   clearWorkers asyncCmdWorkers >>= mapM_ cancelWorker
   clear connCmdsQueued
-  atomically . RQ.clear $ activeSubs c
-  atomically . RQ.clear $ pendingSubs c
+  atomically' . RQ.clear $ activeSubs c
+  atomically' . RQ.clear $ pendingSubs c
   clear subscrConns
   clear getMsgLocks
   where
     clearWorkers :: Ord k => (AgentClient -> TMap k a) -> IO (Map k a)
-    clearWorkers workers = atomically $ swapTVar (workers c) mempty
+    clearWorkers workers = atomically' $ swapTVar (workers c) mempty
     clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
     clear sel = atomically $ writeTVar (sel c) mempty
     cancelReconnect :: SessionVar (Async ()) -> IO ()
-    cancelReconnect v = void . forkIO $ atomically (readTMVar $ sessionVar v) >>= uninterruptibleCancel
+    cancelReconnect v = void . forkIO $ atomically' (readTMVar $ sessionVar v) >>= uninterruptibleCancel
 
 cancelWorker :: Worker -> IO ()
 cancelWorker Worker {doWork, action} = do
   noWorkToDo doWork
-  atomically (tryTakeTMVar action) >>= mapM_ (mapM_ uninterruptibleCancel)
+  atomically' (tryTakeTMVar action) >>= mapM_ (mapM_ uninterruptibleCancel)
 
 waitUntilActive :: AgentClient -> STM ()
 waitUntilActive c = unlessM (readTVar $ active c) retry
@@ -772,7 +778,7 @@ throwWhenNoDelivery c sq =
 
 closeProtocolServerClients :: ProtocolServerClient v err msg => AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO ()
 closeProtocolServerClients c clientsSel =
-  atomically (clientsSel c `swapTVar` M.empty) >>= mapM_ (forkIO . closeClient_ c)
+  atomically' (clientsSel c `swapTVar` M.empty) >>= mapM_ (forkIO . closeClient_ c)
 
 reconnectServerClients :: ProtocolServerClient v err msg => AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO ()
 reconnectServerClients c clientsSel =
@@ -785,7 +791,7 @@ closeClient c clientSel tSess =
 closeClient_ :: ProtocolServerClient v err msg => AgentClient -> ClientVar msg -> IO ()
 closeClient_ c v = do
   NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v) >>= \case
+  tcpConnectTimeout `timeout` atomically' (readTMVar $ sessionVar v) >>= \case
     Just (Right client) -> closeProtocolServerClient client `catchAll_` pure ()
     _ -> pure ()
 
@@ -799,7 +805,7 @@ withConnLock c connId name = ExceptT . withConnLock' c connId name . runExceptT
 
 withConnLock' :: AgentClient -> ConnId -> String -> AM' a -> AM' a
 withConnLock' _ "" _ = id
-withConnLock' AgentClient {connLocks} connId name = withLockMap_ connLocks connId name
+withConnLock' AgentClient {connLocks} connId name = withLockMap connLocks connId name
 {-# INLINE withConnLock' #-}
 
 withInvLock :: AgentClient -> ByteString -> String -> AM a -> AM a
@@ -807,16 +813,16 @@ withInvLock c key name = ExceptT . withInvLock' c key name . runExceptT
 {-# INLINE withInvLock #-}
 
 withInvLock' :: AgentClient -> ByteString -> String -> AM' a -> AM' a
-withInvLock' AgentClient {invLocks} = withLockMap_ invLocks
+withInvLock' AgentClient {invLocks} = withLockMap invLocks
 {-# INLINE withInvLock' #-}
 
 withConnLocks :: AgentClient -> [ConnId] -> String -> AM' a -> AM' a
 withConnLocks AgentClient {connLocks} = withLocksMap_ connLocks . filter (not . B.null)
 {-# INLINE withConnLocks #-}
 
-withLockMap_ :: (Ord k, MonadUnliftIO m) => TMap k Lock -> k -> String -> m a -> m a
-withLockMap_ = withGetLock . getMapLock
-{-# INLINE withLockMap_ #-}
+withLockMap :: (Ord k, MonadUnliftIO m) => TMap k Lock -> k -> String -> m a -> m a
+withLockMap = withGetLock . getMapLock
+{-# INLINE withLockMap #-}
 
 withLocksMap_ :: (Ord k, MonadUnliftIO m) => TMap k Lock -> [k] -> String -> m a -> m a
 withLocksMap_ = withGetLocks . getMapLock
@@ -951,7 +957,7 @@ runXFTPServerTest c userId (ProtoServerWithAuth srv auth) = do
   rcvPath <- getTempFilePath workDir
   liftIO $ do
     let tSess = (userId, srv, Nothing)
-    X.getXFTPClient tSess cfg {xftpNetworkConfig} (\_ -> pure ()) >>= \case
+    X.getXFTPClient g tSess cfg {xftpNetworkConfig} (\_ -> pure ()) >>= \case
       Right xftp -> withTestChunk filePath $ do
         (sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
         (rcvKey, rpKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -1076,7 +1082,7 @@ processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> IO (E
 processSubResult c rq r = do
   case r of
     Left e ->
-      unless (temporaryClientError e) . atomically $ do
+      unless (temporaryClientError e) . atomically' $ do
         RQ.deleteQueue rq (pendingSubs c)
         TM.insert (RQ.qKey rq) e (removedSubs c)
     _ -> addSubscription c rq
@@ -1100,7 +1106,7 @@ temporaryOrHostError = \case
 subscribeQueues :: AgentClient -> [RcvQueue] -> AM' [(RcvQueue, Either AgentErrorType ())]
 subscribeQueues c qs = do
   (errs, qs') <- partitionEithers <$> mapM checkQueue qs
-  atomically $ do
+  atomically' $ do
     modifyTVar' (subscrConns c) (`S.union` S.fromList (map qConnId qs'))
     RQ.batchAddQueues (pendingSubs c) qs'
   env <- ask
@@ -1108,7 +1114,7 @@ subscribeQueues c qs = do
   (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ env) c qs'
   where
     checkQueue rq = do
-      prohibited <- atomically $ hasGetLock c rq
+      prohibited <- atomically' $ hasGetLock c rq
       pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
     subscribeQueues_ :: Env -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
     subscribeQueues_ env smp qs' = do
@@ -1154,7 +1160,7 @@ sendBatch smpCmdFunc smp qs = L.zip qs <$> smpCmdFunc smp (L.map queueCreds qs)
     queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)
 
 addSubscription :: AgentClient -> RcvQueue -> IO ()
-addSubscription c rq@RcvQueue {connId} = atomically $ do
+addSubscription c rq@RcvQueue {connId} = atomically' $ do
   modifyTVar' (subscrConns c) $ S.insert connId
   RQ.addQueue rq $ activeSubs c
   RQ.deleteQueue rq $ pendingSubs c
@@ -1211,7 +1217,7 @@ sendInvitation c userId (Compatible (SMPQueueInfo v SMPQueueAddress {smpServer,
 
 getQueueMessage :: AgentClient -> RcvQueue -> AM (Maybe SMPMsgMeta)
 getQueueMessage c rq@RcvQueue {server, rcvId, rcvPrivateKey} = do
-  atomically createTakeGetLock
+  atomically' createTakeGetLock
   msg_ <- withSMPClient c rq "GET" $ \smp ->
     getSMPMessage smp rcvPrivateKey rcvId
   mapM decryptMeta msg_
@@ -1261,7 +1267,7 @@ sendAck :: AgentClient -> RcvQueue -> MsgId -> AM ()
 sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = do
   withSMPClient c rq ("ACK:" <> logSecret msgId) $ \smp ->
     ackSMPMessage smp rcvPrivateKey rcvId msgId
-  atomically $ releaseGetLock c rq
+  atomically' $ releaseGetLock c rq
 
 hasGetLock :: AgentClient -> RcvQueue -> STM Bool
 hasGetLock c RcvQueue {server, rcvId} =
@@ -1359,7 +1365,7 @@ agentXFTPDeleteChunk c userId DeletedSndChunkReplica {server, replicaId = ChunkR
 
 xftpRcvKeys :: Int -> AM (NonEmpty C.AAuthKeyPair)
 xftpRcvKeys n = do
-  rKeys <- atomically . replicateM n . C.generateAuthKeyPair C.SEd25519 =<< asks random
+  rKeys <- atomically' . replicateM n . C.generateAuthKeyPair C.SEd25519 =<< asks random
   case L.nonEmpty rKeys of
     Just rKeys' -> pure rKeys'
     _ -> throwError $ INTERNAL "non-positive number of recipients"
@@ -1369,7 +1375,7 @@ xftpRcvIdsKeys rIds rKeys = L.map ChunkReplicaId rIds `L.zip` L.map snd rKeys
 
 agentCbEncrypt :: SndQueue -> Maybe C.PublicKeyX25519 -> ByteString -> AM ByteString
 agentCbEncrypt SndQueue {e2eDhSecret, smpClientVersion} e2ePubKey msg = do
-  cmNonce <- atomically . C.randomCbNonce =<< asks random
+  cmNonce <- atomically' . C.randomCbNonce =<< asks random
   let paddedLen = maybe SMP.e2eEncMessageLength (const SMP.e2eEncConfirmationLength) e2ePubKey
   cmEncBody <-
     liftEither . first cryptoError $
@@ -1411,8 +1417,8 @@ cryptoError = \case
   where
     c = AGENT . A_CRYPTO
 
-waitForWork :: MonadIO m => TMVar () -> m ()
-waitForWork = void . atomically . readTMVar
+waitForWork :: (MonadIO m, HasCallStack) => TMVar () -> m ()
+waitForWork v = withFrozenCallStack $ void . atomically' $ readTMVar v
 {-# INLINE waitForWork #-}
 
 withWork :: AgentClient -> TMVar () -> (DB.Connection -> IO (Either StoreError (Maybe a))) -> (a -> AM ()) -> AM ()
@@ -1427,7 +1433,7 @@ withWork c doWork getWork action =
     notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err $ show e)
 
 noWorkToDo :: TMVar () -> IO ()
-noWorkToDo = void . atomically . tryTakeTMVar
+noWorkToDo = void . atomically' . tryTakeTMVar
 {-# INLINE noWorkToDo #-}
 
 hasWorkToDo :: Worker -> STM ()
@@ -1494,8 +1500,8 @@ beginAgentOperation c op = do
 agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> (AgentClient -> STM ()) -> m a -> m a
 agentOperationBracket c op check action =
   E.bracket
-    (atomically (check c) >> atomically (beginAgentOperation c op))
-    (\_ -> atomically $ endAgentOperation c op)
+    (atomically' (check c) >> atomically' (beginAgentOperation c op))
+    (\_ -> atomically' $ endAgentOperation c op)
     (const action)
 
 waitUntilForeground :: AgentClient -> STM ()
@@ -1556,13 +1562,13 @@ incClientStat c userId pc = incClientStatN c userId pc 1
 incServerStat :: AgentClient -> UserId -> ProtocolServer p -> ByteString -> ByteString -> IO ()
 incServerStat c userId ProtocolServer {host} cmd res = do
   threadDelay 100000
-  atomically $ incStat c 1 statsKey
+  atomically' $ incStat c 1 statsKey
   where
     statsKey = AgentStatsKey {userId, host = strEncode $ L.head host, clientTs = "", cmd, res}
 
 incClientStatN :: ProtocolServerClient v err msg => AgentClient -> UserId -> Client msg -> Int -> ByteString -> ByteString -> IO ()
 incClientStatN c userId pc n cmd res = do
-  atomically $ incStat c n statsKey
+  atomically' $ incStat c n statsKey
   where
     statsKey = AgentStatsKey {userId, host = strEncode $ clientTransportHost pc, clientTs = strEncode $ clientSessionTs pc, cmd, res}
 
@@ -1577,7 +1583,7 @@ pickServer = \case
   srv :| [] -> pure srv
   servers -> do
     gen <- asks randomServer
-    atomically $ (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
+    atomically' $ (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
 
 getNextServer :: forall p. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> [ProtocolServer p] -> AM (ProtoServerWithAuth p)
 getNextServer c userId usedSrvs = withUserServers c userId $ \srvs ->
@@ -1595,7 +1601,7 @@ withNextSrv :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> U
 withNextSrv c userId usedSrvs initUsed action = do
   used <- readTVarIO usedSrvs
   srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId used
-  atomically $ do
+  atomically' $ do
     srvs_ <- TM.lookup userId $ userServers c
     let unused = maybe [] ((\\ used) . map protoServer . L.toList) srvs_
         used' = if null unused then initUsed else srv : used
@@ -1683,8 +1689,8 @@ getAgentWorkersDetails AgentClient {smpClients, ntfClients, xftpClients, smpDeli
     workerStats :: StrEncoding k => Map k Worker -> IO (Map Text WorkersDetails)
     workerStats ws = fmap M.fromList . forM (M.toList ws) $ \(qa, Worker {restarts, doWork, action}) -> do
       RestartCount {restartCount} <- readTVarIO restarts
-      hasWork <- atomically $ not <$> isEmptyTMVar doWork
-      hasAction <- atomically $ not <$> isEmptyTMVar action
+      hasWork <- atomically' $ not <$> isEmptyTMVar doWork
+      hasAction <- atomically' $ not <$> isEmptyTMVar action
       pure (textKey qa, WorkersDetails {restarts = restartCount, hasWork, hasAction})
     Env {ntfSupervisor, xftpAgent} = agentEnv
     NtfSupervisor {ntfWorkers, ntfSMPWorkers} = ntfSupervisor
@@ -1749,7 +1755,7 @@ getAgentWorkersSummary AgentClient {smpClients, ntfClients, xftpClients, smpDeli
         byWork WorkersSummary {numActive, numIdle, totalRestarts} Worker {action, restarts} = do
           RestartCount {restartCount} <- readTVarIO restarts
           ifM
-            (atomically $ isJust <$> tryReadTMVar action)
+            (atomically' $ isJust <$> tryReadTMVar action)
             (pure WorkersSummary {numActive, numIdle = numIdle + 1, totalRestarts = totalRestarts + restartCount})
             (pure WorkersSummary {numActive = numActive + 1, numIdle, totalRestarts = totalRestarts + restartCount})
 

src/Simplex/Messaging/Agent/Lock.hs

@@ -15,6 +15,7 @@ import Data.Functor (($>))
 import UnliftIO.Async (forConcurrently)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
+import Simplex.Messaging.Util (atomically')
 
 type Lock = TMVar String
 
@@ -29,23 +30,23 @@ withLock lock name = ExceptT . withLock' lock name . runExceptT
 withLock' :: MonadUnliftIO m => Lock -> String -> m a -> m a
 withLock' lock name =
   E.bracket_
-    (atomically $ putTMVar lock name)
-    (void . atomically $ takeTMVar lock)
+    (atomically' $ putTMVar lock name)
+    (void . atomically' $ takeTMVar lock)
 
 withGetLock :: MonadUnliftIO m => (k -> STM Lock) -> k -> String -> m a -> m a
 withGetLock getLock key name a =
   E.bracket
-    (atomically $ getPutLock getLock key name)
-    (atomically . takeTMVar)
+    (atomically' $ getPutLock getLock key name)
+    (atomically' . takeTMVar)
     (const a)
 
 withGetLocks :: MonadUnliftIO m => (k -> STM Lock) -> [k] -> String -> m a -> m a
 withGetLocks getLock keys name = E.bracket holdLocks releaseLocks . const
   where
-    holdLocks = forConcurrently keys $ \key -> atomically $ getPutLock getLock key name
+    holdLocks = forConcurrently keys $ \key -> atomically' $ getPutLock getLock key name
     -- only this withGetLocks would be holding the locks,
     -- so it's safe to combine all lock releases into one transaction
-    releaseLocks = atomically . mapM_ takeTMVar
+    releaseLocks = atomically' . mapM_ takeTMVar
 
 -- getLock and putTMVar can be in one transaction on the assumption that getLock doesn't write in case the lock already exists,
 -- and in case it is created and added to some shared resource (we use TMap) it also helps avoid contention for the newly created lock.

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -38,7 +38,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol (NtfSubStatus (..), NtfTknStatus (..), SMPQueueNtf (..))
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Protocol (NtfServer, SMPServer, sameSrvAddr)
-import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, unlessM)
+import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, unlessM, atomically')
 import System.Random (randomR)
 import UnliftIO
 import UnliftIO.Concurrent (forkIO, threadDelay)
@@ -48,7 +48,7 @@ runNtfSupervisor :: AgentClient -> AM' ()
 runNtfSupervisor c = do
   ns <- asks ntfSupervisor
   forever $ do
-    cmd@(connId, _) <- atomically . readTBQueue $ ntfSubQ ns
+    cmd@(connId, _) <- atomically' . readTBQueue $ ntfSubQ ns
     handleErr connId . agentOperationBracket c AONtfNetwork waitUntilActive $
       runExceptT (processNtfSub c cmd) >>= \case
         Left e -> notifyErr connId e
@@ -265,7 +265,7 @@ runNtfSMPWorker c srv Worker {doWork} = do
                   setRcvQueueNtfCreds db connId $ Just ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
                   updateNtfSubscription db sub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey} (NtfSubNTFAction NSACreate) ts
                 ns <- asks ntfSupervisor
-                atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
+                atomically' $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
               _ -> workerInternalError c connId "NSASmpKey - no active token"
           NSASmpDelete -> do
             rq_ <- withStore' c $ \db -> do
@@ -278,10 +278,10 @@ rescheduleAction :: TMVar () -> UTCTime -> UTCTime -> AM' Bool
 rescheduleAction doWork ts actionTs
   | actionTs <= ts = pure False
   | otherwise = do
-      void . atomically $ tryTakeTMVar doWork
+      void . atomically' $ tryTakeTMVar doWork
       void . forkIO $ do
         liftIO $ threadDelay' $ diffToMicroseconds $ diffUTCTime actionTs ts
-        atomically $ hasWorkToDo' doWork
+        atomically' $ hasWorkToDo' doWork
       pure True
 
 retryOnError :: AgentClient -> Text -> AM () -> (AgentErrorType -> AM ()) -> AgentErrorType -> AM ()
@@ -293,9 +293,9 @@ retryOnError c name loop done e = do
     _ -> done e
   where
     retryLoop = do
-      atomically $ endAgentOperation c AONtfNetwork
-      atomically $ throwWhenInactive c
-      atomically $ beginAgentOperation c AONtfNetwork
+      atomically' $ endAgentOperation c AONtfNetwork
+      atomically' $ throwWhenInactive c
+      atomically' $ beginAgentOperation c AONtfNetwork
       loop
 
 workerInternalError :: AgentClient -> ConnId -> String -> AM ()
@@ -334,7 +334,7 @@ closeNtfSupervisor ns = do
   stopWorkers $ ntfWorkers ns
   stopWorkers $ ntfSMPWorkers ns
   where
-    stopWorkers workers = atomically (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
+    stopWorkers workers = atomically' (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
 
 getNtfServer :: AgentClient -> AM' (Maybe NtfServer)
 getNtfServer c = do

src/Simplex/Messaging/Agent/Protocol.hs

@@ -163,6 +163,7 @@ import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Aeson.TH as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
@@ -201,7 +202,6 @@ import Simplex.Messaging.Crypto.Ratchet
     SndE2ERatchetParams
   )
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Encoding.Base64 (base64P, encode)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.Protocol

src/Simplex/Messaging/Agent/RetryInterval.hs

@@ -18,7 +18,7 @@ import Control.Concurrent (forkIO)
 import Control.Monad (void)
 import Control.Monad.IO.Class (MonadIO, liftIO)
 import Data.Int (Int64)
-import Simplex.Messaging.Util (threadDelay', whenM)
+import Simplex.Messaging.Util (threadDelay', whenM, atomically')
 import UnliftIO.STM
 
 data RetryInterval = RetryInterval
@@ -82,8 +82,8 @@ withRetryLock2 RetryInterval2 {riSlow, riFast} lock action =
           waiting <- newTVarIO True
           _ <- liftIO . forkIO $ do
             threadDelay' delay
-            atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
-          atomically $ do
+            atomically' $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
+          atomically' $ do
             takeTMVar lock
             writeTVar waiting False
 

src/Simplex/Messaging/Agent/Server.hs

@@ -24,7 +24,7 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
 import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
 import Simplex.Messaging.Transport.Server (defaultTransportServerConfig, loadTLSServerParams, runTransportServer)
-import Simplex.Messaging.Util (bshow)
+import Simplex.Messaging.Util (bshow, atomically')
 import UnliftIO.Async (race_)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -76,7 +76,7 @@ receive h c@AgentClient {rcvQ, subQ} = forever $ do
 
 send :: Transport c => c -> AgentClient -> IO ()
 send h c@AgentClient {subQ} = forever $ do
-  t <- atomically $ readTBQueue subQ
+  t <- atomically' $ readTBQueue subQ
   tPut h t
   logClient c "<--" t
 

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -231,6 +231,7 @@ import Data.Bifunctor (first, second)
 import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
 import Data.ByteString (ByteString)
+import qualified Data.ByteString.Base64.URL as U
 import qualified Data.ByteString.Char8 as B
 import Data.Char (toLower)
 import Data.Functor (($>))
@@ -270,7 +271,6 @@ import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs (..))
 import Simplex.Messaging.Crypto.Ratchet (RatchetX448, SkippedMsgDiff (..), SkippedMsgKeys, PQEncryption (..), PQSupport (..))
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol (DeviceToken (..), NtfSubscriptionId, NtfTknStatus (..), NtfTokenId, SMPQueueNtf (..))
 import Simplex.Messaging.Notifications.Types
@@ -278,7 +278,7 @@ import Simplex.Messaging.Parsers (blobFieldParser, defaultJSON, dropPrefix, from
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport.Client (TransportHost)
-import Simplex.Messaging.Util (bshow, catchAllErrors, eitherToMaybe, ifM, safeDecodeUtf8, ($>>=), (<$$>))
+import Simplex.Messaging.Util (bshow, catchAllErrors, eitherToMaybe, ifM, safeDecodeUtf8, ($>>=), (<$$>), atomically')
 import Simplex.Messaging.Version.Internal
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
@@ -420,11 +420,11 @@ openSQLiteStore st@SQLiteStore {dbClosed} key keepKey =
 openSQLiteStore_ :: SQLiteStore -> ScrubbedBytes -> Bool -> IO ()
 openSQLiteStore_ SQLiteStore {dbConnection, dbFilePath, dbKey, dbClosed} key keepKey =
   bracketOnError
-    (atomically $ takeTMVar dbConnection)
-    (atomically . tryPutTMVar dbConnection)
+    (atomically' $ takeTMVar dbConnection)
+    (atomically' . tryPutTMVar dbConnection)
     $ \DB.Connection {slow} -> do
       DB.Connection {conn} <- connectDB dbFilePath key
-      atomically $ do
+      atomically' $ do
         putTMVar dbConnection DB.Connection {conn, slow}
         writeTVar dbClosed False
         writeTVar dbKey $! storeKey key keepKey
@@ -1214,7 +1214,7 @@ setRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 pqPrivKem =
     db
     [sql|
       UPDATE ratchets
-      SET x3dh_priv_key_1 = ?, x3dh_priv_key_2 = ?, x3dh_pub_key_1 = ?, x3dh_pub_key_2 = ?, pq_priv_kem = ? 
+      SET x3dh_priv_key_1 = ?, x3dh_priv_key_2 = ?, x3dh_pub_key_1 = ?, x3dh_pub_key_2 = ?, pq_priv_kem = ?
       WHERE conn_id = ?
     |]
     (x3dhPrivKey1, x3dhPrivKey2, C.publicKey x3dhPrivKey1, C.publicKey x3dhPrivKey2, pqPrivKem, connId)
@@ -2248,7 +2248,7 @@ createWithRandomId' gVar create = tryCreate 3
           | otherwise -> pure . Left . SEInternal $ bshow e
 
 randomId :: TVar ChaChaDRG -> Int -> IO ByteString
-randomId gVar n = atomically $ U.encode <$> C.randomBytes n gVar
+randomId gVar n = atomically' $ U.encode <$> C.randomBytes n gVar
 
 ntfSubAndSMPAction :: NtfSubAction -> (Maybe NtfSubNTFAction, Maybe NtfSubSMPAction)
 ntfSubAndSMPAction (NtfSubNTFAction action) = (Just action, Nothing)

src/Simplex/Messaging/Agent/Store/SQLite/Common.hs

@@ -21,7 +21,7 @@ import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Database.SQLite.Simple (SQLError)
 import qualified Database.SQLite.Simple as SQL
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
-import Simplex.Messaging.Util (diffToMilliseconds)
+import Simplex.Messaging.Util (diffToMilliseconds, atomically')
 import UnliftIO.Exception (bracket)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -40,8 +40,8 @@ data SQLiteStore = SQLiteStore
 withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
 withConnection SQLiteStore {dbConnection} =
   bracket
-    (atomically $ takeTMVar dbConnection)
-    (atomically . putTMVar dbConnection)
+    (atomically' $ takeTMVar dbConnection)
+    (atomically' . putTMVar dbConnection)
 
 withConnection' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a
 withConnection' st action = withConnection st $ action . DB.conn

src/Simplex/Messaging/Agent/Store/SQLite/DB.hs

@@ -28,7 +28,7 @@ import qualified Database.SQLite.Simple as SQL
 import Simplex.Messaging.Parsers (defaultJSON)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (diffToMilliseconds)
+import Simplex.Messaging.Util (diffToMilliseconds, atomically')
 
 data Connection = Connection
   { conn :: SQL.Connection,
@@ -48,7 +48,7 @@ timeIt slow sql a = do
   r <- a
   t' <- getCurrentTime
   let diff = diffToMilliseconds $ diffUTCTime t' t
-  atomically $ when (diff > 5) $ TM.alter (updateQueryStats diff) sql slow
+  atomically' $ when (diff > 5) $ TM.alter (updateQueryStats diff) sql slow
   pure r
   where
     updateQueryStats :: Int64 -> Maybe SlowQueryStats -> Maybe SlowQueryStats

src/Simplex/Messaging/Client.hs

@@ -110,7 +110,7 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportClientConfig (..), TransportHost (..), runTransportClient)
 import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Transport.WebSockets (WS)
-import Simplex.Messaging.Util (bshow, raceAny_, threadDelay')
+import Simplex.Messaging.Util (bshow, raceAny_, threadDelay', atomically')
 import Simplex.Messaging.Version
 import System.Timeout (timeout)
 
@@ -239,7 +239,7 @@ defaultNetworkConfig =
 
 transportClientConfig :: NetworkConfig -> TransportClientConfig
 transportClientConfig NetworkConfig {socksProxy, tcpKeepAlive, logTLSErrors} =
-  TransportClientConfig {socksProxy, tcpKeepAlive, logTLSErrors, clientCredentials = Nothing}
+  TransportClientConfig {socksProxy, tcpKeepAlive, logTLSErrors, clientCredentials = Nothing, alpn = Nothing}
 {-# INLINE transportClientConfig #-}
 
 -- | protocol client configuration.
@@ -360,8 +360,8 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
       action <-
         async $
           runTransportClient tcConfig (Just username) useHost port' (Just $ keyHash srv) (client t c cVar)
-            `finally` atomically (tryPutTMVar cVar $ Left PCENetworkError)
-      c_ <- tcpConnectTimeout `timeout` atomically (takeTMVar cVar)
+            `finally` atomically' (tryPutTMVar cVar $ Left PCENetworkError)
+      c_ <- tcpConnectTimeout `timeout` atomically' (takeTMVar cVar)
       case c_ of
         Just (Right c') -> pure $ Right c' {action = Just action}
         Just (Left e) -> pure $ Left e
@@ -377,21 +377,21 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
     client _ c cVar h = do
       ks <- atomically $ C.generateKeyPair g
       runExceptT (protocolClientHandshake @v @err @msg h ks (keyHash srv) serverVRange) >>= \case
-        Left e -> atomically . putTMVar cVar . Left $ PCETransportError e
+        Left e -> atomically' . putTMVar cVar . Left $ PCETransportError e
         Right th@THandle {params} -> do
           sessionTs <- getCurrentTime
           let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs}
-          atomically $ do
+          atomically' $ do
             writeTVar (connected c) True
             putTMVar cVar $ Right c'
           raceAny_ ([send c' th, process c', receive c' th] <> [ping c' | smpPingInterval > 0])
             `finally` disconnected c'
 
     send :: Transport c => ProtocolClient v err msg -> THandle v c -> IO ()
-    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPutLog h
+    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically' (readTBQueue sndQ) >>= tPutLog h
 
     receive :: Transport c => ProtocolClient v err msg -> THandle v c -> IO ()
-    receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ
+    receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically' . writeTBQueue rcvQ
 
     ping :: ProtocolClient v err msg -> IO ()
     ping c@ProtocolClient {client_ = PClient {pingErrorCount}} = do
@@ -405,7 +405,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
         maxCnt = smpPingCount networkConfig
 
     process :: ProtocolClient v err msg -> IO ()
-    process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c)
+    process c = forever $ atomically' (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c)
 
     processMsg :: ProtocolClient v err msg -> SignedTransmission err msg -> IO ()
     processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr)) =
@@ -414,7 +414,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
         else do
           atomically (TM.lookup corrId sentCommands) >>= \case
             Nothing -> sendMsg respOrErr
-            Just Request {entityId, responseVar} -> atomically $ do
+            Just Request {entityId, responseVar} -> atomically' $ do
               TM.delete corrId sentCommands
               putTMVar responseVar $ response entityId
       where
@@ -428,7 +428,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
           | otherwise = Left . PCEUnexpectedResponse $ bshow respOrErr
         sendMsg :: Either err msg -> IO ()
         sendMsg = \case
-          Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c entId msg) msgQ
+          Right msg -> atomically' $ mapM_ (`writeTBQueue` serverTransmission c entId msg) msgQ
           Left e -> putStrLn $ "SMP client error: " <> show e
 
 proxyUsername :: TransportSession msg -> ByteString
@@ -525,7 +525,7 @@ processSUBResponse c (Response rId r) = case r of
   Left e -> pure $ Left e
 
 writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
-writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c)
+writeSMPMessage c rId msg = atomically' $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c)
 
 serverTransmission :: ProtocolClient v err msg -> RecipientId -> msg -> ServerTransmission v msg
 serverTransmission ProtocolClient {thParams = THandleParams {thVersion, sessionId}, client_ = PClient {transportSession}} entityId message =
@@ -702,7 +702,7 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
       Left e -> pure . Left $ PCETransportError e
       Right t
         | B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg
-        | otherwise -> atomically (writeTBQueue sndQ s) >> response <$> getResponse c r
+        | otherwise -> atomically' (writeTBQueue sndQ s) >> response <$> getResponse c r
         where
           s
             | batch = tEncodeBatch1 t
@@ -712,17 +712,17 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
 getResponse :: ProtocolClient v err msg -> Request err msg -> IO (Response err msg)
 getResponse ProtocolClient {client_ = PClient {tcpTimeout, pingErrorCount}} Request {entityId, responseVar} = do
   response <-
-    timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case
+    timeout tcpTimeout (atomically' (takeTMVar responseVar)) >>= \case
       Just r -> atomically (writeTVar pingErrorCount 0) $> r
       Nothing -> pure $ Left PCEResponseTimeout
   pure Response {entityId, response}
 
 mkTransmission :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> ClientCommand msg -> IO (PCTransmission err msg)
 mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} (pKey_, entId, cmd) = do
-  corrId <- atomically getNextCorrId
+  corrId <- atomically' getNextCorrId
   let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entId, cmd)
       auth = authTransmission (thAuth thParams) pKey_ corrId tForAuth
-  r <- atomically $ mkRequest corrId
+  r <- atomically' $ mkRequest corrId
   pure ((,tToSend) <$> auth, r)
   where
     getNextCorrId :: STM CorrId

src/Simplex/Messaging/Client/Agent.hs

@@ -43,7 +43,7 @@ import Simplex.Messaging.Protocol (BrokerMsg, NotifierId, NtfPrivateAuthKey, Pro
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, toChunks, ($>>=))
+import Simplex.Messaging.Util (catchAll_, toChunks, ($>>=), atomically')
 import System.Timeout (timeout)
 import UnliftIO (async)
 import UnliftIO.Exception (Exception)
@@ -154,7 +154,7 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
     waitForSMPClient :: SMPClientVar -> ExceptT SMPClientError IO SMPClient
     waitForSMPClient smpVar = do
       let ProtocolClientConfig {networkConfig = NetworkConfig {tcpConnectTimeout}} = smpCfg agentCfg
-      smpClient_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar smpVar)
+      smpClient_ <- liftIO $ tcpConnectTimeout `timeout` atomically' (readTMVar smpVar)
       liftEither $ case smpClient_ of
         Just (Right smpClient) -> Right smpClient
         Just (Left e) -> Left e
@@ -168,12 +168,12 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
           tryE connectClient >>= \r -> case r of
             Right smp -> do
               logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
-              atomically $ putTMVar smpVar r
+              atomically' $ putTMVar smpVar r
               successAction smp
             Left e -> do
               if e == PCENetworkError || e == PCEResponseTimeout
                 then retryAction
-                else atomically $ do
+                else atomically' $ do
                   putTMVar smpVar (Left e)
                   TM.delete srv smpClients
               throwE e
@@ -195,7 +195,7 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
 
     removeClientAndSubs :: IO (Maybe (Map SMPSub C.APrivateAuthKey))
-    removeClientAndSubs = atomically $ do
+    removeClientAndSubs = atomically' $ do
       TM.delete srv smpClients
       TM.lookupDelete srv (srvSubs ca) >>= mapM updateSubs
       where
@@ -229,9 +229,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
     reconnectClient = do
       withSMP ca srv $ \smp -> do
         liftIO $ notify $ CAReconnected srv
-        cs_ <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
+        cs_ <- atomically' $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
         forM_ cs_ $ \cs -> do
-          subs' <- filterM (fmap not . atomically . hasSub (srvSubs ca) srv . fst) $ M.assocs cs
+          subs' <- filterM (fmap not . atomically' . hasSub (srvSubs ca) srv . fst) $ M.assocs cs
           let (nSubs, rSubs) = partition (isNotifier . fst . fst) subs'
           subscribe_ smp SPNotifier nSubs
           subscribe_ smp SPRecipient rSubs
@@ -252,9 +252,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
                     map (\(sub, r) -> bimap (fst sub,) (const sub) r) $ L.toList rs'
                   (errs, oks) = partitionEithers rs''
                   (tempErrs, finalErrs) = partition (temporaryClientError . snd) errs
-              mapM_ (atomically . addSubscription ca srv) oks
+              mapM_ (atomically' . addSubscription ca srv) oks
               mapM_ (liftIO . notify . CAResubscribed srv) $ L.nonEmpty $ map fst oks
-              mapM_ (atomically . removePendingSubscription ca srv . fst) finalErrs
+              mapM_ (atomically' . removePendingSubscription ca srv . fst) finalErrs
               mapM_ (liftIO . notify . CASubError srv) $ L.nonEmpty finalErrs
               mapM_ (throwE . snd) $ listToMaybe tempErrs
 
@@ -271,7 +271,7 @@ closeSMPServerClients :: SMPClientAgent -> IO ()
 closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ (forkIO . closeClient)
   where
     closeClient smpVar =
-      atomically (readTMVar smpVar) >>= \case
+      atomically' (readTMVar smpVar) >>= \case
         Right smp -> closeProtocolClient smp `catchAll_` pure ()
         _ -> pure ()
 
@@ -288,15 +288,15 @@ withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPE
 
 subscribeQueue :: SMPClientAgent -> SMPServer -> (SMPSub, C.APrivateAuthKey) -> ExceptT SMPClientError IO ()
 subscribeQueue ca srv sub = do
-  atomically $ addPendingSubscription ca srv sub
+  atomically' $ addPendingSubscription ca srv sub
   withSMP ca srv $ \smp -> subscribe_ smp `catchE` handleErr
   where
     subscribe_ smp = do
       smpSubscribe smp sub
-      atomically $ addSubscription ca srv sub
+      atomically' $ addSubscription ca srv sub
 
     handleErr e = do
-      atomically . when (e /= PCENetworkError && e /= PCEResponseTimeout) $
+      atomically' . when (e /= PCENetworkError && e /= PCEResponseTimeout) $
         removePendingSubscription ca srv (fst sub)
       throwE e
 
@@ -308,7 +308,7 @@ subscribeQueuesNtfs = subscribeQueues_ SPNotifier
 
 subscribeQueues_ :: SMPSubParty -> SMPClientAgent -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
 subscribeQueues_ party ca srv subs = do
-  atomically $ forM_ subs $ addPendingSubscription ca srv . first (party,)
+  atomically' $ forM_ subs $ addPendingSubscription ca srv . first (party,)
   runExceptT (getSMPServerClient' ca srv) >>= \case
     Left e -> pure $ L.map ((,Left e) . fst) subs
     Right smp -> smpSubscribeQueues party ca smp srv subs
@@ -316,7 +316,7 @@ subscribeQueues_ party ca srv subs = do
 smpSubscribeQueues :: SMPSubParty -> SMPClientAgent -> SMPClient -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
 smpSubscribeQueues party ca smp srv subs = do
   rs <- L.zip subs <$> subscribe smp (L.map swap subs)
-  atomically $ forM rs $ \(sub, r) ->
+  atomically' $ forM rs $ \(sub, r) ->
     (fst sub,) <$> case r of
       Right () -> do
         addSubscription ca srv $ first (party,) sub

src/Simplex/Messaging/Crypto.hs

@@ -211,6 +211,8 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (bimap, first)
 import Data.ByteArray (ByteArrayAccess)
 import qualified Data.ByteArray as BA
+import Data.ByteString.Base64 (decode, encode)
+import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.ByteString.Lazy (fromStrict, toStrict)
@@ -228,8 +230,6 @@ import Database.SQLite.Simple.ToField (ToField (..))
 import GHC.TypeLits (ErrorMessage (..), KnownNat, Nat, TypeError, natVal, type (+))
 import Network.Transport.Internal (decodeWord16, encodeWord16)
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Encoding.Base64 (decode, encode)
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (blobFieldDecoder, parseAll, parseString)
 import Simplex.Messaging.Util ((<$?>))

src/Simplex/Messaging/Crypto/Ratchet.hs

@@ -150,7 +150,7 @@ currentE2EEncryptVersion = VersionE2E 2
 -- TODO v5.7 remove dependency of version range on whether PQ encryption is used
 supportedE2EEncryptVRange :: PQSupport -> VersionRangeE2E
 supportedE2EEncryptVRange pq =
-  mkVersionRange kdfX3DHE2EEncryptVersion $ case pq of 
+  mkVersionRange kdfX3DHE2EEncryptVersion $ case pq of
     PQSupportOn -> pqRatchetE2EEncryptVersion
     PQSupportOff -> currentE2EEncryptVersion
 

src/Simplex/Messaging/Encoding/Base64.hs

@@ -1,29 +0,0 @@
-{-# LANGUAGE OverloadedStrings #-}
-
--- | Compatibility wrappers for base64 package, Base64 (padded) variant.
-module Simplex.Messaging.Encoding.Base64 where
-
-import qualified Data.Attoparsec.ByteString.Char8 as A
-import Data.Base64.Types (extractBase64)
-import Data.Bifunctor (first)
-import Data.ByteString.Base64 (decodeBase64Untyped, encodeBase64')
-import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
-import qualified Data.Text as T
-
-encode :: ByteString -> ByteString
-encode = extractBase64 . encodeBase64'
-{-# INLINE encode #-}
-
-decode :: ByteString -> Either String ByteString
-decode = first T.unpack . decodeBase64Untyped
-{-# INLINE decode #-}
-
-base64P :: A.Parser ByteString
-base64P = do
-  str <- A.takeWhile1 (`B.elem` base64Alphabet)
-  pad <- A.takeWhile (== '=') -- correct amount of padding can be derived from str length
-  either (fail . T.unpack) pure $ decodeBase64Untyped (str <> pad)
-
-base64Alphabet :: ByteString
-base64Alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"

src/Simplex/Messaging/Encoding/Base64/URL.hs

@@ -1,33 +0,0 @@
-{-# LANGUAGE OverloadedStrings #-}
-
--- | Compatibility wrappers for base64 package, Base64URL-padded variant.
-module Simplex.Messaging.Encoding.Base64.URL where
-
-import qualified Data.Attoparsec.ByteString.Char8 as A
-import Data.Base64.Types (extractBase64)
-import Data.Bifunctor (first)
-import Data.ByteString.Base64.URL (decodeBase64Lenient, decodeBase64UnpaddedUntyped, decodeBase64Untyped, encodeBase64')
-import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
-import qualified Data.Text as T
-
-encode :: ByteString -> ByteString
-encode = extractBase64 . encodeBase64'
-{-# INLINE encode #-}
-
-decode :: ByteString -> Either String ByteString
-decode = first T.unpack . decodeBase64Untyped
-{-# INLINE decode #-}
-
-decodeLenient :: ByteString -> ByteString
-decodeLenient = decodeBase64Lenient
-{-# INLINE decodeLenient #-}
-
-base64urlP :: A.Parser ByteString
-base64urlP = do
-  str <- A.takeWhile1 (`B.elem` base64AlphabetURL)
-  _pad <- A.takeWhile (== '=') -- correct amount of padding can be derived from str length
-  either (fail . T.unpack) pure $ decodeBase64UnpaddedUntyped str
-
-base64AlphabetURL :: ByteString
-base64AlphabetURL = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"

src/Simplex/Messaging/Encoding/String.hs

@@ -10,6 +10,7 @@ module Simplex.Messaging.Encoding.String
     strToJSON,
     strToJEncoding,
     strParseJSON,
+    base64urlP,
     strEncodeList,
     strListP,
   )
@@ -22,8 +23,10 @@ import qualified Data.Aeson.Encoding as JE
 import qualified Data.Aeson.Types as JT
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
+import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.Char (isAlphaNum)
 import Data.Int (Int64)
 import qualified Data.List.NonEmpty as L
 import Data.Set (Set)
@@ -35,7 +38,6 @@ import Data.Time.Clock.System (SystemTime (..))
 import Data.Time.Format.ISO8601
 import Data.Word (Word16, Word32)
 import Simplex.Messaging.Encoding
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Util ((<$?>))
 
@@ -52,16 +54,19 @@ class StrEncoding a where
   strDecode :: ByteString -> Either String a
   strDecode = parseAll strP
   strP :: Parser a
-  strP = strDecode <$?> U.base64urlP
+  strP = strDecode <$?> base64urlP
 
 -- 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 = U.base64urlP
-  {-# INLINE strP #-}
+  strP = base64urlP
+
+base64urlP :: Parser ByteString
+base64urlP = do
+  str <- A.takeWhile1 (\c -> isAlphaNum c || c == '-' || c == '_')
+  pad <- A.takeWhile (== '=')
+  either fail pure $ U.decode (str <> pad)
 
 newtype Str = Str {unStr :: ByteString}
   deriving (Eq, Show)

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -11,12 +11,14 @@
 
 module Simplex.Messaging.Notifications.Protocol where
 
+import Control.Applicative ((<|>))
 import Data.Aeson (FromJSON (..), ToJSON (..), (.:), (.=))
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Encoding as JE
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.Functor (($>))
 import Data.Kind
 import Data.Maybe (isNothing)
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
@@ -406,10 +408,12 @@ instance Encoding DeviceToken where
 
 instance StrEncoding DeviceToken where
   strEncode (DeviceToken p t) = strEncode p <> " " <> t
-  strP = DeviceToken <$> strP <* A.space <*> hexStringP
+  strP = nullToken <|> hexToken
     where
+      nullToken = "apns_null test_ntf_token" $> DeviceToken PPApnsNull "test_ntf_token"
+      hexToken = DeviceToken <$> strP <* A.space <*> hexStringP
       hexStringP =
-        A.takeWhile (\c -> A.isDigit c || (c >= 'a' && c <= 'f')) >>= \s ->
+        A.takeWhile (`B.elem` "0123456789abcdef") >>= \s ->
           if even (B.length s) then pure s else fail "odd number of hex characters"
 
 instance ToJSON DeviceToken where

src/Simplex/Messaging/Notifications/Server.hs

@@ -108,7 +108,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
     logServerStats :: Int64 -> Int64 -> FilePath -> M ()
     logServerStats startAt logInterval statsFilePath = do
       initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
-      liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
+      logInfo $ "server stats log enabled: " <> T.pack statsFilePath
       liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
       NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, subCreated, subDeleted, ntfReceived, ntfDelivered, activeTokens, activeSubs} <- asks serverStats
       let interval = 1000000 * logInterval
@@ -116,16 +116,16 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
         withFile statsFilePath AppendMode $ \h -> liftIO $ do
           hSetBuffering h LineBuffering
           ts <- getCurrentTime
-          fromTime' <- atomically $ swapTVar fromTime ts
-          tknCreated' <- atomically $ swapTVar tknCreated 0
-          tknVerified' <- atomically $ swapTVar tknVerified 0
-          tknDeleted' <- atomically $ swapTVar tknDeleted 0
-          subCreated' <- atomically $ swapTVar subCreated 0
-          subDeleted' <- atomically $ swapTVar subDeleted 0
-          ntfReceived' <- atomically $ swapTVar ntfReceived 0
-          ntfDelivered' <- atomically $ swapTVar ntfDelivered 0
-          tkn <- atomically $ periodStatCounts activeTokens ts
-          sub <- atomically $ periodStatCounts activeSubs ts
+          fromTime' <- atomically' $ swapTVar fromTime ts
+          tknCreated' <- atomically' $ swapTVar tknCreated 0
+          tknVerified' <- atomically' $ swapTVar tknVerified 0
+          tknDeleted' <- atomically' $ swapTVar tknDeleted 0
+          subCreated' <- atomically' $ swapTVar subCreated 0
+          subDeleted' <- atomically' $ swapTVar subDeleted 0
+          ntfReceived' <- atomically' $ swapTVar ntfReceived 0
+          ntfDelivered' <- atomically' $ swapTVar ntfDelivered 0
+          tkn <- atomically' $ periodStatCounts activeTokens ts
+          sub <- atomically' $ periodStatCounts activeSubs ts
           hPutStrLn h $
             intercalate
               ","
@@ -151,7 +151,7 @@ resubscribe NtfSubscriber {newSubQ} = do
   logInfo "Preparing SMP resubscriptions..."
   subs <- readTVarIO =<< asks (subscriptions . store)
   subs' <- filterM (fmap ntfShouldSubscribe . readTVarIO . subStatus) $ M.elems subs
-  atomically . writeTBQueue newSubQ $ map NtfSub subs'
+  atomically' . writeTBQueue newSubQ $ map NtfSub subs'
   logInfo $ "SMP resubscriptions queued (" <> tshow (length subs') <> " subscriptions)"
 
 ntfSubscriber :: NtfSubscriber -> M ()
@@ -160,14 +160,14 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
   where
     subscribe :: M ()
     subscribe = forever $ do
-      subs <- atomically (readTBQueue newSubQ)
+      subs <- atomically' (readTBQueue newSubQ)
       let ss = L.groupAllWith server subs
       batchSize <- asks $ subsBatchSize . config
       forM_ ss $ \serverSubs -> do
         let srv = server $ L.head serverSubs
             batches = toChunks batchSize $ L.toList serverSubs
         SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber srv
-        mapM_ (atomically . writeTQueue subscriberSubQ) batches
+        mapM_ (atomically' . writeTQueue subscriberSubQ) batches
 
     server :: NtfEntityRec 'Subscription -> SMPServer
     server (NtfSub sub) = ntfSubServer sub
@@ -186,14 +186,14 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
     runSMPSubscriber :: SMPSubscriber -> M ()
     runSMPSubscriber SMPSubscriber {newSubQ = subscriberSubQ} =
       forever $ do
-        subs <- atomically (peekTQueue subscriberSubQ)
+        subs <- atomically' (peekTQueue subscriberSubQ)
         let subs' = L.map (\(NtfSub sub) -> sub) subs
             srv = server $ L.head subs
         logSubStatus srv "subscribing" $ length subs
         mapM_ (\NtfSubData {smpQueue} -> updateSubStatus smpQueue NSPending) subs'
         rs <- liftIO $ subscribeQueues srv subs'
         (subs'', oks, errs) <- foldM process ([], 0, []) rs
-        atomically $ do
+        atomically' $ do
           void $ readTQueue subscriberSubQ
           mapM_ (writeTQueue subscriberSubQ . L.map NtfSub) $ L.nonEmpty subs''
         logSubStatus srv "retrying" $ length subs''
@@ -218,7 +218,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     receiveSMP :: M ()
     receiveSMP = forever $ do
-      ((_, srv, _), _, _, ntfId, msg) <- atomically $ readTBQueue msgQ
+      ((_, srv, _), _, _, ntfId, msg) <- atomically' $ readTBQueue msgQ
       let smpQueue = SMPQueueNtf srv ntfId
       case msg of
         SMP.NMSG nmsgNonce encNMsgMeta -> do
@@ -226,8 +226,8 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
           st <- asks store
           NtfPushServer {pushQ} <- asks pushServer
           stats <- asks serverStats
-          atomically $ updatePeriodStats (activeSubs stats) ntfId
-          atomically $
+          atomically' $ updatePeriodStats (activeSubs stats) ntfId
+          atomically' $
             findNtfSubscriptionToken st smpQueue
               >>= mapM_ (\tkn -> writeTBQueue pushQ (tkn, PNMessage PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}))
           incNtfStat ntfReceived
@@ -236,7 +236,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     receiveAgent =
       forever $
-        atomically (readTBQueue agentQ) >>= \case
+        atomically' (readTBQueue agentQ) >>= \case
           CAConnected _ -> pure ()
           CADisconnected srv subs -> do
             logSubStatus srv "disconnected" $ length subs
@@ -280,7 +280,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     updateSubStatus smpQueue status = do
       st <- asks store
-      atomically (findNtfSubscription st smpQueue) >>= mapM_ update
+      atomically' (findNtfSubscription st smpQueue) >>= mapM_ update
       where
         update NtfSubData {ntfSubId, subStatus} = do
           old <- atomically $ stateTVar subStatus (,status)
@@ -288,7 +288,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
 ntfPush :: NtfPushServer -> M ()
 ntfPush s@NtfPushServer {pushQ} = forever $ do
-  (tkn@NtfTknData {ntfTknId, token = DeviceToken pp _, tknStatus}, ntf) <- atomically (readTBQueue pushQ)
+  (tkn@NtfTknData {ntfTknId, token = DeviceToken pp _, tknStatus}, ntf) <- atomically' (readTBQueue pushQ)
   liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp)
   status <- readTVarIO tknStatus
   case ntf of
@@ -307,7 +307,7 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
       void $ deliverNotification pp tkn ntf
     PNMessage {} -> checkActiveTkn status $ do
       stats <- asks serverStats
-      atomically $ updatePeriodStats (activeTokens stats) ntfTknId
+      atomically' $ updatePeriodStats (activeTokens stats) ntfTknId
       void $ deliverNotification pp tkn ntf
       incNtfStat ntfDelivered
   where
@@ -343,7 +343,7 @@ runNtfClientTransport :: Transport c => THandleNTF c -> M ()
 runNtfClientTransport th@THandle {params} = do
   qSize <- asks $ clientQSize . config
   ts <- liftIO getSystemTime
-  c <- atomically $ newNtfServerClient qSize params ts
+  c <- atomically' $ newNtfServerClient qSize params ts
   s <- asks subscriber
   ps <- asks pushServer
   expCfg <- asks $ inactiveClientExpiration . config
@@ -373,7 +373,7 @@ receive th@THandle {params = THandleParams {thAuth}} NtfServerClient {rcvQ, sndQ
 
 send :: Transport c => THandleNTF c -> NtfServerClient -> IO ()
 send h@THandle {params} NtfServerClient {sndQ, sndActiveAt} = forever $ do
-  t <- atomically $ readTBQueue sndQ
+  t <- atomically' $ readTBQueue sndQ
   void . liftIO $ tPut h [Right (Nothing, encodeTransmission params t)]
   atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
 
@@ -387,7 +387,7 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
   st <- asks store
   case cmd of
     NtfCmd SToken c@(TNEW tkn@(NewNtfTkn _ k _)) -> do
-      r_ <- atomically $ getNtfTokenRegistration st tkn
+      r_ <- atomically' $ getNtfTokenRegistration st tkn
       pure $
         if verifyCmdAuthorization auth_ tAuth authorized k
           then case r_ of
@@ -397,26 +397,26 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
             _ -> VRVerified (NtfReqNew corrId (ANE SToken tkn))
           else VRFailed
     NtfCmd SToken c -> do
-      t_ <- atomically $ getNtfToken st entId
+      t_ <- atomically' $ getNtfToken st entId
       verifyToken t_ (`verifiedTknCmd` c)
     NtfCmd SSubscription c@(SNEW sub@(NewNtfSub tknId smpQueue _)) -> do
-      s_ <- atomically $ findNtfSubscription st smpQueue
+      s_ <- atomically' $ findNtfSubscription st smpQueue
       case s_ of
         Nothing -> do
-          t_ <- atomically $ getActiveNtfToken st tknId
+          t_ <- atomically' $ getActiveNtfToken st tknId
           verifyToken' t_ $ VRVerified (NtfReqNew corrId (ANE SSubscription sub))
         Just s@NtfSubData {tokenId = subTknId} ->
           if subTknId == tknId
             then do
-              t_ <- atomically $ getActiveNtfToken st subTknId
+              t_ <- atomically' $ getActiveNtfToken st subTknId
               verifyToken' t_ $ verifiedSubCmd s c
             else pure $ maybe False (dummyVerifyCmd auth_ authorized) tAuth `seq` VRFailed
     NtfCmd SSubscription PING -> pure $ VRVerified $ NtfReqPing corrId entId
     NtfCmd SSubscription c -> do
-      s_ <- atomically $ getNtfSubscription st entId
+      s_ <- atomically' $ getNtfSubscription st entId
       case s_ of
         Just s@NtfSubData {tokenId = subTknId} -> do
-          t_ <- atomically $ getActiveNtfToken st subTknId
+          t_ <- atomically' $ getActiveNtfToken st subTknId
           verifyToken' t_ $ verifiedSubCmd s c
         _ -> pure $ maybe False (dummyVerifyCmd auth_ authorized) tAuth `seq` VRFailed
   where
@@ -436,26 +436,26 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
 client :: NtfServerClient -> NtfSubscriber -> NtfPushServer -> M ()
 client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPushServer {pushQ, intervalNotifiers} =
   forever $
-    atomically (readTBQueue rcvQ)
+    atomically' (readTBQueue rcvQ)
       >>= processCommand
-      >>= atomically . writeTBQueue sndQ
+      >>= atomically' . writeTBQueue sndQ
   where
     processCommand :: NtfRequest -> M (Transmission NtfResponse)
     processCommand = \case
-      NtfReqNew corrId (ANE SToken newTkn@(NewNtfTkn _ _ dhPubKey)) -> do
+      NtfReqNew corrId (ANE SToken newTkn@(NewNtfTkn token _ dhPubKey)) -> do
         logDebug "TNEW - new token"
         st <- asks store
         ks@(srvDhPubKey, srvDhPrivKey) <- atomically . C.generateKeyPair =<< asks random
         let dhSecret = C.dh' dhPubKey srvDhPrivKey
         tknId <- getId
         regCode <- getRegCode
-        tkn <- atomically $ mkNtfTknData tknId newTkn ks dhSecret regCode
-        atomically $ addNtfToken st tknId tkn
+        tkn <- atomically' $ mkNtfTknData tknId newTkn ks dhSecret regCode
+        atomically' $ addNtfToken st tknId tkn
         atomically $ writeTBQueue pushQ (tkn, PNVerification regCode)
         withNtfLog (`logCreateToken` tkn)
-        incNtfStat tknCreated
+        incNtfStatT token tknCreated
         pure (corrId, "", NRTknId tknId srvDhPubKey)
-      NtfReqCmd SToken (NtfTkn tkn@NtfTknData {ntfTknId, tknStatus, tknRegCode, tknDhSecret, tknDhKeys = (srvDhPubKey, srvDhPrivKey), tknCronInterval}) (corrId, tknId, cmd) -> do
+      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
           TNEW (NewNtfTkn _ _ dhPubKey) -> do
@@ -472,9 +472,9 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
                 logDebug "TVFY - token verified"
                 st <- asks store
                 updateTknStatus tkn NTActive
-                tIds <- atomically $ removeInactiveTokenRegistrations st tkn
+                tIds <- atomically' $ removeInactiveTokenRegistrations st tkn
                 forM_ tIds cancelInvervalNotifications
-                incNtfStat tknVerified
+                incNtfStatT token tknVerified
                 pure NROk
             | otherwise -> do
                 logDebug "TVFY - incorrect code or token status"
@@ -486,25 +486,25 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
             logDebug "TRPL - replace token"
             st <- asks store
             regCode <- getRegCode
-            atomically $ do
+            atomically' $ do
               removeTokenRegistration st tkn
               writeTVar tknStatus NTRegistered
               let tkn' = tkn {token = token', tknRegCode = regCode}
               addNtfToken st tknId tkn'
               writeTBQueue pushQ (tkn', PNVerification regCode)
             withNtfLog $ \s -> logUpdateToken s tknId token' regCode
-            incNtfStat tknDeleted
-            incNtfStat tknCreated
+            incNtfStatT token tknDeleted
+            incNtfStatT token tknCreated
             pure NROk
           TDEL -> do
             logDebug "TDEL"
             st <- asks store
-            qs <- atomically $ deleteNtfToken st tknId
+            qs <- atomically' $ deleteNtfToken st tknId
             forM_ qs $ \SMPQueueNtf {smpServer, notifierId} ->
-              atomically $ removeSubscription ca smpServer (SPNotifier, notifierId)
+              atomically' $ removeSubscription ca smpServer (SPNotifier, notifierId)
             cancelInvervalNotifications tknId
             withNtfLog (`logDeleteToken` tknId)
-            incNtfStat tknDeleted
+            incNtfStatT token tknDeleted
             pure NROk
           TCRN 0 -> do
             logDebug "TCRN 0"
@@ -538,10 +538,10 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
         logDebug "SNEW - new subscription"
         st <- asks store
         subId <- getId
-        sub <- atomically $ mkNtfSubData subId newSub
+        sub <- atomically' $ mkNtfSubData subId newSub
         resp <-
-          atomically (addNtfSubscription st subId sub) >>= \case
-            Just _ -> atomically (writeTBQueue newSubQ [NtfSub sub]) $> NRSubId subId
+          atomically' (addNtfSubscription st subId sub) >>= \case
+            Just _ -> atomically' (writeTBQueue newSubQ [NtfSub sub]) $> NRSubId subId
             _ -> pure $ NRErr AUTH
         withNtfLog (`logCreateSubscription` sub)
         incNtfStat subCreated
@@ -562,8 +562,8 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
           SDEL -> do
             logDebug "SDEL"
             st <- asks store
-            atomically $ deleteNtfSubscription st subId
-            atomically $ removeSubscription ca smpServer (SPNotifier, notifierId)
+            atomically' $ deleteNtfSubscription st subId
+            atomically' $ removeSubscription ca smpServer (SPNotifier, notifierId)
             withNtfLog (`logDeleteSubscription` subId)
             incNtfStat subDeleted
             pure NROk
@@ -583,6 +583,10 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
 withNtfLog :: (StoreLog 'WriteMode -> IO a) -> M ()
 withNtfLog action = liftIO . mapM_ action =<< asks storeLog
 
+incNtfStatT :: DeviceToken -> (NtfServerStats -> TVar Int) -> M ()
+incNtfStatT (DeviceToken PPApnsNull _) _ = pure ()
+incNtfStatT _ statSel = incNtfStat statSel
+
 incNtfStat :: (NtfServerStats -> TVar Int) -> M ()
 incNtfStat statSel = do
   stats <- asks serverStats
@@ -591,7 +595,7 @@ incNtfStat statSel = do
 saveServerStats :: M ()
 saveServerStats =
   asks (serverStatsBackupFile . config)
-    >>= mapM_ (\f -> asks serverStats >>= atomically . getNtfServerStatsData >>= liftIO . saveStats f)
+    >>= mapM_ (\f -> asks serverStats >>= atomically' . getNtfServerStatsData >>= liftIO . saveStats f)
   where
     saveStats f stats = do
       logInfo $ "saving server stats to file " <> T.pack f
@@ -606,7 +610,7 @@ restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStat
       liftIO (strDecode <$> B.readFile f) >>= \case
         Right d -> do
           s <- asks serverStats
-          atomically $ setNtfServerStats s d
+          atomically' $ setNtfServerStats s d
           renameFile f $ f <> ".bak"
           logInfo "server stats restored"
         Left e -> do

src/Simplex/Messaging/Notifications/Server/Push/APNS.hs

@@ -27,9 +27,8 @@ import Data.Aeson (ToJSON, (.=))
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Encoding as JE
 import qualified Data.Aeson.TH as JQ
-import Data.Base64.Types (extractBase64)
 import Data.Bifunctor (first)
-import qualified Data.ByteString.Base64.URL as UP
+import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Builder (lazyByteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Lazy.Char8 as LB
@@ -47,7 +46,6 @@ import Network.HTTP2.Client (Request)
 import qualified Network.HTTP2.Client as H
 import Network.Socket (HostName, ServiceName)
 import qualified Simplex.Messaging.Crypto as C
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS.Internal
@@ -56,7 +54,7 @@ import Simplex.Messaging.Parsers (defaultJSON)
 import Simplex.Messaging.Protocol (EncNMsgMeta)
 import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..))
 import Simplex.Messaging.Transport.HTTP2.Client
-import Simplex.Messaging.Util (safeDecodeUtf8)
+import Simplex.Messaging.Util (safeDecodeUtf8, tshow)
 import System.Environment (getEnv)
 import UnliftIO.STM
 
@@ -93,8 +91,8 @@ signedJWTToken pk (JWTToken hdr claims) = do
   pure $ hc <> "." <> serialize sig
   where
     jwtEncode :: ToJSON a => a -> ByteString
-    jwtEncode = extractBase64 . UP.encodeBase64Unpadded' . LB.toStrict . J.encode
-    serialize sig = extractBase64 . UP.encodeBase64Unpadded' $ encodeASN1' DER [Start Sequence, IntVal (EC.sign_r sig), IntVal (EC.sign_s sig), End Sequence]
+    jwtEncode = U.encodeUnpadded . LB.toStrict . J.encode
+    serialize sig = U.encodeUnpadded $ encodeASN1' DER [Start Sequence, IntVal (EC.sign_r sig), IntVal (EC.sign_s sig), End Sequence]
 
 readECPrivateKey :: FilePath -> IO EC.PrivateKey
 readECPrivateKey f = do
@@ -260,11 +258,11 @@ mkApnsJWTToken appTeamId jwtHeader privateKey = do
 connectHTTPS2 :: HostName -> APNSPushClientConfig -> TVar (Maybe HTTP2Client) -> IO (Either HTTP2ClientError HTTP2Client)
 connectHTTPS2 apnsHost APNSPushClientConfig {apnsPort, http2cfg, caStoreFile} https2Client = do
   caStore_ <- XS.readCertificateStore caStoreFile
-  when (isNothing caStore_) $ putStrLn $ "Error loading CertificateStore from " <> caStoreFile
+  when (isNothing caStore_) $ logError $ "Error loading CertificateStore from " <> T.pack caStoreFile
   r <- getHTTP2Client apnsHost apnsPort caStore_ http2cfg disconnected
   case r of
     Right client -> atomically . writeTVar https2Client $ Just client
-    Left e -> putStrLn $ "Error connecting to APNS: " <> show e
+    Left e -> logError $ "Error connecting to APNS: " <> tshow e
   pure r
   where
     disconnected = atomically $ writeTVar https2Client Nothing

src/Simplex/Messaging/Notifications/Server/StoreLog.hs

@@ -24,10 +24,12 @@ module Simplex.Messaging.Notifications.Server.StoreLog
 where
 
 import Control.Concurrent.STM
+import Control.Logger.Simple
 import Control.Monad
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
+import qualified Data.Text as T
 import Data.Word (Word16)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -35,7 +37,7 @@ import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Store
 import Simplex.Messaging.Protocol (NtfPrivateAuthKey)
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (whenM)
+import Simplex.Messaging.Util (safeDecodeUtf8, whenM, atomically')
 import System.Directory (doesFileExist, renameFile)
 import System.IO
 
@@ -193,8 +195,8 @@ readNtfStore :: FilePath -> NtfStore -> IO ()
 readNtfStore f st = mapM_ (addNtfLogRecord . LB.toStrict) . LB.lines =<< LB.readFile f
   where
     addNtfLogRecord s = case strDecode s of
-      Left e -> B.putStrLn $ "Log parsing error (" <> B.pack e <> "): " <> B.take 100 s
-      Right lr -> atomically $ case lr of
+      Left e -> logError $ "Log parsing error (" <> T.pack e <> "): " <> safeDecodeUtf8 (B.take 100 s)
+      Right lr -> atomically' $ case lr of
         CreateToken r@NtfTknRec {ntfTknId} -> do
           tkn <- mkTknData r
           addNtfToken st ntfTknId tkn

src/Simplex/Messaging/Parsers.hs

@@ -10,9 +10,10 @@ import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
+import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Char (toLower)
+import Data.Char (isAlphaNum, toLower)
 import Data.String
 import Data.Text (Text)
 import qualified Data.Text as T
@@ -23,8 +24,23 @@ import Database.SQLite.Simple (ResultError (..), SQLData (..))
 import Database.SQLite.Simple.FromField (FieldParser, returnError)
 import Database.SQLite.Simple.Internal (Field (..))
 import Database.SQLite.Simple.Ok (Ok (Ok))
+import Simplex.Messaging.Util ((<$?>))
 import Text.Read (readMaybe)
 
+base64P :: Parser ByteString
+base64P = decode <$?> paddedBase64 rawBase64P
+
+paddedBase64 :: Parser ByteString -> Parser ByteString
+paddedBase64 raw = (<>) <$> raw <*> pad
+  where
+    pad = A.takeWhile (== '=')
+
+rawBase64P :: Parser ByteString
+rawBase64P = A.takeWhile1 (\c -> isAlphaNum c || c == '+' || c == '/')
+
+-- rawBase64UriP :: Parser ByteString
+-- rawBase64UriP = A.takeWhile1 (\c -> isAlphaNum c || c == '-' || c == '_')
+
 tsISO8601P :: Parser UTCTime
 tsISO8601P = maybe (fail "timestamp") pure . parseISO8601 . B.unpack =<< A.takeTill wordEnd
 

src/Simplex/Messaging/Protocol.hs

@@ -174,6 +174,7 @@ import qualified Data.Aeson.TH as J
 import Data.Attoparsec.ByteString.Char8 (Parser, (<?>))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
+import qualified Data.ByteString.Base64 as B64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Char (isPrint, isSpace)
@@ -191,7 +192,6 @@ import GHC.TypeLits (ErrorMessage (..), TypeError, type (+))
 import Network.Socket (ServiceName)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
-import qualified Simplex.Messaging.Encoding.Base64 as B64
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.ServiceScheme

src/Simplex/Messaging/Server.hs

@@ -45,6 +45,7 @@ import Control.Monad.IO.Unlift
 import Control.Monad.Reader
 import Crypto.Random
 import Data.Bifunctor (first)
+import Data.ByteString.Base64 (encode)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
@@ -68,7 +69,6 @@ import Network.Socket (ServiceName, Socket, socketToHandle)
 import Simplex.Messaging.Agent.Lock
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding (Encoding (smpEncode))
-import Simplex.Messaging.Encoding.Base64 (encode)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Control
@@ -155,7 +155,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
     serverThread s label subQ subs clientSubs unsub = do
       labelMyThread label
       forever $
-        atomically updateSubscribers
+        atomically' updateSubscribers
           $>>= endPreviousSubscriptions
           >>= liftIO . mapM_ unsub
       where
@@ -176,7 +176,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
             labelMyThread $ label <> ".endPreviousSubscriptions"
             atomically $ writeTBQueue (sndQ c) [(CorrId "", qId, END)]
             atomically $ modifyTVar' (endThreads c) $ IM.delete tId
-          mkWeakThreadId t >>= atomically . modifyTVar' (endThreads c) . IM.insert tId
+          mkWeakThreadId t >>= atomically' . modifyTVar' (endThreads c) . IM.insert tId
           atomically $ TM.lookupDelete qId (clientSubs c)
 
     expireMessagesThread_ :: ServerConfig -> [M ()]
@@ -195,8 +195,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
         old <- liftIO $ expireBeforeEpoch expCfg
         rIds <- M.keysSet <$> readTVarIO ms
         forM_ rIds $ \rId -> do
-          q <- atomically (getMsgQueue ms rId quota)
-          deleted <- atomically $ deleteExpiredMsgs q old
+          q <- atomically' (getMsgQueue ms rId quota)
+          deleted <- atomically' $ deleteExpiredMsgs q old
           atomically $ modifyTVar' (msgExpired stats) (+ deleted)
 
     serverStatsThread_ :: ServerConfig -> [M ()]
@@ -216,19 +216,19 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
         withFile statsFilePath AppendMode $ \h -> liftIO $ do
           hSetBuffering h LineBuffering
           ts <- getCurrentTime
-          fromTime' <- atomically $ swapTVar fromTime ts
-          qCreated' <- atomically $ swapTVar qCreated 0
-          qSecured' <- atomically $ swapTVar qSecured 0
-          qDeletedAll' <- atomically $ swapTVar qDeletedAll 0
-          qDeletedNew' <- atomically $ swapTVar qDeletedNew 0
-          qDeletedSecured' <- atomically $ swapTVar qDeletedSecured 0
-          msgSent' <- atomically $ swapTVar msgSent 0
-          msgRecv' <- atomically $ swapTVar msgRecv 0
-          msgExpired' <- atomically $ swapTVar msgExpired 0
-          ps <- atomically $ periodStatCounts activeQueues ts
-          msgSentNtf' <- atomically $ swapTVar msgSentNtf 0
-          msgRecvNtf' <- atomically $ swapTVar msgRecvNtf 0
-          psNtf <- atomically $ periodStatCounts activeQueuesNtf ts
+          fromTime' <- atomically' $ swapTVar fromTime ts
+          qCreated' <- atomically' $ swapTVar qCreated 0
+          qSecured' <- atomically' $ swapTVar qSecured 0
+          qDeletedAll' <- atomically' $ swapTVar qDeletedAll 0
+          qDeletedNew' <- atomically' $ swapTVar qDeletedNew 0
+          qDeletedSecured' <- atomically' $ swapTVar qDeletedSecured 0
+          msgSent' <- atomically' $ swapTVar msgSent 0
+          msgRecv' <- atomically' $ swapTVar msgRecv 0
+          msgExpired' <- atomically' $ swapTVar msgExpired 0
+          ps <- atomically' $ periodStatCounts activeQueues ts
+          msgSentNtf' <- atomically' $ swapTVar msgSentNtf 0
+          msgRecvNtf' <- atomically' $ swapTVar msgRecvNtf 0
+          psNtf <- atomically' $ periodStatCounts activeQueuesNtf ts
           qCount' <- readTVarIO qCount
           msgCount' <- readTVarIO msgCount
           hPutStrLn h $
@@ -354,7 +354,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
 #endif
               CPSockets -> withAdminRole $ do
                 (accepted', closed', active') <- unliftIO u $ asks sockets
-                (accepted, closed, active) <- atomically $ (,,) <$> readTVar accepted' <*> readTVar closed' <*> readTVar active'
+                (accepted, closed, active) <- atomically' $ (,,) <$> readTVar accepted' <*> readTVar closed' <*> readTVar active'
                 hPutStrLn h "Sockets: "
                 hPutStrLn h $ "accepted: " <> show accepted
                 hPutStrLn h $ "closed: " <> show closed
@@ -377,10 +377,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
               CPDelete queueId' -> withUserRole $ unliftIO u $ do
                 st <- asks queueStore
                 ms <- asks msgStore
-                queueId <- atomically (getQueue st SSender queueId') >>= \case
+                queueId <- atomically' (getQueue st SSender queueId') >>= \case
                   Left _ -> pure queueId' -- fallback to using as recipientId directly
                   Right QueueRec {recipientId} -> pure recipientId
-                r <- atomically $
+                r <- atomically' $
                   deleteQueue st queueId $>>= \q ->
                     Right . (q,) <$> delMsgQueueSize ms queueId
                 case r of
@@ -415,7 +415,7 @@ runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} =
   ts <- liftIO getSystemTime
   active <- asks clients
   nextClientId <- asks clientSeq
-  c <- atomically $ do
+  c <- atomically' $ do
     new@Client {clientId} <- newClient nextClientId q thVersion sessionId ts
     modifyTVar' active $ IM.insert clientId new
     pure new
@@ -427,20 +427,20 @@ runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} =
   where
     disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport th (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
     disconnectThread_ _ _ = []
-    noSubscriptions c = atomically $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)
+    noSubscriptions c = atomically' $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)
 
 clientDisconnected :: Client -> M ()
 clientDisconnected c@Client {clientId, subscriptions, connected, sessionId, endThreads} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " disc"
-  subs <- atomically $ do
+  subs <- atomically' $ do
     writeTVar connected False
     swapTVar subscriptions M.empty
   liftIO $ mapM_ cancelSub subs
   srvSubs <- asks $ subscribers . server
   atomically $ modifyTVar' srvSubs $ \cs ->
     M.foldrWithKey (\sub _ -> M.update deleteCurrentClient sub) cs subs
-  asks clients >>= atomically . (`modifyTVar'` IM.delete clientId)
-  tIds <- atomically $ swapTVar endThreads IM.empty
+  asks clients >>= atomically' . (`modifyTVar'` IM.delete clientId)
+  tIds <- atomically' $ swapTVar endThreads IM.empty
   liftIO $ mapM_ (mapM_ killThread <=< deRefWeak) tIds
   where
     deleteCurrentClient :: Client -> Maybe Client
@@ -476,13 +476,13 @@ receive th@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActi
             verified = \case
               VRVerified qr -> Right (qr, (corrId, queueId, cmd))
               VRFailed -> Left (corrId, queueId, ERR AUTH)
-    write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
+    write q = mapM_ (atomically' . writeTBQueue q) . L.nonEmpty
 
 send :: Transport c => THandleSMP c -> Client -> IO ()
 send h@THandle {params} Client {sndQ, sessionId, sndActiveAt} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
   forever $ do
-    ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
+    ts <- atomically' $ L.sortWith tOrder <$> readTBQueue sndQ
     -- TODO we can authorize responses as well
     void . liftIO . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
     atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
@@ -534,7 +534,7 @@ verifyTransmission auth_ tAuth authorized queueId cmd =
     get :: SParty p -> M (Either ErrorType QueueRec)
     get party = do
       st <- asks queueStore
-      atomically $ getQueue st party queueId
+      atomically' $ getQueue st party queueId
 
 verifyCmdAuthorization :: Maybe (THandleAuth, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> C.APublicAuthKey -> Bool
 verifyCmdAuthorization auth_ tAuth authorized key = maybe False (verify key) tAuth
@@ -585,9 +585,9 @@ client :: Client -> Server -> M ()
 client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Server {subscribedQ, ntfSubscribedQ, notifiers} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
   forever $
-    atomically (readTBQueue rcvQ)
+    atomically' (readTBQueue rcvQ)
       >>= mapM processCommand
-      >>= atomically . writeTBQueue sndQ
+      >>= atomically' . writeTBQueue sndQ
   where
     processCommand :: (Maybe QueueRec, Transmission Cmd) -> M (Transmission BrokerMsg)
     processCommand (qr_, (corrId, queueId, cmd)) = do
@@ -642,7 +642,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
               ids@(rId, _) <- getIds
               -- create QueueRec record with these ids and keys
               let qr = qRec ids
-              atomically (addQueue st qr) >>= \case
+              atomically' (addQueue st qr) >>= \case
                 Left DUPLICATE_ -> addQueueRetry (n - 1) qik qRec
                 Left e -> pure $ ERR e
                 Right _ -> do
@@ -657,7 +657,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
 
             logCreateById :: StoreLog 'WriteMode -> RecipientId -> IO ()
             logCreateById s rId =
-              atomically (getQueue st SRecipient rId) >>= \case
+              atomically' (getQueue st SRecipient rId) >>= \case
                 Right q -> logCreateQueue s q
                 _ -> pure ()
 
@@ -671,7 +671,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
           withLog $ \s -> logSecureQueue s queueId sKey
           stats <- asks serverStats
           atomically $ modifyTVar' (qSecured stats) (+ 1)
-          atomically $ (corrId,queueId,) . either ERR (const OK) <$> secureQueue st queueId sKey
+          atomically' $ (corrId,queueId,) . either ERR (const OK) <$> secureQueue st queueId sKey
 
         addQueueNotifier_ :: QueueStore -> NtfPublicAuthKey -> RcvNtfPublicDhKey -> M (Transmission BrokerMsg)
         addQueueNotifier_ st notifierKey dhKey = time "NKEY" $ do
@@ -684,7 +684,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
             addNotifierRetry n rcvPublicDhKey rcvNtfDhSecret = do
               notifierId <- randomId =<< asks (queueIdBytes . config)
               let ntfCreds = NtfCreds {notifierId, notifierKey, rcvNtfDhSecret}
-              atomically (addQueueNotifier st queueId ntfCreds) >>= \case
+              atomically' (addQueueNotifier st queueId ntfCreds) >>= \case
                 Left DUPLICATE_ -> addNotifierRetry (n - 1) rcvPublicDhKey rcvNtfDhSecret
                 Left e -> pure $ ERR e
                 Right _ -> do
@@ -694,12 +694,12 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
         deleteQueueNotifier_ :: QueueStore -> M (Transmission BrokerMsg)
         deleteQueueNotifier_ st = do
           withLog (`logDeleteNotifier` queueId)
-          okResp <$> atomically (deleteQueueNotifier st queueId)
+          okResp <$> atomically' (deleteQueueNotifier st queueId)
 
         suspendQueue_ :: QueueStore -> M (Transmission BrokerMsg)
         suspendQueue_ st = do
           withLog (`logSuspendQueue` queueId)
-          okResp <$> atomically (suspendQueue st queueId)
+          okResp <$> atomically' (suspendQueue st queueId)
 
         subscribeQueue :: QueueRec -> RecipientId -> M (Transmission BrokerMsg)
         subscribeQueue qr rId = do
@@ -712,10 +712,10 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                   -- cannot use SUB in the same connection where GET was used
                   pure (corrId, rId, ERR $ CMD PROHIBITED)
                 s ->
-                  atomically (tryTakeTMVar $ delivered s) >> deliver sub
+                  atomically' (tryTakeTMVar $ delivered s) >> deliver sub
           where
             newSub :: M (TVar Sub)
-            newSub = time "SUB newSub" . atomically $ do
+            newSub = time "SUB newSub" . atomically' $ do
               writeTQueue subscribedQ (rId, clnt)
               sub <- newTVar =<< newSubscription NoSub
               TM.insert rId sub subscriptions
@@ -723,7 +723,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
             deliver :: TVar Sub -> M (Transmission BrokerMsg)
             deliver sub = do
               q <- getStoreMsgQueue "SUB" rId
-              msg_ <- atomically $ tryPeekMsg q
+              msg_ <- atomically' $ tryPeekMsg q
               deliverMessage "SUB" qr rId sub q msg_
 
         getMessage :: QueueRec -> M (Transmission BrokerMsg)
@@ -734,7 +734,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
             Just sub ->
               readTVarIO sub >>= \case
                 s@Sub {subThread = ProhibitSub} ->
-                  atomically (tryTakeTMVar $ delivered s)
+                  atomically' (tryTakeTMVar $ delivered s)
                     >> getMessage_ s
                 -- cannot use GET in the same connection where there is an active subscription
                 _ -> pure (corrId, queueId, ERR $ CMD PROHIBITED)
@@ -748,7 +748,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
             getMessage_ :: Sub -> M (Transmission BrokerMsg)
             getMessage_ s = do
               q <- getStoreMsgQueue "GET" queueId
-              atomically $
+              atomically' $
                 tryPeekMsg q >>= \case
                   Just msg ->
                     let encMsg = encryptMsg qr msg
@@ -759,7 +759,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
         withQueue action = maybe (pure $ err AUTH) action qr_
 
         subscribeNotifications :: M (Transmission BrokerMsg)
-        subscribeNotifications = time "NSUB" . atomically $ do
+        subscribeNotifications = time "NSUB" . atomically' $ do
           unlessM (TM.member queueId ntfSubscriptions) $ do
             writeTQueue ntfSubscribedQ (queueId, clnt)
             TM.insert queueId () ntfSubscriptions
@@ -770,16 +770,16 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
           atomically (TM.lookup queueId subscriptions) >>= \case
             Nothing -> pure $ err NO_MSG
             Just sub ->
-              atomically (getDelivered sub) >>= \case
+              atomically' (getDelivered sub) >>= \case
                 Just s -> do
                   q <- getStoreMsgQueue "ACK" queueId
                   case s of
                     Sub {subThread = ProhibitSub} -> do
-                      deletedMsg_ <- atomically $ tryDelMsg q msgId
+                      deletedMsg_ <- atomically' $ tryDelMsg q msgId
                       mapM_ updateStats deletedMsg_
                       pure ok
                     _ -> do
-                      (deletedMsg_, msg_) <- atomically $ tryDelPeekMsg q msgId
+                      (deletedMsg_, msg_) <- atomically' $ tryDelPeekMsg q msgId
                       mapM_ updateStats deletedMsg_
                       deliverMessage "ACK" qr queueId sub q msg_
                 _ -> pure $ err NO_MSG
@@ -798,10 +798,10 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                 stats <- asks serverStats
                 atomically $ modifyTVar' (msgRecv stats) (+ 1)
                 atomically $ modifyTVar' (msgCount stats) (subtract 1)
-                atomically $ updatePeriodStats (activeQueues stats) queueId
+                atomically' $ updatePeriodStats (activeQueues stats) queueId
                 when (notification msgFlags) $ do
                   atomically $ modifyTVar' (msgRecvNtf stats) (+ 1)
-                  atomically $ updatePeriodStats (activeQueuesNtf stats) queueId
+                  atomically' $ updatePeriodStats (activeQueuesNtf stats) queueId
 
         sendMessage :: QueueRec -> MsgFlags -> MsgBody -> M (Transmission BrokerMsg)
         sendMessage qr msgFlags msgBody
@@ -815,18 +815,18 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                     msg_ <- time "SEND" $ do
                       q <- getStoreMsgQueue "SEND" $ recipientId qr
                       expireMessages q
-                      atomically . writeMsg q =<< mkMessage body
+                      atomically' . writeMsg q =<< mkMessage body
                     case msg_ of
                       Nothing -> pure $ err QUOTA
                       Just msg -> time "SEND ok" $ do
                         stats <- asks serverStats
                         when (notification msgFlags) $ do
-                          atomically . trySendNotification msg =<< asks random
+                          atomically' . trySendNotification msg =<< asks random
                           atomically $ modifyTVar' (msgSentNtf stats) (+ 1)
-                          atomically $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
+                          atomically' $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
                         atomically $ modifyTVar' (msgSent stats) (+ 1)
                         atomically $ modifyTVar' (msgCount stats) (+ 1)
-                        atomically $ updatePeriodStats (activeQueues stats) (recipientId qr)
+                        atomically' $ updatePeriodStats (activeQueues stats) (recipientId qr)
                         pure ok
           where
             mkMessage :: C.MaxLenBS MaxMessageLen -> M Message
@@ -840,7 +840,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
               msgExp <- asks $ messageExpiration . config
               old <- liftIO $ mapM expireBeforeEpoch msgExp
               stats <- asks serverStats
-              deleted <- atomically $ sum <$> mapM (deleteExpiredMsgs q) old
+              deleted <- atomically' $ sum <$> mapM (deleteExpiredMsgs q) old
               atomically $ modifyTVar' (msgExpired stats) (+ deleted)
 
             trySendNotification :: Message -> TVar ChaChaDRG -> STM ()
@@ -870,22 +870,22 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
               case msg_ of
                 Just msg ->
                   let encMsg = encryptMsg qr msg
-                   in atomically (setDelivered s msg) $> (corrId, rId, MSG encMsg)
+                   in atomically' (setDelivered s msg) $> (corrId, rId, MSG encMsg)
                 _ -> forkSub $> ok
             _ -> pure ok
           where
             forkSub :: M ()
             forkSub = do
-              atomically . modifyTVar' sub $ \s -> s {subThread = SubPending}
+              atomically' . modifyTVar' sub $ \s -> s {subThread = SubPending}
               t <- mkWeakThreadId =<< forkIO subscriber
-              atomically . modifyTVar' sub $ \case
+              atomically' . modifyTVar' sub $ \case
                 s@Sub {subThread = SubPending} -> s {subThread = SubThread t}
                 s -> s
               where
                 subscriber = do
                   labelMyThread $ B.unpack ("client $" <> encode sessionId) <> " subscriber/" <> T.unpack name
-                  msg <- atomically $ peekMsg q
-                  time "subscriber" . atomically $ do
+                  msg <- atomically' $ peekMsg q
+                  time "subscriber" . atomically' $ do
                     let encMsg = encryptMsg qr msg
                     writeTBQueue sndQ [(CorrId "", rId, MSG encMsg)]
                     s <- readTVar sub
@@ -912,13 +912,13 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
         getStoreMsgQueue name rId = time (name <> " getMsgQueue") $ do
           ms <- asks msgStore
           quota <- asks $ msgQueueQuota . config
-          atomically $ getMsgQueue ms rId quota
+          atomically' $ getMsgQueue ms rId quota
 
         delQueueAndMsgs :: QueueStore -> M (Transmission BrokerMsg)
         delQueueAndMsgs st = do
           withLog (`logDeleteQueue` queueId)
           ms <- asks msgStore
-          atomically (deleteQueue st queueId $>>= \q -> delMsgQueue ms queueId $> Right q) >>= \case
+          atomically' (deleteQueue st queueId $>>= \q -> delMsgQueue ms queueId $> Right q) >>= \case
             Right q -> updateDeletedStats q $> ok
             Left e -> pure $ err e
 
@@ -971,7 +971,7 @@ saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessag
       where
         getMessages = if keepMsgs then snapshotMsgQueue else flushMsgQueue
         saveQueueMsgs ms h rId =
-          atomically (getMessages ms rId)
+          atomically' (getMessages ms rId)
             >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)
 
 restoreServerMessages :: M Int
@@ -999,7 +999,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= \case
           where
             s = LB.toStrict s'
             addToMsgQueue rId msg = do
-              (isExpired, logFull) <- atomically $ do
+              (isExpired, logFull) <- atomically' $ do
                 q <- getMsgQueue ms rId quota
                 case msg of
                   Message {msgTs}
@@ -1014,7 +1014,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= \case
 saveServerStats :: M ()
 saveServerStats =
   asks (serverStatsBackupFile . config)
-    >>= mapM_ (\f -> asks serverStats >>= atomically . getServerStatsData >>= liftIO . saveStats f)
+    >>= mapM_ (\f -> asks serverStats >>= atomically' . getServerStatsData >>= liftIO . saveStats f)
   where
     saveStats f stats = do
       logInfo $ "saving server stats to file " <> T.pack f
@@ -1031,7 +1031,7 @@ restoreServerStats expiredWhileRestoring = asks (serverStatsBackupFile . config)
           s <- asks serverStats
           _qCount <- fmap M.size . readTVarIO . queues =<< asks queueStore
           _msgCount <- foldM (\(!n) q -> (n +) <$> readTVarIO (size q)) 0 =<< readTVarIO =<< asks msgStore
-          atomically $ setServerStats s d {_qCount, _msgCount, _msgExpired = _msgExpired d + expiredWhileRestoring}
+          atomically' $ setServerStats s d {_qCount, _msgCount, _msgExpired = _msgExpired d + expiredWhileRestoring}
           renameFile f $ f <> ".bak"
           logInfo "server stats restored"
           when (_qCount /= statsQCount) $ logWarn $ "Queue count differs: stats: " <> tshow statsQCount <> ", store: " <> tshow _qCount

src/Simplex/Messaging/Transport.hs

@@ -54,6 +54,7 @@ module Simplex.Messaging.Transport
     -- * TLS Transport
     TLS (..),
     SessionId,
+    ALPN,
     connectTLS,
     closeTLS,
     supportedParameters,
@@ -228,10 +229,13 @@ data TLS = TLS
     tlsPeer :: TransportPeer,
     tlsUniq :: ByteString,
     tlsBuffer :: TBuffer,
+    tlsALPN :: Maybe ALPN,
     tlsServerCerts :: X.CertificateChain,
     tlsTransportConfig :: TransportConfig
   }
 
+type ALPN = ByteString
+
 connectTLS :: T.TLSParams p => Maybe HostName -> TransportConfig -> p -> Socket -> IO T.Context
 connectTLS host_ TransportConfig {logTLSErrors} params sock =
   E.bracketOnError (T.contextNew sock params) closeTLS $ \ctx ->
@@ -246,7 +250,8 @@ getTLS tlsPeer cfg tlsServerCerts cxt = withTlsUnique tlsPeer cxt newTLS
   where
     newTLS tlsUniq = do
       tlsBuffer <- atomically newTBuffer
-      pure TLS {tlsContext = cxt, tlsTransportConfig = cfg, tlsServerCerts, tlsPeer, tlsUniq, tlsBuffer}
+      tlsALPN <- T.getNegotiatedProtocol cxt
+      pure TLS {tlsContext = cxt, tlsALPN, tlsTransportConfig = cfg, tlsServerCerts, tlsPeer, tlsUniq, tlsBuffer}
 
 withTlsUnique :: TransportPeer -> T.Context -> (ByteString -> IO c) -> IO c
 withTlsUnique peer cxt f =

src/Simplex/Messaging/Transport/Buffer.hs

@@ -11,6 +11,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import GHC.IO.Exception (IOErrorType (..), IOException (..), ioException)
 import System.Timeout (timeout)
+import Simplex.Messaging.Util (atomically')
 
 data TBuffer = TBuffer
   { buffer :: TVar ByteString,
@@ -26,8 +27,8 @@ newTBuffer = do
 withBufferLock :: TBuffer -> IO a -> IO a
 withBufferLock TBuffer {getLock} =
   E.bracket_
-    (atomically $ takeTMVar getLock)
-    (atomically $ putTMVar getLock ())
+    (atomically' $ takeTMVar getLock)
+    (atomically' $ putTMVar getLock ())
 
 -- | Attempt to read some bytes, appending it to the existing buffer
 peekBuffered :: TBuffer -> Int -> IO ByteString -> IO (ByteString, Maybe ByteString)

src/Simplex/Messaging/Transport/Client.hs

@@ -17,6 +17,7 @@ module Simplex.Messaging.Transport.Client
     TransportHost (..),
     TransportHosts (..),
     TransportHosts_ (..),
+    validateCertificateChain
   )
 where
 
@@ -49,7 +50,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow)
+import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow, atomically')
 import System.IO.Error
 import Text.Read (readMaybe)
 import UnliftIO.Exception (IOException)
@@ -113,12 +114,13 @@ data TransportClientConfig = TransportClientConfig
   { socksProxy :: Maybe SocksProxy,
     tcpKeepAlive :: Maybe KeepAliveOpts,
     logTLSErrors :: Bool,
-    clientCredentials :: Maybe (X.CertificateChain, T.PrivKey)
+    clientCredentials :: Maybe (X.CertificateChain, T.PrivKey),
+    alpn :: Maybe [ALPN]
   }
   deriving (Eq, Show)
 
 defaultTransportClientConfig :: TransportClientConfig
-defaultTransportClientConfig = TransportClientConfig Nothing (Just defaultKeepAliveOpts) True Nothing
+defaultTransportClientConfig = TransportClientConfig Nothing (Just defaultKeepAliveOpts) True Nothing Nothing
 
 clientTransportConfig :: TransportClientConfig -> TransportConfig
 clientTransportConfig TransportClientConfig {logTLSErrors} =
@@ -129,10 +131,10 @@ runTransportClient :: Transport c => TransportClientConfig -> Maybe ByteString -
 runTransportClient = runTLSTransportClient supportedParameters Nothing
 
 runTLSTransportClient :: Transport c => T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c -> IO a) -> IO a
-runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy, tcpKeepAlive, clientCredentials} proxyUsername host port keyHash client = do
+runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy, tcpKeepAlive, clientCredentials, alpn} proxyUsername host port keyHash client = do
   serverCert <- newEmptyTMVarIO
   let hostName = B.unpack $ strEncode host
-      clientParams = mkTLSClientParams tlsParams caStore_ hostName port keyHash clientCredentials serverCert
+      clientParams = mkTLSClientParams tlsParams caStore_ hostName port keyHash clientCredentials alpn serverCert
       connectTCP = case socksProxy of
         Just proxy -> connectSocksClient proxy proxyUsername $ hostAddr host
         _ -> connectTCPClient hostName
@@ -141,7 +143,7 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
     mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
     let tCfg = clientTransportConfig cfg
     connectTLS (Just hostName) tCfg clientParams sock >>= \tls -> do
-      chain <- atomically (tryTakeTMVar serverCert) >>= \case
+      chain <- atomically' (tryTakeTMVar serverCert) >>= \case
         Nothing -> do
           logError "onServerCertificate didn't fire or failed to get cert chain"
           closeTLS tls >> error "onServerCertificate failed"
@@ -215,14 +217,15 @@ instance ToJSON SocksProxy where
 instance FromJSON SocksProxy where
   parseJSON = strParseJSON "SocksProxy"
 
-mkTLSClientParams :: T.Supported -> Maybe XS.CertificateStore -> HostName -> ServiceName -> Maybe C.KeyHash -> Maybe (X.CertificateChain, T.PrivKey) -> TMVar X.CertificateChain -> T.ClientParams
-mkTLSClientParams supported caStore_ host port cafp_ clientCreds_ serverCerts =
+mkTLSClientParams :: T.Supported -> Maybe XS.CertificateStore -> HostName -> ServiceName -> Maybe C.KeyHash -> Maybe (X.CertificateChain, T.PrivKey) -> Maybe [ALPN] -> TMVar X.CertificateChain -> T.ClientParams
+mkTLSClientParams supported caStore_ host port cafp_ clientCreds_ alpn_ serverCerts =
   (T.defaultParamsClient host p)
     { T.clientShared = def {T.sharedCAStore = fromMaybe (T.sharedCAStore def) caStore_},
       T.clientHooks =
         def
           { T.onServerCertificate = onServerCert,
-            T.onCertificateRequest = maybe def (const . pure . Just) clientCreds_
+            T.onCertificateRequest = maybe def (const . pure . Just) clientCreds_,
+            T.onSuggestALPN = pure alpn_
           },
       T.clientSupported = supported
     }
@@ -231,13 +234,13 @@ mkTLSClientParams supported caStore_ host port cafp_ clientCreds_ serverCerts =
     onServerCert _ _ _ c = do
       errs <- maybe def (\ca -> validateCertificateChain ca host p c) cafp_
       when (null errs) $
-        atomically (putTMVar serverCerts c)
+        atomically' (putTMVar serverCerts c)
       pure errs
 
 validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]
 validateCertificateChain _ _ _ (X.CertificateChain []) = pure [XV.EmptyChain]
 validateCertificateChain _ _ _ (X.CertificateChain [_]) = pure [XV.EmptyChain]
-validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_, caCert]) =
+validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain [_, caCert]) =
   if Fingerprint kh == XV.getFingerprint caCert X.HashSHA256
     then x509validate
     else pure [XV.UnknownCA]
@@ -247,7 +250,7 @@ validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_,
       where
         hooks = XV.defaultHooks
         checks = XV.defaultChecks {XV.checkFQHN = False}
-        certStore = XS.makeCertificateStore sc
+        certStore = XS.makeCertificateStore [caCert]
         cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)
         serviceID = (host, port)
 validateCertificateChain _ _ _ _ = pure [XV.AuthorityTooDeep]

src/Simplex/Messaging/Transport/Credentials.hs

@@ -23,6 +23,7 @@ import Data.X509.Validation (Fingerprint (..), getFingerprint)
 import qualified Network.TLS as TLS
 import qualified Simplex.Messaging.Crypto as C
 import qualified Time.System as Hourglass
+import Simplex.Messaging.Util (atomically')
 
 -- | Generate a certificate chain to be used with TLS fingerprint-pinning
 --

src/Simplex/Messaging/Transport/HTTP2.hs

@@ -16,15 +16,15 @@ import qualified Network.HTTP2.Server as HS
 import Network.Socket (SockAddr (..))
 import qualified Network.TLS as T
 import qualified Network.TLS.Extra as TE
-import Simplex.Messaging.Transport (SessionId, TLS (tlsUniq), Transport (cGet, cPut))
+import Simplex.Messaging.Transport (TLS, Transport (cGet, cPut))
 import Simplex.Messaging.Transport.Buffer
 import qualified System.TimeManager as TI
 
 defaultHTTP2BufferSize :: BufferSize
 defaultHTTP2BufferSize = 32768
 
-withHTTP2 :: BufferSize -> (Config -> SessionId -> IO a) -> TLS -> IO a
-withHTTP2 sz run c = E.bracket (allocHTTP2Config c sz) freeSimpleConfig (`run` tlsUniq c)
+withHTTP2 :: BufferSize -> (Config -> IO a) -> IO () -> TLS -> IO a
+withHTTP2 sz run fin c = E.bracket (allocHTTP2Config c sz) (\cfg -> freeSimpleConfig cfg `E.finally` fin) run
 
 allocHTTP2Config :: TLS -> BufferSize -> IO Config
 allocHTTP2Config c sz = do

src/Simplex/Messaging/Transport/HTTP2/Client.hs

@@ -23,15 +23,20 @@ import qualified Network.TLS as T
 import Numeric.Natural (Natural)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Transport (SessionId, TLS)
+import Simplex.Messaging.Transport (ALPN, SessionId, TLS (tlsALPN), getServerCerts, getServerVerifyKey, tlsUniq)
 import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost (..), runTLSTransportClient)
 import Simplex.Messaging.Transport.HTTP2
 import UnliftIO.STM
 import UnliftIO.Timeout
+import qualified Data.X509 as X
+import Simplex.Messaging.Util (atomically')
 
 data HTTP2Client = HTTP2Client
   { action :: Maybe (Async HTTP2Response),
     sessionId :: SessionId,
+    sessionALPN :: Maybe ALPN,
+    serverKey :: C.APublicVerifyKey,
+    serverCerts :: X.CertificateChain,
     sessionTs :: UTCTime,
     sendReq :: Request -> (Response -> IO HTTP2Response) -> IO HTTP2Response,
     client_ :: HClient
@@ -66,7 +71,7 @@ defaultHTTP2ClientConfig =
   HTTP2ClientConfig
     { qSize = 64,
       connTimeout = 10000000,
-      transportConfig = TransportClientConfig Nothing Nothing True Nothing,
+      transportConfig = TransportClientConfig Nothing Nothing True Nothing Nothing,
       bufferSize = defaultHTTP2BufferSize,
       bodyHeadSize = 16384,
       suportedTLSParams = http2TLSParams
@@ -86,9 +91,10 @@ getVerifiedHTTP2Client proxyUsername host port keyHash caStore config disconnect
 attachHTTP2Client :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> Int -> TLS -> IO (Either HTTP2ClientError HTTP2Client)
 attachHTTP2Client config host port disconnected bufferSize tls = getVerifiedHTTP2ClientWith config host port disconnected setup
   where
+    setup :: (TLS -> H.Client HTTP2Response) -> IO HTTP2Response
     setup = runHTTP2ClientWith bufferSize host ($ tls)
 
-getVerifiedHTTP2ClientWith :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> ((SessionId -> H.Client HTTP2Response) -> IO HTTP2Response) -> IO (Either HTTP2ClientError HTTP2Client)
+getVerifiedHTTP2ClientWith :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> ((TLS -> H.Client HTTP2Response) -> IO HTTP2Response) -> IO (Either HTTP2ClientError HTTP2Client)
 getVerifiedHTTP2ClientWith config host port disconnected setup =
   (atomically mkHTTPS2Client >>= runClient)
     `E.catch` \(e :: IOException) -> pure . Left $ HCIOError e
@@ -102,29 +108,39 @@ getVerifiedHTTP2ClientWith config host port disconnected setup =
     runClient :: HClient -> IO (Either HTTP2ClientError HTTP2Client)
     runClient c = do
       cVar <- newEmptyTMVarIO
-      action <- async $ setup (client c cVar) `E.finally` atomically (putTMVar cVar $ Left HCNetworkError)
-      c_ <- connTimeout config `timeout` atomically (takeTMVar cVar)
+      action <- async $ setup (client c cVar) `E.finally` atomically' (putTMVar cVar $ Left HCNetworkError)
+      c_ <- connTimeout config `timeout` atomically' (takeTMVar cVar)
       pure $ case c_ of
         Just (Right c') -> Right c' {action = Just action}
         Just (Left e) -> Left e
         Nothing -> Left HCNetworkError
 
-    client :: HClient -> TMVar (Either HTTP2ClientError HTTP2Client) -> SessionId -> H.Client HTTP2Response
-    client c cVar sessionId sendReq = do
+    client :: HClient -> TMVar (Either HTTP2ClientError HTTP2Client) -> TLS -> H.Client HTTP2Response
+    client c cVar tls sendReq = do
       sessionTs <- getCurrentTime
-      let c' = HTTP2Client {action = Nothing, client_ = c, sendReq, sessionId, sessionTs}
-      atomically $ do
+      let c' =
+            HTTP2Client
+              { action = Nothing,
+                client_ = c,
+                serverKey = either (error "assert: TLS has server chain and key") id $ getServerVerifyKey tls,
+                serverCerts = getServerCerts tls,
+                sendReq,
+                sessionTs,
+                sessionId = tlsUniq tls,
+                sessionALPN = tlsALPN tls
+              }
+      atomically' $ do
         writeTVar (connected c) True
         putTMVar cVar (Right c')
       process c' sendReq `E.finally` disconnected
 
     process :: HTTP2Client -> H.Client HTTP2Response
     process HTTP2Client {client_ = HClient {reqQ}} sendReq = forever $ do
-      (req, respVar) <- atomically $ readTBQueue reqQ
+      (req, respVar) <- atomically' $ readTBQueue reqQ
       sendReq req $ \r -> do
         respBody <- getHTTP2Body r (bodyHeadSize config)
         let resp = HTTP2Response {response = r, respBody}
-        atomically $ putTMVar respVar resp
+        atomically' $ putTMVar respVar resp
         pure resp
 
 -- | Disconnects client from the server and terminates client threads.
@@ -136,7 +152,7 @@ sendRequest HTTP2Client {client_ = HClient {config, reqQ}} req reqTimeout_ = do
   resp <- newEmptyTMVarIO
   atomically $ writeTBQueue reqQ (req, resp)
   let reqTimeout = http2RequestTimeout config reqTimeout_
-  maybe (Left HCResponseTimeout) Right <$> (reqTimeout `timeout` atomically (takeTMVar resp))
+  maybe (Left HCResponseTimeout) Right <$> (reqTimeout `timeout` atomically' (takeTMVar resp))
 
 -- | this function should not be used until HTTP2 is thread safe, use sendRequest
 sendRequestDirect :: HTTP2Client -> Request -> Maybe Int -> IO (Either HTTP2ClientError HTTP2Response)
@@ -154,13 +170,14 @@ sendRequestDirect HTTP2Client {client_ = HClient {config, disconnected}, sendReq
 http2RequestTimeout :: HTTP2ClientConfig -> Maybe Int -> Int
 http2RequestTimeout HTTP2ClientConfig {connTimeout} = maybe connTimeout (connTimeout +)
 
-runHTTP2Client :: forall a. T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> BufferSize -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (SessionId -> H.Client a) -> IO a
+runHTTP2Client :: forall a. T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> BufferSize -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (TLS -> H.Client a) -> IO a
 runHTTP2Client tlsParams caStore tcConfig bufferSize proxyUsername host port keyHash = runHTTP2ClientWith bufferSize host setup
   where
+    setup :: (TLS -> IO a) -> IO a
     setup = runTLSTransportClient tlsParams caStore tcConfig proxyUsername host port keyHash
 
-runHTTP2ClientWith :: forall a. BufferSize -> TransportHost -> ((TLS -> IO a) -> IO a) -> (SessionId -> H.Client a) -> IO a
-runHTTP2ClientWith bufferSize host setup client = setup $ withHTTP2 bufferSize run
+runHTTP2ClientWith :: forall a. BufferSize -> TransportHost -> ((TLS -> IO a) -> IO a) -> (TLS -> H.Client a) -> IO a
+runHTTP2ClientWith bufferSize host setup client = setup $ \tls -> withHTTP2 bufferSize (run tls) (pure ()) tls
   where
-    run :: H.Config -> SessionId -> IO a
-    run cfg sessId = H.run (ClientConfig "https" (strEncode host) 20) cfg $ client sessId
+    run :: TLS -> H.Config -> IO a
+    run tls cfg = H.run (ClientConfig "https" (strEncode host) 20) cfg $ client tls

src/Simplex/Messaging/Transport/HTTP2/Server.hs

@@ -13,14 +13,14 @@ import Network.Socket
 import qualified Network.TLS as T
 import Numeric.Natural (Natural)
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (SessionId, TLS, closeConnection)
+import Simplex.Messaging.Transport (ALPN, SessionId, TLS, closeConnection, tlsALPN, tlsUniq)
 import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Transport.Server (TransportServerConfig (..), loadSupportedTLSServerParams, runTransportServer)
-import Simplex.Messaging.Util (threadDelay')
+import Simplex.Messaging.Util (threadDelay', atomically')
 import UnliftIO (finally)
 import UnliftIO.Concurrent (forkIO, killThread)
 
-type HTTP2ServerFunc = SessionId -> Request -> (Response -> IO ()) -> IO ()
+type HTTP2ServerFunc = SessionId -> Maybe ALPN -> Request -> (Response -> IO ()) -> IO ()
 
 data HTTP2ServerConfig = HTTP2ServerConfig
   { qSize :: Natural,
@@ -37,6 +37,7 @@ data HTTP2ServerConfig = HTTP2ServerConfig
 
 data HTTP2Request = HTTP2Request
   { sessionId :: SessionId,
+    sessionALPN :: Maybe ALPN,
     request :: Request,
     reqBody :: HTTP2Body,
     sendResponse :: Response -> IO ()
@@ -54,32 +55,32 @@ getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, se
   started <- newEmptyTMVarIO
   reqQ <- newTBQueueIO qSize
   action <- async $
-    runHTTP2Server started http2Port bufferSize tlsServerParams transportConfig Nothing $ \sessionId r sendResponse -> do
+    runHTTP2Server started http2Port bufferSize tlsServerParams transportConfig Nothing (const $ pure ()) $ \sessionId sessionALPN r sendResponse -> do
       reqBody <- getHTTP2Body r bodyHeadSize
-      atomically $ writeTBQueue reqQ HTTP2Request {sessionId, request = r, reqBody, sendResponse}
-  void . atomically $ takeTMVar started
+      atomically $ writeTBQueue reqQ HTTP2Request {sessionId, sessionALPN, request = r, reqBody, sendResponse}
+  void . atomically' $ takeTMVar started
   pure HTTP2Server {action, reqQ}
 
 closeHTTP2Server :: HTTP2Server -> IO ()
 closeHTTP2Server = uninterruptibleCancel . action
 
-runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.ServerParams -> TransportServerConfig -> Maybe ExpirationConfig -> HTTP2ServerFunc -> IO ()
-runHTTP2Server started port bufferSize serverParams transportConfig expCfg_ = runHTTP2ServerWith_ expCfg_ bufferSize setup
+runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.ServerParams -> TransportServerConfig -> Maybe ExpirationConfig -> (SessionId -> IO ()) -> HTTP2ServerFunc -> IO ()
+runHTTP2Server started port bufferSize serverParams transportConfig expCfg_ clientFinished = runHTTP2ServerWith_ expCfg_ clientFinished bufferSize setup
   where
     setup = runTransportServer started port serverParams transportConfig
 
 runHTTP2ServerWith :: BufferSize -> ((TLS -> IO ()) -> a) -> HTTP2ServerFunc -> a
-runHTTP2ServerWith = runHTTP2ServerWith_ Nothing
+runHTTP2ServerWith = runHTTP2ServerWith_ Nothing (\_sessId -> pure ())
 
-runHTTP2ServerWith_ :: Maybe ExpirationConfig -> BufferSize -> ((TLS -> IO ()) -> a) -> HTTP2ServerFunc -> a
-runHTTP2ServerWith_ expCfg_ bufferSize setup http2Server = setup $ \tls -> do
+runHTTP2ServerWith_ :: Maybe ExpirationConfig -> (SessionId -> IO ()) -> BufferSize -> ((TLS -> IO ()) -> a) -> HTTP2ServerFunc -> a
+runHTTP2ServerWith_ expCfg_ clientFinished bufferSize setup http2Server = setup $ \tls -> do
   activeAt <- newTVarIO =<< getSystemTime
   tid_ <- mapM (forkIO . expireInactiveClient tls activeAt) expCfg_
-  withHTTP2 bufferSize (run activeAt) tls `finally` mapM_ killThread tid_
+  withHTTP2 bufferSize (run tls activeAt) (clientFinished $ tlsUniq tls) tls `finally` mapM_ killThread tid_
   where
-    run activeAt cfg sessId = H.run cfg $ \req _aux sendResp -> do
+    run tls activeAt cfg = H.run cfg $ \req _aux sendResp -> do
       getSystemTime >>= atomically . writeTVar activeAt
-      http2Server sessId req (`sendResp` [])
+      http2Server (tlsUniq tls) (tlsALPN tls) req (`sendResp` [])
     expireInactiveClient tls activeAt expCfg = loop
       where
         loop = do

src/Simplex/Messaging/Transport/Server.hs

@@ -38,7 +38,7 @@ import qualified Data.X509.Validation as XV
 import Network.Socket
 import qualified Network.TLS as T
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
+import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow, atomically')
 import System.Exit (exitFailure)
 import System.Mem.Weak (Weak, deRefWeak)
 import UnliftIO (timeout)
@@ -114,7 +114,7 @@ runTCPServerSocket (accepted, gracefullyClosed, clients) started getSocket serve
     forever . E.bracketOnError (accept sock) (close . fst) $ \(conn, _peer) -> do
       cId <- atomically $ stateTVar accepted $ \cId -> let cId' = cId + 1 in cId `seq` (cId', cId')
       let closeConn _ = do
-            atomically $ modifyTVar' clients $ IM.delete cId 
+            atomically $ modifyTVar' clients $ IM.delete cId
             gracefulClose conn 5000 `catchAll_` pure () -- catchAll_ is needed here in case the connection was closed earlier
             atomically $ modifyTVar' gracefullyClosed (+1)
       tId <- mkWeakThreadId =<< server conn `forkFinally` closeConn
@@ -129,7 +129,7 @@ closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO ()
 closeServer started clients sock = do
   readTVarIO clients >>= mapM_ (deRefWeak >=> mapM_ killThread)
   close sock
-  void . atomically $ tryPutTMVar started False
+  void . atomically' $ tryPutTMVar started False
 
 startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
 startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
@@ -148,7 +148,7 @@ startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
       bind sock $ addrAddress addr
       listen sock 1024
       pure sock
-    setStarted sock = atomically (tryPutTMVar started True) >> pure sock
+    setStarted sock = atomically' (tryPutTMVar started True) >> pure sock
 
 loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
 loadTLSServerParams = loadSupportedTLSServerParams supportedParameters

src/Simplex/Messaging/Util.hs

@@ -4,6 +4,7 @@
 module Simplex.Messaging.Util where
 
 import qualified Control.Exception as E
+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
@@ -19,6 +20,7 @@ import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
 import Data.Time (NominalDiffTime)
 import GHC.Conc (labelThread, myThreadId, threadDelay)
+import GHC.Stack (HasCallStack, withFrozenCallStack)
 import UnliftIO
 import qualified UnliftIO.Exception as UE
 
@@ -167,3 +169,11 @@ diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer)
 
 labelMyThread :: MonadIO m => String -> m ()
 labelMyThread label = liftIO $ myThreadId >>= (`labelThread` label)
+
+{-# INLINE atomically' #-}
+atomically' :: (MonadIO m, HasCallStack) => STM a -> m a
+atomically' f =
+  liftIO $
+    atomically f `UE.catch` \e@E.BlockedIndefinitelyOnSTM -> do
+      withFrozenCallStack $ logError "BlockedIndefinitelyOnSTM"
+      throwIO e

src/Simplex/RemoteControl/Client.hs

@@ -102,16 +102,16 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
   r <- newEmptyTMVarIO
   found@(RCCtrlAddress {address} :| _) <- findCtrlAddress
   c@RCHClient_ {startedPort, announcer} <- liftIO mkClient
-  hostKeys <- atomically genHostKeys
+  hostKeys <- atomically' genHostKeys
   action <- liftIO $ runClient c r hostKeys
   -- wait for the port to make invitation
-  portNum <- atomically $ readTMVar startedPort
+  portNum <- atomically' $ readTMVar startedPort
   signedInv@RCSignedInvitation {invitation} <- maybe (throwError RCETLSStartFailed) (liftIO . mkInvitation hostKeys address) portNum
   when multicast $ case knownHost of
     Nothing -> throwError RCENewController
     Just KnownHostPairing {hostDhPubKey} -> do
       ann <- liftIO . async . runExceptT $ announceRC drg 60 idPrivKey hostDhPubKey hostKeys invitation
-      atomically $ putTMVar announcer ann
+      atomically' $ putTMVar announcer ann
   pure (found, signedInv, RCHostClient {action, client_ = c}, r)
   where
     findCtrlAddress :: ExceptT RCErrorType IO (NonEmpty RCCtrlAddress)
@@ -131,33 +131,33 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
       startTLSServer port_ startedPort tlsCreds (tlsHooks r knownHost hostCAHash) $ \tls ->
         void . runExceptT $ do
           r' <- newEmptyTMVarIO
-          whenM (atomically $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $
+          whenM (atomically' $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $
             runSession tls r' `putRCError` r'
       where
         runSession tls r' = do
           logDebug "Incoming TLS connection"
           hostEncHello <- receiveRCPacket tls
           logDebug "Received host HELLO"
-          hostCA <- atomically $ takeTMVar hostCAHash
+          hostCA <- atomically' $ takeTMVar hostCAHash
           (ctrlEncHello, sessionKeys, helloBody, pairing') <- prepareHostSession drg hostCA pairing hostKeys hostEncHello
           sendRCPacket tls ctrlEncHello
           logDebug "Sent ctrl HELLO"
-          whenM (atomically $ tryPutTMVar r' $ Right (RCHostSession {tls, sessionKeys}, helloBody, pairing')) $ do
-            atomically (tryReadTMVar announcer) >>= mapM_ uninterruptibleCancel
+          whenM (atomically' $ tryPutTMVar r' $ Right (RCHostSession {tls, sessionKeys}, helloBody, pairing')) $ do
+            atomically' (tryReadTMVar announcer) >>= mapM_ uninterruptibleCancel
             -- can use `RCHostSession` until `endSession` is signalled
             logDebug "Holding session"
-            atomically $ takeTMVar endSession
+            atomically' $ takeTMVar endSession
     tlsHooks :: TMVar a -> Maybe KnownHostPairing -> TMVar C.KeyHash -> TLS.ServerHooks
     tlsHooks r knownHost_ hostCAHash =
       def
-        { TLS.onNewHandshake = \_ -> atomically $ isNothing <$> tryReadTMVar r,
+        { TLS.onNewHandshake = \_ -> atomically' $ isNothing <$> tryReadTMVar r,
           TLS.onClientCertificate = \(X509.CertificateChain chain) ->
             case chain of
               [_leaf, ca] -> do
                 let kh = certFingerprint ca
                     accept = maybe True (\h -> hostFingerprint h == kh) knownHost_
                 if accept
-                  then atomically (putTMVar hostCAHash kh) $> TLS.CertificateUsageAccept
+                  then atomically' (putTMVar hostCAHash kh) $> TLS.CertificateUsageAccept
                   else pure $ TLS.CertificateUsageReject TLS.CertificateRejectUnknownCA
               _ ->
                 pure $ TLS.CertificateUsageReject TLS.CertificateRejectUnknownCA
@@ -197,8 +197,8 @@ certFingerprint caCert = C.KeyHash fp
 
 cancelHostClient :: RCHostClient -> IO ()
 cancelHostClient RCHostClient {action, client_ = RCHClient_ {announcer, endSession}} = do
-  atomically $ putTMVar endSession ()
-  atomically (tryTakeTMVar announcer) >>= mapM_ uninterruptibleCancel
+  atomically' $ putTMVar endSession ()
+  atomically' (tryTakeTMVar announcer) >>= mapM_ uninterruptibleCancel
   uninterruptibleCancel action
 
 prepareHostSession :: TVar ChaChaDRG -> C.KeyHash -> RCHostPairing -> RCHostKeys -> RCHostEncHello -> ExceptT RCErrorType IO (RCCtrlEncHello, HostSessKeys, RCHostHello, RCHostPairing)
@@ -285,9 +285,9 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
         liftIO $ peekBuffered tlsBuffer 100000 (TLS.recvData tlsContext) >>= logDebug . tshow -- should normally be ("", Nothing) here
         logDebug "Got TLS connection"
         r' <- newEmptyTMVarIO
-        whenM (atomically $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $ do
+        whenM (atomically' $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $ do
           logDebug "Waiting for session confirmation"
-          whenM (atomically $ readTMVar confirmSession) $ runSession tls r' `putRCError` r'
+          whenM (atomically' $ readTMVar confirmSession) $ runSession tls r' `putRCError` r'
       where
         runSession tls r' = do
           (sharedKey, kemPrivKey, hostEncHello) <- prepareHostHello drg pairing' inv hostAppInfo
@@ -295,11 +295,11 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
           ctrlEncHello <- receiveRCPacket tls
           logDebug "Received ctrl HELLO"
           ctrlSessKeys <- prepareCtrlSession pairing' inv sharedKey kemPrivKey ctrlEncHello
-          whenM (atomically $ tryPutTMVar r' $ Right (RCCtrlSession {tls, sessionKeys = ctrlSessKeys}, pairing')) $ do
+          whenM (atomically' $ tryPutTMVar r' $ Right (RCCtrlSession {tls, sessionKeys = ctrlSessKeys}, pairing')) $ do
             logDebug "Session started"
             -- release second putTMVar in confirmCtrlSession
-            void . atomically $ takeTMVar confirmSession
-            atomically $ takeTMVar endSession
+            void . atomically' $ takeTMVar confirmSession
+            atomically' $ takeTMVar endSession
             logDebug "Session ended"
 
 catchRCError :: ExceptT RCErrorType IO a -> (RCErrorType -> ExceptT RCErrorType IO a) -> ExceptT RCErrorType IO a
@@ -307,7 +307,7 @@ catchRCError = catchAllErrors (RCEException . show)
 {-# INLINE catchRCError #-}
 
 putRCError :: ExceptT RCErrorType IO a -> TMVar (Either RCErrorType b) -> ExceptT RCErrorType IO a
-a `putRCError` r = a `catchRCError` \e -> atomically (tryPutTMVar r $ Left e) >> throwError e
+a `putRCError` r = a `catchRCError` \e -> atomically' (tryPutTMVar r $ Left e) >> throwError e
 
 sendRCPacket :: Encoding a => TLS -> a -> ExceptT RCErrorType IO ()
 sendRCPacket tls pkt = do
@@ -411,14 +411,14 @@ findRCCtrlPairing pairings RCEncInvitation {dhPubKey, nonce, encInvitation} = do
 -- application should call this function when TMVar resolves
 confirmCtrlSession :: RCCtrlClient -> Bool -> IO ()
 confirmCtrlSession RCCtrlClient {client_ = RCCClient_ {confirmSession}} res = do
-  atomically $ putTMVar confirmSession res
+  atomically' $ putTMVar confirmSession res
   -- controler does takeTMVar, freeing the slot
   -- TODO add timeout
-  atomically $ putTMVar confirmSession res -- wait for Ctrl to take the var
+  atomically' $ putTMVar confirmSession res -- wait for Ctrl to take the var
 
 cancelCtrlClient :: RCCtrlClient -> IO ()
 cancelCtrlClient RCCtrlClient {action, client_ = RCCClient_ {endSession}} = do
-  atomically $ putTMVar endSession ()
+  atomically' $ putTMVar endSession ()
   uninterruptibleCancel action
 
 -- * Session encryption

src/Simplex/RemoteControl/Discovery.hs

@@ -27,7 +27,7 @@ import Simplex.Messaging.Transport (supportedParameters)
 import qualified Simplex.Messaging.Transport as Transport
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server (defaultTransportServerConfig, runTransportServerSocket, startTCPServer)
-import Simplex.Messaging.Util (ifM, tshow)
+import Simplex.Messaging.Util (ifM, tshow, atomically')
 import Simplex.RemoteControl.Discovery.Multicast (setMembership)
 import Simplex.RemoteControl.Types
 import UnliftIO
@@ -73,7 +73,7 @@ startTLSServer port_ startedOnPort credentials hooks server = async . liftIO $ d
   started <- newEmptyTMVarIO
   bracketOnError (startTCPServer started $ maybe "0" show port_) (\_e -> setPort Nothing) $ \socket ->
     ifM
-      (atomically $ readTMVar started)
+      (atomically' $ readTMVar started)
       (runServer started socket)
       (setPort Nothing)
   where
@@ -82,7 +82,7 @@ startTLSServer port_ startedOnPort credentials hooks server = async . liftIO $ d
       logInfo $ "System-assigned port: " <> tshow port
       setPort $ Just port
       runTransportServerSocket started (pure socket) "RCP TLS" serverParams defaultTransportServerConfig server
-    setPort = void . atomically . tryPutTMVar startedOnPort
+    setPort = void . atomically' . tryPutTMVar startedOnPort
     serverParams =
       def
         { TLS.serverWantClientCert = True,
@@ -112,19 +112,19 @@ closeListener subscribers sock =
 
 joinMulticast :: TMVar Int -> N.Socket -> N.HostAddress -> IO ()
 joinMulticast subscribers sock group = do
-  now <- atomically $ takeTMVar subscribers
+  now <- atomically' $ takeTMVar subscribers
   when (now == 0) $ do
     setMembership sock group True >>= \case
-      Left e -> atomically (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
-      Right () -> atomically $ putTMVar subscribers (now + 1)
+      Left e -> atomically' (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
+      Right () -> atomically' $ putTMVar subscribers (now + 1)
 
 partMulticast :: TMVar Int -> N.Socket -> N.HostAddress -> IO ()
 partMulticast subscribers sock group = do
-  now <- atomically $ takeTMVar subscribers
+  now <- atomically' $ takeTMVar subscribers
   when (now == 1) $
     setMembership sock group False >>= \case
-      Left e -> atomically (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
-      Right () -> atomically $ putTMVar subscribers (now - 1)
+      Left e -> atomically' (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
+      Right () -> atomically' $ putTMVar subscribers (now - 1)
 
 listenerHostAddr4 :: UDP.ListenSocket -> N.HostAddress
 listenerHostAddr4 sock = case UDP.mySockAddr sock of

tests/AgentTests/FunctionalAPITests.hs

@@ -64,6 +64,7 @@ import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality
 import Data.Word (Word16)
 import qualified Database.SQLite.Simple as SQL
+import GHC.Stack (withFrozenCallStack)
 import SMPAgentClient
 import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn, withSmpServerV7)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
@@ -76,15 +77,16 @@ import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteS
 import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultSMPClientConfig)
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern PQEncOn, pattern PQEncOff, pattern PQSupportOn, pattern PQSupportOff)
+import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Notifications.Transport (NTFVersion, pattern VersionNTF, authBatchCmdsNTFVersion)
+import Simplex.Messaging.Notifications.Transport (NTFVersion, authBatchCmdsNTFVersion, pattern VersionNTF)
 import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolServer (..), SubscriptionMode (..), supportedSMPClientVRange)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, batchCmdsSMPVersion, basicAuthSMPVersion, currentServerSMPRelayVersion)
+import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion)
+import Simplex.Messaging.Util (atomically')
 import Simplex.Messaging.Version (VersionRange (..))
 import qualified Simplex.Messaging.Version as V
 import Simplex.Messaging.Version.Internal (Version (..))
@@ -113,20 +115,20 @@ withTimeout a test =
     Nothing -> error "operation timed out"
     Just t -> liftIO $ test t
 
-get :: MonadIO m => AgentClient -> m (AEntityTransmission 'AEConn)
-get = get' @'AEConn
+get :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AEConn)
+get c = withFrozenCallStack $ get' @'AEConn c
 
-rfGet :: MonadIO m => AgentClient -> m (AEntityTransmission 'AERcvFile)
-rfGet = get' @'AERcvFile
+rfGet :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AERcvFile)
+rfGet c = withFrozenCallStack $ get' @'AERcvFile c
 
-sfGet :: MonadIO m => AgentClient -> m (AEntityTransmission 'AESndFile)
-sfGet = get' @'AESndFile
+sfGet :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AESndFile)
+sfGet c = withFrozenCallStack $ get' @'AESndFile c
 
-nGet :: MonadIO m => AgentClient -> m (AEntityTransmission 'AENone)
-nGet = get' @'AENone
+nGet :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AENone)
+nGet c = withFrozenCallStack $ get' @'AENone c
 
-get' :: forall e m. (MonadIO m, AEntityI e) => AgentClient -> m (AEntityTransmission e)
-get' c = do
+get' :: forall e m. (MonadIO m, AEntityI e, HasCallStack) => AgentClient -> m (AEntityTransmission e)
+get' c = withFrozenCallStack $ do
   (corrId, connId, APC e cmd) <- pGet c
   case testEquality e (sAEntity @e) of
     Just Refl -> pure (corrId, connId, cmd)
@@ -134,7 +136,7 @@ get' c = do
 
 pGet :: forall m. MonadIO m => AgentClient -> m (ATransmission 'Agent)
 pGet c = do
-  t@(_, _, APC _ cmd) <- atomically (readTBQueue $ subQ c)
+  t@(_, _, APC _ cmd) <- atomically' (readTBQueue $ subQ c)
   case cmd of
     CONNECT {} -> pGet c
     DISCONNECT {} -> pGet c
@@ -219,11 +221,11 @@ runRight action =
     Left e -> error $ "Unexpected error: " <> show e
 
 getInAnyOrder :: HasCallStack => AgentClient -> [ATransmission 'Agent -> Bool] -> Expectation
-getInAnyOrder c = inAnyOrder (pGet c)
+getInAnyOrder c ts = withFrozenCallStack $ inAnyOrder (pGet c) ts
 
 inAnyOrder :: (Show a, MonadIO m, HasCallStack) => m a -> [a -> Bool] -> m ()
 inAnyOrder _ [] = pure ()
-inAnyOrder g rs = do
+inAnyOrder g rs = withFrozenCallStack $ do
   r <- g
   let rest = filter (not . expected r) rs
   if length rest < length rs
@@ -280,7 +282,7 @@ functionalAPITests t = do
         testIncreaseConnAgentVersionMaxCompatible t
       it "should increase when connection was negotiated on different versions" $
         testIncreaseConnAgentVersionStartDifferentVersion t
-      -- TODO PQ tests for upgrading connection to PQ encryption
+    -- TODO PQ tests for upgrading connection to PQ encryption
     it "should deliver message after client restart" $
       testDeliverClientRestart t
     it "should deliver messages to the user once, even if repeat delivery is made by the server (no ACK)" $
@@ -440,7 +442,7 @@ canCreateQueue allowNew (srvAuth, srvVersion) (clntAuth, clntVersion) =
 
 testMatrix2 :: ATransport -> (PQSupport -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testMatrix2 t runTest = do
-  it "v7" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn 
+  it "v7" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn
   it "v7 to current" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfg 3 $ runTest PQSupportOn
   it "current to v7" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfgV7 3 $ runTest PQSupportOn
   it "current with v7 server" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfg 3 $ runTest PQSupportOn
@@ -451,10 +453,10 @@ testMatrix2 t runTest = do
 
 testRatchetMatrix2 :: ATransport -> (PQSupport -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testRatchetMatrix2 t runTest = do
-  it "ratchet next" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn 
-  it "ratchet next to current" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfg 3 $ runTest PQSupportOn 
-  it "ratchet current to next" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfgV7 3 $ runTest PQSupportOn 
-  it "ratchet current" $ withSmpServer t $ runTestCfg2 agentCfg agentCfg 3 $ runTest PQSupportOn 
+  it "ratchet next" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn
+  it "ratchet next to current" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfg 3 $ runTest PQSupportOn
+  it "ratchet current to next" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfgV7 3 $ runTest PQSupportOn
+  it "ratchet current" $ withSmpServer t $ runTestCfg2 agentCfg agentCfg 3 $ runTest PQSupportOn
   it "ratchet prev" $ withSmpServer t $ runTestCfg2 agentCfgRatchetVPrev agentCfgRatchetVPrev 3 $ runTest PQSupportOff
   it "ratchets prev to current" $ withSmpServer t $ runTestCfg2 agentCfgRatchetVPrev agentCfg 3 $ runTest PQSupportOff
   it "ratchets current to prev" $ withSmpServer t $ runTestCfg2 agentCfg agentCfgRatchetVPrev 3 $ runTest PQSupportOff
@@ -1366,8 +1368,8 @@ testInactiveNoSubs t = do
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
     alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMOnlyCreate -- do not subscribe to pass noSubscriptions check
-    Just (_, _, APC SAENone (CONNECT _ _)) <- timeout 2000000 $ atomically (readTBQueue $ subQ alice)
-    Just (_, _, APC SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically (readTBQueue $ subQ alice)
+    Just (_, _, APC SAENone (CONNECT _ _)) <- timeout 2000000 $ atomically' (readTBQueue $ subQ alice)
+    Just (_, _, APC SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically' (readTBQueue $ subQ alice)
     disposeAgentClient alice
 
 testInactiveWithSubs :: ATransport -> IO ()
@@ -1654,6 +1656,7 @@ testDeleteConnectionAsync t = do
     pure ([bId1, bId2, bId3] :: [ConnId])
   runRight_ $ do
     deleteConnectionsAsync a False connIds
+    nGet a =##> \case ("", "", DOWN {}) -> True; _ -> False
     get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
     get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
     get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False

tests/AgentTests/NotificationTests.hs

@@ -42,6 +42,7 @@ import Control.Monad.Trans.Except
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Types as JT
 import Data.Bifunctor (bimap, first)
+import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Text.Encoding (encodeUtf8)
@@ -50,7 +51,6 @@ import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, te
 import SMPClient (cfg, cfgV7, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO)
 import Simplex.Messaging.Agent.Store.SQLite (getSavedNtfToken)
@@ -68,6 +68,7 @@ import System.Directory (doesFileExist, removeFile)
 import Test.Hspec
 import UnliftIO
 import Util
+import Simplex.Messaging.Util (atomically')
 
 removeFileIfExists :: FilePath -> IO ()
 removeFileIfExists filePath = do
@@ -170,7 +171,7 @@ testNotificationToken APNSMockServer {apnsQ} = do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     verification <- ntfData .-> "verification"
     nonce <- C.cbNonce <$> ntfData .-> "nonce"
     liftIO $ sendApnsResponse APNSRespOk
@@ -198,13 +199,13 @@ testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     verification <- ntfData .-> "verification"
     nonce <- C.cbNonce <$> ntfData .-> "nonce"
     liftIO $ sendApnsResponse APNSRespOk
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     _ <- ntfData' .-> "verification"
     _ <- C.cbNonce <$> ntfData' .-> "nonce"
     liftIO $ sendApnsResponse' APNSRespOk
@@ -223,7 +224,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     verification <- ntfData .-> "verification"
     nonce <- C.cbNonce <$> ntfData .-> "nonce"
     liftIO $ sendApnsResponse APNSRespOk
@@ -231,7 +232,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
 
     NTRegistered <- registerNtfToken a' tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     verification' <- ntfData' .-> "verification"
     nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
     liftIO $ sendApnsResponse' APNSRespOk
@@ -258,7 +259,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
   ntfData <- withNtfServer t . runRight $ do
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     liftIO $ sendApnsResponse APNSRespOk
     pure ntfData
   -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
@@ -272,7 +273,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
     nonce <- C.cbNonce <$> ntfData .-> "nonce"
     Left (NTF AUTH) <- tryE $ verifyNtfToken a' tkn nonce verification
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     verification' <- ntfData' .-> "verification"
     nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
     liftIO $ sendApnsResponse' APNSRespOk
@@ -295,7 +296,7 @@ testNtfTokenMultipleServers t APNSMockServer {apnsQ} = do
       -- register a new token, the agent picks a server and stores its choice
       NTRegistered <- registerNtfToken a tkn NMPeriodic
       APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-        atomically $ readTBQueue apnsQ
+        atomically' $ readTBQueue apnsQ
       verification <- ntfData .-> "verification"
       nonce <- C.cbNonce <$> ntfData .-> "nonce"
       liftIO $ sendApnsResponse APNSRespOk
@@ -365,7 +366,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} alice@Agen
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken alice tkn NMInstant
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
     verification <- ntfData .-> "verification"
     vNonce <- C.cbNonce <$> ntfData .-> "nonce"
     liftIO $ sendApnsResponse APNSRespOk
@@ -450,7 +451,7 @@ registerTestToken a token mode apnsQ = do
   let tkn = DeviceToken PPApnsTest token
   NTRegistered <- registerNtfToken a tkn mode
   Just APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-    timeout 1000000 . atomically $ readTBQueue apnsQ
+    timeout 1000000 . atomically' $ readTBQueue apnsQ
   verification' <- ntfData' .-> "verification"
   nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
   liftIO $ sendApnsResponse' APNSRespOk
@@ -764,7 +765,7 @@ testMessage_ apnsQ a aId b bId msg = do
 
 messageNotification :: HasCallStack => TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
 messageNotification apnsQ = do
-  1000000 `timeout` atomically (readTBQueue apnsQ) >>= \case
+  1000000 `timeout` atomically' (readTBQueue apnsQ) >>= \case
     Nothing -> error "no notification"
     Just APNSMockRequest {notification = APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData}, sendApnsResponse} -> do
       nonce <- C.cbNonce <$> ntfData .-> "nonce"
@@ -782,6 +783,6 @@ messageNotificationData c apnsQ = do
 
 noNotification :: TBQueue APNSMockRequest -> ExceptT AgentErrorType IO ()
 noNotification apnsQ = do
-  500000 `timeout` atomically (readTBQueue apnsQ) >>= \case
+  500000 `timeout` atomically' (readTBQueue apnsQ) >>= \case
     Nothing -> pure ()
     _ -> error "unexpected notification"

tests/AgentTests/SQLiteTests.hs

@@ -54,6 +54,7 @@ import qualified Simplex.Messaging.Protocol as SMP
 import System.Random
 import Test.Hspec
 import UnliftIO.Directory (removeFile)
+import Simplex.Messaging.Util (atomically')
 
 testDB :: String
 testDB = "tests/tmp/smp-agent.test.db"
@@ -88,7 +89,7 @@ removeStore db = do
   removeFile $ dbFilePath db
   where
     close :: SQLiteStore -> IO ()
-    close st = mapM_ DB.close =<< atomically (tryTakeTMVar $ dbConnection st)
+    close st = mapM_ DB.close =<< atomically' (tryTakeTMVar $ dbConnection st)
 
 storeTests :: Spec
 storeTests = do

tests/CoreTests/CryptoTests.hs

@@ -16,9 +16,14 @@ import Data.Text.Encoding (encodeUtf8)
 import qualified Data.Text.Lazy as LT
 import qualified Data.Text.Lazy.Encoding as LE
 import Data.Type.Equality
+import qualified Data.X509 as X
+import qualified Data.X509.CertificateStore as XS
+import qualified Data.X509.Validation as XV
+import qualified SMPClient
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings
+import Simplex.Messaging.Transport.Client
 import Test.Hspec
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
@@ -91,6 +96,8 @@ cryptoTests = do
     describe "Ed448" $ testEncoding C.SEd448
     describe "X25519" $ testEncoding C.SX25519
     describe "X448" $ testEncoding C.SX448
+  describe "X509 chains" $ do
+    it "should validate certificates" testValidateX509
   describe "sntrup761" $
     it "should enc/dec key" testSNTRUP761
 
@@ -223,6 +230,39 @@ testEncoding alg = it "should encode / decode key" . ioProperty $ do
     C.decodePubKey (C.encodePubKey k) == Right k
       && C.decodePrivKey (C.encodePrivKey pk) == Right pk
 
+testValidateX509 :: IO ()
+testValidateX509 = do
+  let checkChain = validateCertificateChain SMPClient.testKeyHash "localhost" "5223" . X.CertificateChain
+  checkChain [] `shouldReturn` [XV.EmptyChain]
+
+  caCreds <- XS.readCertificates "tests/fixtures/ca.crt"
+  caCreds `shouldNotBe` []
+  let ca = head caCreds
+
+  serverCreds <- XS.readCertificates "tests/fixtures/server.crt"
+  serverCreds `shouldNotBe` []
+  let server = head serverCreds
+  checkChain [server, ca] `shouldReturn` []
+
+  ca2Creds <- XS.readCertificates "tests/fixtures/ca2.crt"
+  ca2Creds `shouldNotBe` []
+  let ca2 = head ca2Creds
+
+  -- signed by another CA
+  server2Creds <- XS.readCertificates "tests/fixtures/server2.crt"
+  server2Creds `shouldNotBe` []
+  let server2 = head server2Creds
+  checkChain [server2, ca2] `shouldReturn` [XV.UnknownCA]
+
+  -- messed up key rotation or other configuration problems
+  checkChain [server2, ca] `shouldReturn` [XV.InvalidSignature XV.SignatureInvalid]
+
+  -- self-signed, unrelated to CA
+  ssCreds <- XS.readCertificates "tests/fixtures/ss.crt"
+  ssCreds `shouldNotBe` []
+  let ss = head ssCreds
+  checkChain [ss, ca] `shouldReturn` [XV.SelfSigned]
+
 testSNTRUP761 :: IO ()
 testSNTRUP761 = do
   drg <- C.newRandom

tests/CoreTests/TRcvQueuesTests.hs

@@ -16,6 +16,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (SMPServer, pattern VersionSMPC)
 import Test.Hspec
 import UnliftIO
+import Simplex.Messaging.Util (atomically')
 
 tRcvQueuesTests :: Spec
 tRcvQueuesTests = do
@@ -28,7 +29,7 @@ tRcvQueuesTests = do
     it "getDelSessQueues" getDelSessQueuesTest
 
 checkDataInvariant :: RQ.TRcvQueues -> IO Bool
-checkDataInvariant trq = atomically $ do
+checkDataInvariant trq = atomically' $ do
   conns <- readTVar $ RQ.getConnections trq
   qs <- readTVar $ RQ.getRcvQueues trq
   -- three invariant checks
@@ -39,87 +40,87 @@ checkDataInvariant trq = atomically $ do
 
 hasConnTest :: IO ()
 hasConnTest = do
-  trq <- atomically RQ.empty
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
+  trq <- atomically' RQ.empty
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
-  atomically (RQ.hasConn "nope" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c3" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "nope" trq) `shouldReturn` False
 
 hasConnTestBatch :: IO ()
 hasConnTestBatch = do
-  trq <- atomically RQ.empty
+  trq <- atomically' RQ.empty
   let qs = [dummyRQ 0 "smp://1234-w==@alpha" "c1", dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@beta" "c3"]
-  atomically $ RQ.batchAddQueues trq qs
+  atomically' $ RQ.batchAddQueues trq qs
   checkDataInvariant trq `shouldReturn` True
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
-  atomically (RQ.hasConn "nope" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c3" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "nope" trq) `shouldReturn` False
 
 deleteConnTest :: IO ()
 deleteConnTest = do
-  trq <- atomically RQ.empty
-  atomically $ do
+  trq <- atomically' RQ.empty
+  atomically' $ do
     RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
     RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
     RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.deleteConn "c1" trq
+  atomically' $ RQ.deleteConn "c1" trq
   checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.deleteConn "nope" trq
+  atomically' $ RQ.deleteConn "nope" trq
   checkDataInvariant trq `shouldReturn` True
   M.keys <$> readTVarIO (RQ.getConnections trq) `shouldReturn` ["c2", "c3"]
 
 getSessQueuesTest :: IO ()
 getSessQueuesTest = do
-  trq <- atomically RQ.empty
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
+  trq <- atomically' RQ.empty
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 1 "smp://1234-w==@beta" "c4") trq
+  atomically' $ RQ.addQueue (dummyRQ 1 "smp://1234-w==@beta" "c4") trq
   checkDataInvariant trq `shouldReturn` True
-  atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c1"]
-  atomically (RQ.getSessQueues (1, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` []
-  atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "nope") trq) `shouldReturn` []
-  atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"]
+  atomically' (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c1"]
+  atomically' (RQ.getSessQueues (1, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` []
+  atomically' (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "nope") trq) `shouldReturn` []
+  atomically' (RQ.getSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"]
 
 getDelSessQueuesTest :: IO ()
 getDelSessQueuesTest = do
-  trq <- atomically RQ.empty
+  trq <- atomically' RQ.empty
   let qs =
         [ dummyRQ 0 "smp://1234-w==@alpha" "c1",
           dummyRQ 0 "smp://1234-w==@alpha" "c2",
           dummyRQ 0 "smp://1234-w==@beta" "c3",
           dummyRQ 1 "smp://1234-w==@beta" "c4"
         ]
-  atomically $ RQ.batchAddQueues trq qs
+  atomically' $ RQ.batchAddQueues trq qs
   checkDataInvariant trq `shouldReturn` True
   -- no user
-  atomically (RQ.getDelSessQueues (2, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
+  atomically' (RQ.getDelSessQueues (2, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
   checkDataInvariant trq `shouldReturn` True
   -- wrong user
-  atomically (RQ.getDelSessQueues (1, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
+  atomically' (RQ.getDelSessQueues (1, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
   checkDataInvariant trq `shouldReturn` True
   -- connections intact
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` True
-  atomically (RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"], ["c1", "c2"])
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` True
+  atomically' (RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"], ["c1", "c2"])
   checkDataInvariant trq `shouldReturn` True
   -- connections gone
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` False
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` False
   -- non-matched connections intact
-  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c4" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c3" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c4" trq) `shouldReturn` True
 
 dummyRQ :: UserId -> SMPServer -> ConnId -> RcvQueue
 dummyRQ userId server connId =

tests/NtfClient.hs

@@ -51,6 +51,7 @@ import UnliftIO.Async
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
+import Simplex.Messaging.Util (atomically')
 
 testHost :: NonEmpty TransportHost
 testHost = "localhost"
@@ -223,7 +224,7 @@ getAPNSMockServer config@HTTP2ServerConfig {qSize} = do
   pure APNSMockServer {action, apnsQ, http2Server}
   where
     runAPNSMockServer apnsQ HTTP2Server {reqQ} = forever $ do
-      HTTP2Request {reqBody = HTTP2Body {bodyHead}, sendResponse} <- atomically $ readTBQueue reqQ
+      HTTP2Request {reqBody = HTTP2Body {bodyHead}, sendResponse} <- atomically' $ readTBQueue reqQ
       let sendApnsResponse = \case
             APNSRespOk -> sendResponse $ H.responseNoBody N.ok200 []
             APNSRespError status reason ->

tests/NtfServerTests.hs

@@ -15,6 +15,7 @@ import Control.Concurrent (threadDelay)
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Types as JT
 import Data.Bifunctor (first)
+import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import Data.Text.Encoding (encodeUtf8)
 import NtfClient
@@ -34,7 +35,6 @@ import ServerTests
 import qualified Simplex.Messaging.Agent.Protocol as AP
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS
@@ -45,6 +45,7 @@ import Simplex.Messaging.Protocol hiding (notification)
 import Simplex.Messaging.Transport
 import Test.Hspec
 import UnliftIO.STM
+import Simplex.Messaging.Util (atomically')
 
 ntfServerTests :: ATransport -> Spec
 ntfServerTests t = do
@@ -112,7 +113,7 @@ testNotificationSubscription (ATransport t) =
           -- register and verify token
           RespNtf "1" "" (NRTknId tId ntfDh) <- signSendRecvNtf nh tknKey ("1", "", TNEW $ NewNtfTkn tkn tknPub dhPub)
           APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse = send} <-
-            atomically $ readTBQueue apnsQ
+            atomically' $ readTBQueue apnsQ
           send APNSRespOk
           let dhSecret = C.dh' ntfDh dhPriv
               Right verification = ntfData .-> "verification"
@@ -131,7 +132,7 @@ testNotificationSubscription (ATransport t) =
           threadDelay 50000
           Resp "5" _ OK <- signSendRecv sh sKey ("5", sId, _SEND' "hello")
           -- receive notification
-          APNSMockRequest {notification, sendApnsResponse = send'} <- atomically $ readTBQueue apnsQ
+          APNSMockRequest {notification, sendApnsResponse = send'} <- atomically' $ readTBQueue apnsQ
           let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData'} = notification
               Right nonce' = C.cbNonce <$> ntfData' .-> "nonce"
               Right message = ntfData' .-> "message"
@@ -155,7 +156,7 @@ testNotificationSubscription (ATransport t) =
           RespNtf "7" tId' NROk <- signSendRecvNtf nh tknKey ("7", tId, TRPL tkn')
           tId `shouldBe` tId'
           APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData2}, sendApnsResponse = send2} <-
-            atomically $ readTBQueue apnsQ
+            atomically' $ readTBQueue apnsQ
           send2 APNSRespOk
           let Right verification2 = ntfData2 .-> "verification"
               Right nonce2 = C.cbNonce <$> ntfData2 .-> "nonce"
@@ -164,7 +165,7 @@ testNotificationSubscription (ATransport t) =
           RespNtf "8a" _ (NRTkn NTActive) <- signSendRecvNtf nh tknKey ("8a", tId, TCHK)
           -- send message
           Resp "9" _ OK <- signSendRecv sh sKey ("9", sId, _SEND' "hello 2")
-          APNSMockRequest {notification = notification3, sendApnsResponse = send3} <- atomically $ readTBQueue apnsQ
+          APNSMockRequest {notification = notification3, sendApnsResponse = send3} <- atomically' $ readTBQueue apnsQ
           let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData3} = notification3
               Right nonce3 = C.cbNonce <$> ntfData3 .-> "nonce"
               Right message3 = ntfData3 .-> "message"

tests/RemoteControl.hs

@@ -19,6 +19,7 @@ import Simplex.RemoteControl.Types
 import Test.Hspec
 import UnliftIO
 import UnliftIO.Concurrent
+import Simplex.Messaging.Util (atomically')
 
 remoteControlTests :: Spec
 remoteControlTests = do
@@ -72,9 +73,9 @@ testNewPairing = do
     logNote "c 2"
     putMVar invVar (inv, hc)
     logNote "c 3"
-    Right (sessId, _tls, r') <- atomically $ takeTMVar r
+    Right (sessId, _tls, r') <- atomically' $ takeTMVar r
     logNote "c 4"
-    Right (_rcHostSession, _rcHelloBody, _hp') <- atomically $ takeTMVar r'
+    Right (_rcHostSession, _rcHelloBody, _hp') <- atomically' $ takeTMVar r'
     logNote "c 5"
     threadDelay 250000
     logNote "ctrl: ciao"
@@ -89,11 +90,11 @@ testNewPairing = do
     logNote "h 1"
     (rcCtrlClient, r) <- RC.connectRCCtrl drg inv Nothing (J.String "app")
     logNote "h 2"
-    Right (sessId', _tls, r') <- atomically $ takeTMVar r
+    Right (sessId', _tls, r') <- atomically' $ takeTMVar r
     logNote "h 3"
     liftIO $ RC.confirmCtrlSession rcCtrlClient True
     logNote "h 4"
-    Right (_rcCtrlSession, _rcCtrlPairing) <- atomically $ takeTMVar r'
+    Right (_rcCtrlSession, _rcCtrlPairing) <- atomically' $ takeTMVar r'
     logNote "h 5"
     threadDelay 250000
     logNote "ctrl: adios"
@@ -162,8 +163,8 @@ runCtrl :: TVar ChaChaDRG -> Bool -> RCHostPairing -> MVar RCSignedInvitation ->
 runCtrl drg multicast hp invVar = async . runRight $ do
   (_found, inv, hc, r) <- RC.connectRCHost drg hp (J.String "app") multicast Nothing Nothing
   putMVar invVar inv
-  Right (_sessId, _tls, r') <- atomically $ takeTMVar r
-  Right (_rcHostSession, _rcHelloBody, hp') <- atomically $ takeTMVar r'
+  Right (_sessId, _tls, r') <- atomically' $ takeTMVar r
+  Right (_rcHostSession, _rcHelloBody, hp') <- atomically' $ takeTMVar r'
   threadDelay 250000
   liftIO $ RC.cancelHostClient hc
   pure hp'
@@ -172,9 +173,9 @@ runHostURI :: TVar ChaChaDRG -> Maybe RCCtrlPairing -> RCSignedInvitation -> IO
 runHostURI drg cp_ signedInv = async . runRight $ do
   inv <- maybe (fail "bad invite") pure $ verifySignedInvitation signedInv
   (rcCtrlClient, r) <- RC.connectRCCtrl drg inv cp_ (J.String "app")
-  Right (_sessId', _tls, r') <- atomically $ takeTMVar r
+  Right (_sessId', _tls, r') <- atomically' $ takeTMVar r
   liftIO $ RC.confirmCtrlSession rcCtrlClient True
-  Right (_rcCtrlSession, cp') <- atomically $ takeTMVar r'
+  Right (_rcCtrlSession, cp') <- atomically' $ takeTMVar r'
   threadDelay 250000
   pure cp'
 
@@ -182,8 +183,8 @@ runHostMulticast :: TVar ChaChaDRG -> TMVar Int -> RCCtrlPairing -> IO (Async RC
 runHostMulticast drg subscribers cp = async . runRight $ do
   (pairing, inv) <- RC.discoverRCCtrl subscribers (cp :| [])
   (rcCtrlClient, r) <- RC.connectRCCtrl drg inv (Just pairing) (J.String "app")
-  Right (_sessId', _tls, r') <- atomically $ takeTMVar r
+  Right (_sessId', _tls, r') <- atomically' $ takeTMVar r
   liftIO $ RC.confirmCtrlSession rcCtrlClient True
-  Right (_rcCtrlSession, cp') <- atomically $ takeTMVar r'
+  Right (_rcCtrlSession, cp') <- atomically' $ takeTMVar r'
   threadDelay 250000
   pure cp'

tests/SMPClient.hs

@@ -25,6 +25,7 @@ import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
 import Simplex.Messaging.Transport.Server
+import Simplex.Messaging.Util (atomically')
 import Simplex.Messaging.Version (mkVersionRange)
 import System.Environment (lookupEnv)
 import System.Info (os)
@@ -123,7 +124,7 @@ withSmpServerConfigOn :: HasCallStack => ATransport -> ServerConfig -> ServiceNa
 withSmpServerConfigOn t cfg' port' =
   serverBracket
     (\started -> runSMPServerBlocking started cfg' {transports = [(port', t)]})
-    (pure ())
+    (threadDelay 10000)
 
 withSmpServerThreadOn :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerThreadOn t = withSmpServerConfigOn t cfg
@@ -137,7 +138,7 @@ serverBracket process afterProcess f = do
     (\t -> waitFor started "start" >> f t >>= \r -> r <$ threadDelay 100000)
   where
     waitFor started s =
-      5_000_000 `timeout` atomically (takeTMVar started) >>= \case
+      5_000_000 `timeout` atomically' (takeTMVar started) >>= \case
         Nothing -> error $ "server did not " <> s
         _ -> pure ()
 

tests/ServerTests.hs

@@ -22,6 +22,7 @@ import Control.Exception (SomeException, try)
 import Control.Monad
 import Control.Monad.IO.Class
 import Data.Bifunctor (first)
+import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.Set as S
@@ -30,7 +31,6 @@ import GHC.Stack (withFrozenCallStack)
 import SMPClient
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Encoding.Base64 (encode)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol
@@ -44,6 +44,7 @@ import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
 import Test.Hspec
+import Simplex.Messaging.Util (atomically')
 
 serverTests :: ATransport -> Spec
 serverTests t@(ATransport t') = do
@@ -396,9 +397,9 @@ testGetCommand t =
     smpTest t $ \sh -> do
       queue <- newEmptyTMVarIO
       testSMPClient @c $ \rh ->
-        atomically . putTMVar queue =<< createAndSecureQueue rh sPub
+        atomically' . putTMVar queue =<< createAndSecureQueue rh sPub
       testSMPClient @c $ \rh -> do
-        (sId, rId, rKey, dhShared) <- atomically $ takeTMVar queue
+        (sId, rId, rKey, dhShared) <- atomically' $ takeTMVar queue
         let dec = decryptMsgV3 dhShared
         Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, _SEND "hello")
         Resp "2" _ (Msg mId1 msg1) <- signSendRecv rh rKey ("2", rId, GET)

tests/XFTPClient.hs

@@ -5,7 +5,7 @@
 
 module XFTPClient where
 
-import Control.Concurrent (ThreadId)
+import Control.Concurrent (ThreadId, threadDelay)
 import Data.String (fromString)
 import Network.Socket (ServiceName)
 import SMPClient (serverBracket)
@@ -13,6 +13,7 @@ import Simplex.FileTransfer.Client
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Server (runXFTPServerBlocking)
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defaultFileExpiration, defaultInactiveClientExpiration)
+import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (XFTPServer)
 import Simplex.Messaging.Transport.Server
 import Test.Hspec
@@ -52,7 +53,7 @@ withXFTPServerCfg :: HasCallStack => XFTPServerConfig -> (HasCallStack => Thread
 withXFTPServerCfg cfg =
   serverBracket
     (`runXFTPServerBlocking` cfg)
-    (pure ())
+    (threadDelay 10000)
 
 withXFTPServerThreadOn :: HasCallStack => (HasCallStack => ThreadId -> IO a) -> IO a
 withXFTPServerThreadOn = withXFTPServerCfg testXFTPServerConfig
@@ -124,7 +125,8 @@ testXFTPClientConfig :: XFTPClientConfig
 testXFTPClientConfig = defaultXFTPClientConfig
 
 testXFTPClient :: HasCallStack => (HasCallStack => XFTPClient -> IO a) -> IO a
-testXFTPClient client =
-  getXFTPClient (1, testXFTPServer, Nothing) testXFTPClientConfig (\_ -> pure ()) >>= \case
+testXFTPClient client = do
+  g <- C.newRandom
+  getXFTPClient g (1, testXFTPServer, Nothing) testXFTPClientConfig (\_ -> pure ()) >>= \case
     Right c -> client c
     Left e -> error $ show e

tests/XFTPServerTests.hs

@@ -12,6 +12,7 @@ import Control.Exception (SomeException)
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
+import qualified Data.ByteString.Base64.URL as B64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
@@ -21,11 +22,10 @@ import Simplex.FileTransfer.Client
 import Simplex.FileTransfer.Description (kb)
 import Simplex.FileTransfer.Protocol (FileInfo (..))
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..))
-import Simplex.FileTransfer.Transport (XFTPRcvChunkSpec (..), XFTPErrorType (..))
+import Simplex.FileTransfer.Transport (XFTPErrorType (..), XFTPRcvChunkSpec (..))
 import Simplex.Messaging.Client (ProtocolClientError (..))
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
-import qualified Simplex.Messaging.Encoding.Base64.URL as U
 import Simplex.Messaging.Protocol (BasicAuth, SenderId)
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
 import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive, removeFile)
@@ -33,6 +33,7 @@ import System.FilePath ((</>))
 import Test.Hspec
 import UnliftIO.STM
 import XFTPClient
+import Simplex.Messaging.Util (atomically')
 
 xftpServerTests :: Spec
 xftpServerTests =
@@ -75,7 +76,7 @@ createTestChunk fp = do
   pure bytes
 
 readChunk :: SenderId -> IO ByteString
-readChunk sId = B.readFile (xftpServerFiles </> B.unpack (U.encode sId))
+readChunk sId = B.readFile (xftpServerFiles </> B.unpack (B64.encode sId))
 
 testFileChunkDelivery :: Expectation
 testFileChunkDelivery = xftpTest $ \c -> runRight_ $ runTestFileChunkDelivery c c
@@ -185,7 +186,7 @@ testWrongChunkSize = xftpTest $ \c -> do
   g <- C.newRandom
   (sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
   (rcvKey, _rpKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-  B.writeFile testChunkPath =<< atomically (C.randomBytes (kb 96) g)
+  B.writeFile testChunkPath =<< atomically' (C.randomBytes (kb 96) g)
   digest <- LC.sha256Hash <$> LB.readFile testChunkPath
   let file = FileInfo {sndKey, size = kb 96, digest}
   runRight_ $
@@ -219,15 +220,16 @@ testFileChunkExpiration = withXFTPServerCfg testXFTPServerConfig {fileExpiration
 testInactiveClientExpiration :: Expectation
 testInactiveClientExpiration = withXFTPServerCfg testXFTPServerConfig {inactiveClientExpiration} $ \_ -> runRight_ $ do
   disconnected <- newEmptyTMVarIO
-  c <- ExceptT $ getXFTPClient (1, testXFTPServer, Nothing) testXFTPClientConfig (\_ -> atomically $ putTMVar disconnected ())
+  g <- liftIO C.newRandom
+  c <- ExceptT $ getXFTPClient g (1, testXFTPServer, Nothing) testXFTPClientConfig (\_ -> atomically' $ putTMVar disconnected ())
   pingXFTP c
   liftIO $ do
     threadDelay 100000
-    atomically (tryReadTMVar disconnected) `shouldReturn` Nothing
+    atomically' (tryReadTMVar disconnected) `shouldReturn` Nothing
   pingXFTP c
   liftIO $ do
     threadDelay 3000000
-    atomically (tryTakeTMVar disconnected) `shouldReturn` Just ()
+    atomically' (tryTakeTMVar disconnected) `shouldReturn` Just ()
   where
     inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}
 

tests/fixtures/ca2.crt

@@ -0,0 +1,12 @@
+-----BEGIN CERTIFICATE-----
+MIIBtjCCATagAwIBAgIUDJc0ixVBYPdcL5W7zE8dhm0UMpswBQYDK2VxMCoxFjAU
+BgNVBAMMDVNNUCBzZXJ2ZXIgQ0ExEDAOBgNVBAoMB1NpbXBsZVgwIBcNMjQwNDA4
+MTg0ODIwWhgPNDc2MjAzMDUxODQ4MjBaMCoxFjAUBgNVBAMMDVNNUCBzZXJ2ZXIg
+Q0ExEDAOBgNVBAoMB1NpbXBsZVgwQzAFBgMrZXEDOgAST4assVdIwL/kbtWmbyJm
+X/CNGUQFkArvgvcRTZOwJPu9ypmv0mSz2I6acsw6gr8LHq8mlv7iPICjUzBRMB0G
+A1UdDgQWBBTBsA6VVhkO61ixwlel+g7D08shnjAfBgNVHSMEGDAWgBTBsA6VVhkO
+61ixwlel+g7D08shnjAPBgNVHRMBAf8EBTADAQH/MAUGAytlcQNzAKAfQ0EEQtnR
+HvNiKBajo77prZX680apmxBxSZuLNORQMvKBLDm2qaGv5S/c9gmvLjLz2Avrspow
+ANBF71DKcvgb25D2LLDp0CQOBt/dP41Cgd/ZigyHyOq2/Oj15Skbu0TdXYuIxf/k
+MZ0XUvYwG6IKAA==
+-----END CERTIFICATE-----

tests/fixtures/ca2.key

@@ -0,0 +1,4 @@
+-----BEGIN PRIVATE KEY-----
+MEcCAQAwBQYDK2VxBDsEOcBpozc2TnAf6lQaxN5bA6JdbKWuxUecsW9P2dzncCnB
+/alBtYXqW6SprBj1DqzeZyU4rQ7OqFrgBw==
+-----END PRIVATE KEY-----

tests/fixtures/server2.crt

@@ -0,0 +1,12 @@
+-----BEGIN CERTIFICATE-----
+MIIBvDCCATygAwIBAgIUbx6kKw7PGGxhTPutroJFZbOcVk0wBQYDK2VxMCoxFjAU
+BgNVBAMMDVNNUCBzZXJ2ZXIgQ0ExEDAOBgNVBAoMB1NpbXBsZVgwIBcNMjQwNDA4
+MTg0ODI2WhgPNDc2MjAzMDUxODQ4MjZaMBQxEjAQBgNVBAMMCWxvY2FsaG9zdDBD
+MAUGAytlcQM6ACKox7DzkUjK6ZN0pCzABv5vcqk5Tu+zaLWEWlHFnpIN/f/AcBI1
+GbZCmD/zb6OG49vsAKPnMAyIgKNvMG0wCQYDVR0TBAIwADALBgNVHQ8EBAMCA8gw
+EwYDVR0lBAwwCgYIKwYBBQUHAwEwHQYDVR0OBBYEFGFx/ISB2xEW2tGhVYVncWTd
+lwkrMB8GA1UdIwQYMBaAFMGwDpVWGQ7rWLHCV6X6DsPTyyGeMAUGAytlcQNzAMp9
+EL+22OkeGG6s7LxpXJgVG6dxbcNn6aTgTX2pDYt8n+cRQTeTZ1MLDYVIe289pIQK
+tbKmI+HIgHExuNurJw6f6FknVmEeJpOXLV5lybL4f/fZGKrAE5rbhtNnQAp1mw0c
+ngt8dhyISxv/zoQLSkIcAA==
+-----END CERTIFICATE-----

tests/fixtures/server2.key

@@ -0,0 +1,4 @@
+-----BEGIN PRIVATE KEY-----
+MEcCAQAwBQYDK2VxBDsEObrN+1gIRcwmahmitb6ltVoZjjnVoHj0/1waYkjmMtQl
+PiGhWP5/B6Y1fLH/YiO/tfX2YPGCOJJSJQ==
+-----END PRIVATE KEY-----

tests/fixtures/ss.crt

@@ -0,0 +1,13 @@
+-----BEGIN CERTIFICATE-----
+MIIB6zCCAXKgAwIBAgIUbLI6PjnyP24ukjmIE3LsjFqn/LYwCgYIKoZIzj0EAwIw
+FjEUMBIGA1UEAwwLZXhhbXBsZS5jb20wHhcNMjQwNDA4MTg1OTA2WhcNMzQwNDA2
+MTg1OTA2WjAWMRQwEgYDVQQDDAtleGFtcGxlLmNvbTB2MBAGByqGSM49AgEGBSuB
+BAAiA2IABMIrmyP4FDY+P8Tulv8Bcp5U7QlHigoOW6JPRPTETTFBl2e7t9UApa/E
+AYl805mkaIdrDJzdtAqkttHmPm4vXdCCualxVRZ/thtpvdNocxyJOD9BVv3QKqiu
+SGuCGHp+m6OBgDB+MB0GA1UdDgQWBBR5D79SM77XfY/bii0NJ1OllAWw1TAfBgNV
+HSMEGDAWgBR5D79SM77XfY/bii0NJ1OllAWw1TAPBgNVHRMBAf8EBTADAQH/MCsG
+A1UdEQQkMCKCC2V4YW1wbGUuY29tgg0qLmV4YW1wbGUuY29thwQKAAABMAoGCCqG
+SM49BAMCA2cAMGQCMDuIOZBcKI/OXOWx75o5xgwIDio4P7zK9kJt7D4YJMxPvTV6
+vVajYSuJwiIF3/GwoQIwPUbNndNNnf1tYJdPhEJ3e8bA2a3bDbb2dgfiUfj6amaS
+RcSkYms1WDLMFP0LHo/Z
+-----END CERTIFICATE-----

tests/fixtures/ss.key

@@ -0,0 +1,6 @@
+-----BEGIN PRIVATE KEY-----
+MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDBH/6IpMqcKFrXNU8Nb
+QFvdzQOJtfoAEEDRBmMqbihPrCgbtCJ3FIVnxqGlFIXADaqhZANiAATCK5sj+BQ2
+Pj/E7pb/AXKeVO0JR4oKDluiT0T0xE0xQZdnu7fVAKWvxAGJfNOZpGiHawyc3bQK
+pLbR5j5uL13QgrmpcVUWf7Ybab3TaHMciTg/QVb90Cqorkhrghh6fps=
+-----END PRIVATE KEY-----