mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 15:12:26 +00:00
xftp: negotiate protocol with ALPN (#1047)
* 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>
This commit is contained in:
1 parent
791368c7be
commit
4c20ff6d00
24 files changed
+533
-114
No files matched your search
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
{-# LANGUAGE OverloadedLists #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE TupleSections #-}
|
||||
|
||||
module Simplex.FileTransfer.Server where
|
||||
|
||||
@@ -18,7 +19,7 @@ import Control.Monad.Except
|
||||
import Control.Monad.Reader
|
||||
import Data.Bifunctor (first)
|
||||
import qualified Data.ByteString.Base64.URL as B64
|
||||
import Data.ByteString.Builder (byteString)
|
||||
import Data.ByteString.Builder (Builder, byteString)
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Int (Int64)
|
||||
@@ -32,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
|
||||
@@ -46,18 +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 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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -281,7 +342,7 @@ verifyXFTPTransmission tAuth authorized fId cmd =
|
||||
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
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -167,7 +167,7 @@ import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientCo
|
||||
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
|
||||
@@ -196,6 +196,7 @@ import Simplex.Messaging.Protocol
|
||||
ErrorType,
|
||||
MsgFlags (..),
|
||||
MsgId,
|
||||
NtfPublicAuthKey,
|
||||
NtfServer,
|
||||
NtfServerWithAuth,
|
||||
ProtoServer,
|
||||
@@ -207,22 +208,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
|
||||
@@ -532,7 +532,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
|
||||
@@ -635,7 +636,8 @@ 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
|
||||
@@ -655,9 +657,11 @@ 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
|
||||
@@ -952,7 +956,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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -17,6 +17,7 @@ module Simplex.Messaging.Transport.Client
|
||||
TransportHost (..),
|
||||
TransportHosts (..),
|
||||
TransportHosts_ (..),
|
||||
validateCertificateChain
|
||||
)
|
||||
where
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
}
|
||||
@@ -237,7 +240,7 @@ mkTLSClientParams supported caStore_ host port cafp_ clientCreds_ serverCerts =
|
||||
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]
|
||||
@@ -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
|
||||
|
||||
@@ -23,15 +23,19 @@ 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
|
||||
|
||||
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 +70,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 +90,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
|
||||
@@ -109,10 +114,20 @@ getVerifiedHTTP2ClientWith config host port disconnected setup =
|
||||
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}
|
||||
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')
|
||||
@@ -154,13 +169,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
|
||||
@@ -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 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}
|
||||
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
|
||||
|
||||
Reference in new issue
Block a user