mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-29 09:58:41 +00:00
agent: transport isolation mode "Session" (default) to use new SOCKS credentials when client restarts or SOCKS proxy configuration changes (#1321)
* agent: transport isolation mode "Session" (default) to use new SOCKS credentials when client restarts or SOCKS proxy configuration changes * fix test
This commit is contained in:
Evgeny
@@ -22,6 +22,7 @@ import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Int (Int64)
|
||||
import Data.List.NonEmpty (NonEmpty (..))
|
||||
import Data.Time.Clock (UTCTime)
|
||||
import Data.Word (Word32)
|
||||
import qualified Data.X509 as X
|
||||
import qualified Data.X509.Validation as XV
|
||||
@@ -36,7 +37,6 @@ import Simplex.Messaging.Client
|
||||
TransportSession,
|
||||
chooseTransportHost,
|
||||
defaultNetworkConfig,
|
||||
proxyUsername,
|
||||
transportClientConfig,
|
||||
clientSocksCredentials,
|
||||
unexpectedResponse,
|
||||
@@ -99,9 +99,9 @@ defaultXFTPClientConfig =
|
||||
clientALPN = Just supportedXFTPhandshakes
|
||||
}
|
||||
|
||||
getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
|
||||
getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, xftpNetworkConfig, serverVRange} disconnected = runExceptT $ do
|
||||
let socksCreds = clientSocksCredentials xftpNetworkConfig $ proxyUsername transportSession
|
||||
getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> UTCTime -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
|
||||
getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, xftpNetworkConfig, serverVRange} proxySessTs disconnected = runExceptT $ do
|
||||
let socksCreds = clientSocksCredentials xftpNetworkConfig proxySessTs transportSession
|
||||
ProtocolServer _ host port keyHash = srv
|
||||
useHost <- liftEither $ chooseTransportHost xftpNetworkConfig host
|
||||
let tcConfig = (transportClientConfig xftpNetworkConfig useHost) {alpn = clientALPN}
|
||||
|
||||
@@ -16,6 +16,7 @@ import Data.Bifunctor (first)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Text (Text)
|
||||
import Data.Text.Encoding (decodeUtf8)
|
||||
import Data.Time.Clock (UTCTime, getCurrentTime)
|
||||
import Simplex.FileTransfer.Client
|
||||
import Simplex.Messaging.Agent.RetryInterval
|
||||
import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientError (..), temporaryClientError)
|
||||
@@ -30,6 +31,7 @@ type XFTPClientVar = TMVar (Either XFTPClientAgentError XFTPClient)
|
||||
|
||||
data XFTPClientAgent = XFTPClientAgent
|
||||
{ xftpClients :: TMap XFTPServer XFTPClientVar,
|
||||
startedAt :: UTCTime,
|
||||
config :: XFTPClientAgentConfig
|
||||
}
|
||||
|
||||
@@ -56,19 +58,20 @@ data XFTPClientAgentError = XFTPClientAgentError XFTPServer XFTPClientError
|
||||
newXFTPAgent :: XFTPClientAgentConfig -> IO XFTPClientAgent
|
||||
newXFTPAgent config = do
|
||||
xftpClients <- TM.emptyIO
|
||||
pure XFTPClientAgent {xftpClients, config}
|
||||
startedAt <- getCurrentTime
|
||||
pure XFTPClientAgent {xftpClients, startedAt, config}
|
||||
|
||||
type ME a = ExceptT XFTPClientAgentError IO a
|
||||
|
||||
getXFTPServerClient :: XFTPClientAgent -> XFTPServer -> ME XFTPClient
|
||||
getXFTPServerClient XFTPClientAgent {xftpClients, config} srv = do
|
||||
getXFTPServerClient XFTPClientAgent {xftpClients, startedAt, 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 (1, srv, Nothing) (xftpConfig config) startedAt clientDisconnected
|
||||
|
||||
clientDisconnected :: XFTPClient -> IO ()
|
||||
clientDisconnected _ = do
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
{-# LANGUAGE BangPatterns #-}
|
||||
{-# LANGUAGE ConstraintKinds #-}
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE DuplicateRecordFields #-}
|
||||
@@ -473,14 +474,14 @@ testProtocolServer c userId srv = withAgentEnv' c $ case protocolTypeI @p of
|
||||
|
||||
-- | set SOCKS5 proxy on/off and optionally set TCP timeouts for fast network
|
||||
setNetworkConfig :: AgentClient -> NetworkConfig -> IO ()
|
||||
setNetworkConfig c@AgentClient {useNetworkConfig} cfg' = do
|
||||
changed <- atomically $ do
|
||||
setNetworkConfig c@AgentClient {useNetworkConfig, proxySessTs} cfg' = do
|
||||
(spChanged, changed) <- atomically $ do
|
||||
(_, cfg) <- readTVar useNetworkConfig
|
||||
if cfg == cfg'
|
||||
then pure False
|
||||
else
|
||||
let cfgSlow = slowNetworkConfig cfg'
|
||||
in True <$ (cfgSlow `seq` writeTVar useNetworkConfig (cfgSlow, cfg'))
|
||||
let changed = cfg /= cfg'
|
||||
!cfgSlow = slowNetworkConfig cfg'
|
||||
when changed $ writeTVar useNetworkConfig (cfgSlow, cfg')
|
||||
pure (socksProxy cfg /= socksProxy cfg', changed)
|
||||
when spChanged $ getCurrentTime >>= atomically . writeTVar proxySessTs
|
||||
when changed $ reconnectAllServers c
|
||||
|
||||
setUserNetworkInfo :: AgentClient -> UserNetworkInfo -> IO ()
|
||||
|
||||
@@ -333,6 +333,7 @@ data AgentClient = AgentClient
|
||||
smpSubWorkers :: TMap SMPTransportSession (SessionVar (Async ())),
|
||||
clientId :: Int,
|
||||
agentEnv :: Env,
|
||||
proxySessTs :: TVar UTCTime,
|
||||
smpServersStats :: TMap (UserId, SMPServer) AgentSMPServerStats,
|
||||
xftpServersStats :: TMap (UserId, XFTPServer) AgentXFTPServerStats,
|
||||
ntfServersStats :: TMap (UserId, NtfServer) AgentNtfServerStats,
|
||||
@@ -461,6 +462,7 @@ newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Env -> IO AgentClient
|
||||
newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs agentEnv = do
|
||||
let cfg = config agentEnv
|
||||
qSize = tbqSize cfg
|
||||
proxySessTs <- newTVarIO =<< getCurrentTime
|
||||
acThread <- newTVarIO Nothing
|
||||
active <- newTVarIO True
|
||||
subQ <- newTBQueueIO qSize
|
||||
@@ -533,6 +535,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
|
||||
smpSubWorkers,
|
||||
clientId,
|
||||
agentEnv,
|
||||
proxySessTs,
|
||||
smpServersStats,
|
||||
xftpServersStats,
|
||||
ntfServersStats,
|
||||
@@ -657,7 +660,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq
|
||||
Nothing -> Left $ BROKER (B.unpack $ strEncode srv) TIMEOUT
|
||||
|
||||
smpConnectClient :: AgentClient -> SMPTransportSession -> TMap SMPServer ProxiedRelayVar -> SMPClientVar -> AM SMPConnectedClient
|
||||
smpConnectClient c@AgentClient {smpClients, msgQ} tSess@(_, srv, _) prs v =
|
||||
smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs} tSess@(_, srv, _) prs v =
|
||||
newProtocolClient c tSess smpClients connectClient v
|
||||
`catchAgentError` \e -> lift (resubscribeSMPSession c tSess) >> throwE e
|
||||
where
|
||||
@@ -667,7 +670,8 @@ smpConnectClient c@AgentClient {smpClients, msgQ} tSess@(_, srv, _) prs v =
|
||||
g <- asks random
|
||||
env <- ask
|
||||
liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ do
|
||||
smp <- ExceptT $ getProtocolClient g tSess cfg (Just msgQ) $ smpClientDisconnected c tSess env v' prs
|
||||
ts <- readTVarIO proxySessTs
|
||||
smp <- ExceptT $ getProtocolClient g tSess cfg (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
|
||||
pure SMPConnectedClient {connectedClient = smp, proxiedRelays = prs}
|
||||
|
||||
smpClientDisconnected :: AgentClient -> SMPTransportSession -> Env -> SMPClientVar -> TMap SMPServer ProxiedRelayVar -> SMPClient -> IO ()
|
||||
@@ -756,7 +760,7 @@ reconnectSMPClient c tSess@(_, srv, _) qs = handleNotify $ do
|
||||
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, AEvt (sAEntity @e) cmd)
|
||||
|
||||
getNtfServerClient :: AgentClient -> NtfTransportSession -> AM NtfClient
|
||||
getNtfServerClient c@AgentClient {active, ntfClients, workerSeq} tSess@(_, srv, _) = do
|
||||
getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs} tSess@(_, srv, _) = do
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess ntfClients ts)
|
||||
@@ -768,8 +772,9 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq} tSess@(_, srv,
|
||||
connectClient v = do
|
||||
cfg <- lift $ getClientConfig c ntfCfg
|
||||
g <- asks random
|
||||
ts <- readTVarIO proxySessTs
|
||||
liftError' (protocolClientError NTF $ B.unpack $ strEncode srv) $
|
||||
getProtocolClient g tSess cfg Nothing $
|
||||
getProtocolClient g tSess cfg Nothing ts $
|
||||
clientDisconnected v
|
||||
|
||||
clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
|
||||
@@ -779,7 +784,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq} tSess@(_, srv,
|
||||
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
|
||||
|
||||
getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
|
||||
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq} tSess@(_, srv, _) = do
|
||||
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs} tSess@(_, srv, _) = do
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess xftpClients ts)
|
||||
@@ -791,8 +796,9 @@ getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq} tSess@(_, srv
|
||||
connectClient v = do
|
||||
cfg <- asks $ xftpCfg . config
|
||||
xftpNetworkConfig <- getNetworkConfig c
|
||||
ts <- readTVarIO proxySessTs
|
||||
liftError' (protocolClientError XFTP $ B.unpack $ strEncode srv) $
|
||||
X.getXFTPClient tSess cfg {xftpNetworkConfig} $
|
||||
X.getXFTPClient tSess cfg {xftpNetworkConfig} ts $
|
||||
clientDisconnected v
|
||||
|
||||
clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
|
||||
@@ -1199,7 +1205,8 @@ runSMPServerTest c userId (ProtoServerWithAuth srv auth) = do
|
||||
g <- asks random
|
||||
liftIO $ do
|
||||
let tSess = (userId, srv, Nothing)
|
||||
getProtocolClient g tSess cfg Nothing (\_ -> pure ()) >>= \case
|
||||
ts <- readTVarIO $ proxySessTs c
|
||||
getProtocolClient g tSess cfg Nothing ts (\_ -> pure ()) >>= \case
|
||||
Right smp -> do
|
||||
rKeys@(_, rpKey) <- atomically $ C.generateAuthKeyPair ra g
|
||||
(sKey, spKey) <- atomically $ C.generateAuthKeyPair sa g
|
||||
@@ -1229,7 +1236,8 @@ runXFTPServerTest c userId (ProtoServerWithAuth srv auth) = do
|
||||
rcvPath <- getTempFilePath workDir
|
||||
liftIO $ do
|
||||
let tSess = (userId, srv, Nothing)
|
||||
X.getXFTPClient tSess cfg {xftpNetworkConfig} (\_ -> pure ()) >>= \case
|
||||
ts <- readTVarIO $ proxySessTs c
|
||||
X.getXFTPClient tSess cfg {xftpNetworkConfig} ts (\_ -> pure ()) >>= \case
|
||||
Right xftp -> withTestChunk filePath $ do
|
||||
(sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
|
||||
(rcvKey, rpKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
|
||||
@@ -1273,7 +1281,8 @@ runNTFServerTest c userId (ProtoServerWithAuth srv _) = do
|
||||
g <- asks random
|
||||
liftIO $ do
|
||||
let tSess = (userId, srv, Nothing)
|
||||
getProtocolClient g tSess cfg Nothing (\_ -> pure ()) >>= \case
|
||||
ts <- readTVarIO $ proxySessTs c
|
||||
getProtocolClient g tSess cfg Nothing ts (\_ -> pure ()) >>= \case
|
||||
Right ntf -> do
|
||||
(nKey, npKey) <- atomically $ C.generateAuthKeyPair a g
|
||||
(dhKey, _) <- atomically $ C.generateKeyPair g
|
||||
|
||||
@@ -82,7 +82,6 @@ module Simplex.Messaging.Client
|
||||
transportClientConfig,
|
||||
clientSocksCredentials,
|
||||
chooseTransportHost,
|
||||
proxyUsername,
|
||||
temporaryClientError,
|
||||
smpProxyError,
|
||||
textToHostMode,
|
||||
@@ -118,6 +117,7 @@ import qualified Data.Aeson.TH as J
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import qualified Data.ByteString.Base64 as B64
|
||||
import Data.Functor (($>))
|
||||
import Data.Int (Int64)
|
||||
import Data.List (find)
|
||||
@@ -299,7 +299,7 @@ data NetworkConfig = NetworkConfig
|
||||
}
|
||||
deriving (Eq, Show)
|
||||
|
||||
data TransportSessionMode = TSMUser | TSMEntity
|
||||
data TransportSessionMode = TSMUser | TSMSession | TSMServer | TSMEntity
|
||||
deriving (Eq, Show)
|
||||
|
||||
-- SMP proxy mode for sending messages
|
||||
@@ -349,7 +349,7 @@ defaultNetworkConfig =
|
||||
socksMode = SMAlways,
|
||||
hostMode = HMOnionViaSocks,
|
||||
requiredHostMode = False,
|
||||
sessionMode = TSMUser,
|
||||
sessionMode = TSMSession,
|
||||
smpProxyMode = SPMNever,
|
||||
smpProxyFallback = SPFAllow,
|
||||
tcpConnectTimeout = defaultTcpConnectTimeout,
|
||||
@@ -371,16 +371,22 @@ transportClientConfig NetworkConfig {socksProxy, socksMode, tcpConnectTimeout, t
|
||||
useSocksProxy SMOnion = case host of
|
||||
THOnionHost _ -> socksProxy'
|
||||
_ -> Nothing
|
||||
{-# INLINE transportClientConfig #-}
|
||||
|
||||
clientSocksCredentials :: NetworkConfig -> ByteString -> Maybe SocksCredentials
|
||||
clientSocksCredentials NetworkConfig {socksProxy} sessionUsername = case socksProxy of
|
||||
clientSocksCredentials :: ProtocolTypeI (ProtoType msg) => NetworkConfig -> UTCTime -> TransportSession msg -> Maybe SocksCredentials
|
||||
clientSocksCredentials NetworkConfig {socksProxy, sessionMode} proxySessTs (userId, srv, entityId_) = case socksProxy of
|
||||
Just (SocksProxyWithAuth auth _) -> case auth of
|
||||
SocksAuthUsername {username, password} -> Just $ SocksCredentials username password
|
||||
SocksAuthNull -> Nothing
|
||||
SocksIsolateByAuth -> Just $ SocksCredentials sessionUsername ""
|
||||
Nothing -> Nothing
|
||||
{-# INLINE clientSocksCredentials #-}
|
||||
where
|
||||
sessionUsername =
|
||||
B64.encode $ C.sha256Hash $
|
||||
bshow userId <> case sessionMode of
|
||||
TSMUser -> ""
|
||||
TSMSession -> ":" <> bshow proxySessTs
|
||||
TSMServer -> ":" <> bshow proxySessTs <> "@" <> strEncode srv
|
||||
TSMEntity -> ":" <> bshow proxySessTs <> "@" <> strEncode srv <> maybe "" ("/" <>) entityId_
|
||||
|
||||
-- | protocol client configuration.
|
||||
data ProtocolClientConfig v = ProtocolClientConfig
|
||||
@@ -470,8 +476,8 @@ type TransportSession msg = (UserId, ProtoServer msg, Maybe ByteString)
|
||||
--
|
||||
-- A single queue can be used for multiple 'SMPClient' instances,
|
||||
-- as 'SMPServerTransmission' includes server information.
|
||||
getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmissionBatch v err msg)) -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
|
||||
getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serverVRange, agreeSecret} msgQ disconnected = do
|
||||
getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> TransportSession msg -> ProtocolClientConfig v -> Maybe (TBQueue (ServerTransmissionBatch v err msg)) -> UTCTime -> (ProtocolClient v err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
|
||||
getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serverVRange, agreeSecret} msgQ proxySessTs disconnected = do
|
||||
case chooseTransportHost networkConfig (host srv) of
|
||||
Right useHost ->
|
||||
(getCurrentTime >>= mkProtocolClient useHost >>= runClient useTransport useHost)
|
||||
@@ -510,7 +516,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
|
||||
runClient (port', ATransport t) useHost c = do
|
||||
cVar <- newEmptyTMVarIO
|
||||
let tcConfig = (transportClientConfig networkConfig useHost) {alpn = clientALPN}
|
||||
socksCreds = clientSocksCredentials networkConfig $ proxyUsername transportSession
|
||||
socksCreds = clientSocksCredentials networkConfig proxySessTs transportSession
|
||||
tId <-
|
||||
runTransportClient tcConfig socksCreds useHost port' (Just $ keyHash srv) (client t c cVar)
|
||||
`forkFinally` \_ -> void (atomically . tryPutTMVar cVar $ Left PCENetworkError)
|
||||
@@ -618,10 +624,6 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
|
||||
unexpectedResponse :: Show r => r -> ProtocolClientError err
|
||||
unexpectedResponse = PCEUnexpectedResponse . B.pack . take 32 . show
|
||||
|
||||
proxyUsername :: TransportSession msg -> ByteString
|
||||
proxyUsername (userId, _, entityId_) = C.sha256Hash $ bshow userId <> maybe "" (":" <>) entityId_
|
||||
{-# INLINE proxyUsername #-}
|
||||
|
||||
-- | Disconnects client from the server and terminates client threads.
|
||||
closeProtocolClient :: ProtocolClient v err msg -> IO ()
|
||||
closeProtocolClient = mapM_ (deRefWeak >=> mapM_ killThread) . action
|
||||
|
||||
@@ -95,6 +95,7 @@ defaultSMPClientAgentConfig =
|
||||
data SMPClientAgent = SMPClientAgent
|
||||
{ agentCfg :: SMPClientAgentConfig,
|
||||
active :: TVar Bool,
|
||||
startedAt :: UTCTime,
|
||||
msgQ :: TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg),
|
||||
agentQ :: TBQueue SMPClientAgentEvent,
|
||||
randomDrg :: TVar ChaChaDRG,
|
||||
@@ -111,6 +112,7 @@ type OwnServer = Bool
|
||||
newSMPClientAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO SMPClientAgent
|
||||
newSMPClientAgent agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} randomDrg = do
|
||||
active <- newTVarIO True
|
||||
startedAt <- getCurrentTime
|
||||
msgQ <- newTBQueueIO msgQSize
|
||||
agentQ <- newTBQueueIO agentQSize
|
||||
smpClients <- TM.emptyIO
|
||||
@@ -123,6 +125,7 @@ newSMPClientAgent agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} randomDrg
|
||||
SMPClientAgent
|
||||
{ agentCfg,
|
||||
active,
|
||||
startedAt,
|
||||
msgQ,
|
||||
agentQ,
|
||||
randomDrg,
|
||||
@@ -194,8 +197,8 @@ isOwnServer SMPClientAgent {agentCfg} ProtocolServer {host} =
|
||||
|
||||
-- | Run an SMP client for SMPClientVar
|
||||
connectClient :: SMPClientAgent -> SMPServer -> SMPClientVar -> IO (Either SMPClientError SMPClient)
|
||||
connectClient ca@SMPClientAgent {agentCfg, smpClients, smpSessions, msgQ, randomDrg} srv v =
|
||||
getProtocolClient randomDrg (1, srv, Nothing) (smpCfg agentCfg) (Just msgQ) clientDisconnected
|
||||
connectClient ca@SMPClientAgent {agentCfg, smpClients, smpSessions, msgQ, randomDrg, startedAt} srv v =
|
||||
getProtocolClient randomDrg (1, srv, Nothing) (smpCfg agentCfg) (Just msgQ) startedAt clientDisconnected
|
||||
where
|
||||
clientDisconnected :: SMPClient -> IO ()
|
||||
clientDisconnected smp = do
|
||||
|
||||
@@ -1278,7 +1278,7 @@ transmissionP THandleParams {sessionId, implySessId} = do
|
||||
command <- A.takeByteString
|
||||
pure RawTransmission {authenticator, authorized = authorized', sessId, corrId, entityId, command}
|
||||
|
||||
class (ProtocolEncoding v err msg, ProtocolEncoding v err (ProtoCommand msg), Show err, Show msg) => Protocol v err msg | msg -> v, msg -> err where
|
||||
class (ProtocolTypeI (ProtoType msg), ProtocolEncoding v err msg, ProtocolEncoding v err (ProtoCommand msg), Show err, Show msg) => Protocol v err msg | msg -> v, msg -> err where
|
||||
type ProtoCommand msg = cmd | cmd -> msg
|
||||
type ProtoType msg = (sch :: ProtocolType) | sch -> msg
|
||||
protocolClientHandshake :: forall c. Transport c => c -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRange v -> ExceptT TransportError IO (THandle v c 'TClient)
|
||||
|
||||
@@ -85,7 +85,7 @@ import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ, SENT)
|
||||
import qualified Simplex.Messaging.Agent.Protocol as A
|
||||
import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
|
||||
import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), SMPProxyFallback (..), SMPProxyMode (..), TransportSessionMode (TSMEntity, TSMUser), defaultClientConfig)
|
||||
import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), SMPProxyFallback (..), SMPProxyMode (..), TransportSessionMode (..), defaultClientConfig)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOff, pattern IKPQOn, pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
|
||||
import qualified Simplex.Messaging.Crypto.Ratchet as CR
|
||||
@@ -3014,7 +3014,7 @@ testDeliveryReceiptsConcurrent t =
|
||||
testTwoUsers :: HasCallStack => IO ()
|
||||
testTwoUsers = withAgentClients2 $ \a b -> do
|
||||
let nc = netCfg initAgentServers
|
||||
sessionMode nc `shouldBe` TSMUser
|
||||
sessionMode nc `shouldBe` TSMSession
|
||||
runRight_ $ do
|
||||
(aId1, bId1) <- makeConnectionForUsers a 1 b 1
|
||||
exchangeGreetings a bId1 b aId1
|
||||
|
||||
@@ -22,6 +22,7 @@ import Control.Monad.Trans.Except (ExceptT, runExceptT)
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import Data.List.NonEmpty (NonEmpty)
|
||||
import qualified Data.List.NonEmpty as L
|
||||
import Data.Time.Clock (getCurrentTime)
|
||||
import SMPAgentClient
|
||||
import SMPClient
|
||||
import ServerTests (decryptMsgV3, sendRecv)
|
||||
@@ -150,12 +151,13 @@ deliverMessagesViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => SMPServer -> S
|
||||
deliverMessagesViaProxy proxyServ relayServ alg unsecuredMsgs securedMsgs = do
|
||||
g <- C.newRandom
|
||||
-- set up proxy
|
||||
pc' <- getProtocolClient g (1, proxyServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion} Nothing (\_ -> pure ())
|
||||
ts <- getCurrentTime
|
||||
pc' <- getProtocolClient g (1, proxyServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion} Nothing ts (\_ -> pure ())
|
||||
pc <- either (fail . show) pure pc'
|
||||
THAuthClient {} <- maybe (fail "getProtocolClient returned no thAuth") pure $ thAuth $ thParams pc
|
||||
-- set up relay
|
||||
msgQ <- newTBQueueIO 1024
|
||||
rc' <- getProtocolClient g (2, relayServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion} (Just msgQ) (\_ -> pure ())
|
||||
rc' <- getProtocolClient g (2, relayServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion} (Just msgQ) ts (\_ -> pure ())
|
||||
rc <- either (fail . show) pure rc'
|
||||
-- prepare receiving queue
|
||||
(rPub, rPriv) <- atomically $ C.generateAuthKeyPair alg g
|
||||
@@ -192,7 +194,8 @@ proxyConnectDeadRelay :: Int -> Int -> SMPServer -> IO ()
|
||||
proxyConnectDeadRelay n d proxyServ = do
|
||||
g <- C.newRandom
|
||||
-- set up proxy
|
||||
pc' <- getProtocolClient g (1, proxyServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion} Nothing (\_ -> pure ())
|
||||
ts <- getCurrentTime
|
||||
pc' <- getProtocolClient g (1, proxyServ, Nothing) defaultSMPClientConfig {serverVRange = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion} Nothing ts (\_ -> pure ())
|
||||
pc <- either (fail . show) pure pc'
|
||||
THAuthClient {} <- maybe (fail "getProtocolClient returned no thAuth") pure $ thAuth $ thParams pc
|
||||
-- get proxy session
|
||||
|
||||
+4
-2
@@ -8,6 +8,7 @@ module XFTPClient where
|
||||
|
||||
import Control.Concurrent (ThreadId, threadDelay)
|
||||
import Data.String (fromString)
|
||||
import Data.Time.Clock (getCurrentTime)
|
||||
import Network.Socket (ServiceName)
|
||||
import SMPClient (serverBracket)
|
||||
import Simplex.FileTransfer.Client
|
||||
@@ -135,7 +136,8 @@ testXFTPClient :: HasCallStack => (HasCallStack => XFTPClient -> IO a) -> IO a
|
||||
testXFTPClient = testXFTPClientWith testXFTPClientConfig
|
||||
|
||||
testXFTPClientWith :: HasCallStack => XFTPClientConfig -> (HasCallStack => XFTPClient -> IO a) -> IO a
|
||||
testXFTPClientWith cfg client =
|
||||
getXFTPClient (1, testXFTPServer, Nothing) cfg (\_ -> pure ()) >>= \case
|
||||
testXFTPClientWith cfg client = do
|
||||
ts <- getCurrentTime
|
||||
getXFTPClient (1, testXFTPServer, Nothing) cfg ts (\_ -> pure ()) >>= \case
|
||||
Right c -> client c
|
||||
Left e -> error $ show e
|
||||
|
||||
@@ -18,6 +18,7 @@ import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import qualified Data.ByteString.Lazy.Char8 as LB
|
||||
import Data.List (isInfixOf)
|
||||
import Data.Time.Clock (getCurrentTime)
|
||||
import ServerTests (logSize)
|
||||
import Simplex.FileTransfer.Client
|
||||
import Simplex.FileTransfer.Description (kb)
|
||||
@@ -220,7 +221,8 @@ testFileChunkExpiration = withXFTPServerCfg testXFTPServerConfig {fileExpiration
|
||||
testInactiveClientExpiration :: Expectation
|
||||
testInactiveClientExpiration = withXFTPServerCfg testXFTPServerConfig {inactiveClientExpiration} $ \_ -> runRight_ $ do
|
||||
disconnected <- newEmptyTMVarIO
|
||||
c <- ExceptT $ getXFTPClient (1, testXFTPServer, Nothing) testXFTPClientConfig (\_ -> atomically $ putTMVar disconnected ())
|
||||
ts <- liftIO getCurrentTime
|
||||
c <- ExceptT $ getXFTPClient (1, testXFTPServer, Nothing) testXFTPClientConfig ts (\_ -> atomically $ putTMVar disconnected ())
|
||||
pingXFTP c
|
||||
liftIO $ do
|
||||
threadDelay 100000
|
||||
|
||||
Reference in New Issue
Block a user