mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 08:42:27 +00:00
use throwE instead of throwError (#1187)
* use throwE instead of throwError * test delay
This commit is contained in:
1 parent
44b8f265ae
commit
3dab330480
18 files changed
+155
-145
No files matched your search
@@ -32,6 +32,7 @@ import Control.Logger.Simple (logError)
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Reader
|
||||
import Control.Monad.Trans.Except
|
||||
import Data.Bifunctor (first)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import qualified Data.ByteString.Lazy.Char8 as LB
|
||||
@@ -141,7 +142,7 @@ xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redi
|
||||
downloadChunk :: AgentClient -> FileChunk -> AM ()
|
||||
downloadChunk c FileChunk {replicas = (FileChunkReplica {server} : _)} = do
|
||||
lift . void $ getXFTPRcvWorker True c (Just server)
|
||||
downloadChunk _ _ = throwError $ INTERNAL "no replicas"
|
||||
downloadChunk _ _ = throwE $ INTERNAL "no replicas"
|
||||
|
||||
getPrefixPath :: String -> AM' FilePath
|
||||
getPrefixPath suffix = do
|
||||
@@ -194,7 +195,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
|
||||
retryDone = rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath)
|
||||
downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> Bool -> AM ()
|
||||
downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvFileEntityId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica approvedRelays = do
|
||||
unlessM ((approvedRelays ||) <$> ipAddressProtected') $ throwError $ XFTP "" XFTP.NOT_APPROVED
|
||||
unlessM ((approvedRelays ||) <$> ipAddressProtected') $ throwE $ XFTP "" XFTP.NOT_APPROVED
|
||||
fsFileTmpPath <- lift $ toFSFilePath fileTmpPath
|
||||
chunkPath <- uniqueCombine fsFileTmpPath $ show chunkNo
|
||||
let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
|
||||
@@ -267,9 +268,9 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
|
||||
withStore' c $ \db -> updateRcvFileStatus db rcvFileId RFSDecrypting
|
||||
chunkPaths <- getChunkPaths chunks
|
||||
encSize <- liftIO $ foldM (\s path -> (s +) . fromIntegral <$> getFileSize path) 0 chunkPaths
|
||||
when (FileSize encSize /= size) $ throwError $ XFTP "" XFTP.SIZE
|
||||
when (FileSize encSize /= size) $ throwE $ XFTP "" XFTP.SIZE
|
||||
encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
|
||||
when (FileDigest encDigest /= digest) $ throwError $ XFTP "" XFTP.DIGEST
|
||||
when (FileDigest encDigest /= digest) $ throwE $ XFTP "" XFTP.DIGEST
|
||||
let destFile = CryptoFile fsSavePath cfArgs
|
||||
void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure destFile
|
||||
case redirect of
|
||||
@@ -287,10 +288,10 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
|
||||
yaml <- liftError (INTERNAL . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `agentFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
|
||||
next@FileDescription {chunks = nextChunks} <- case strDecode (LB.toStrict yaml) of
|
||||
-- TODO switch to another error constructor
|
||||
Left _ -> throwError . XFTP "" $ XFTP.REDIRECT "decode error"
|
||||
Left _ -> throwE . XFTP "" $ XFTP.REDIRECT "decode error"
|
||||
Right (ValidFileDescription fd@FileDescription {size = dstSize, digest = dstDigest})
|
||||
| dstSize /= redirectSize -> throwError . XFTP "" $ XFTP.REDIRECT "size mismatch"
|
||||
| dstDigest /= redirectDigest -> throwError . XFTP "" $ XFTP.REDIRECT "digest mismatch"
|
||||
| dstSize /= redirectSize -> throwE . XFTP "" $ XFTP.REDIRECT "size mismatch"
|
||||
| dstDigest /= redirectDigest -> throwE . XFTP "" $ XFTP.REDIRECT "digest mismatch"
|
||||
| otherwise -> pure fd
|
||||
-- register and download chunks from the actual file
|
||||
withStore c $ \db -> updateRcvFileRedirect db redirectDbId next
|
||||
@@ -303,7 +304,7 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
|
||||
fsPath <- lift $ toFSFilePath path
|
||||
pure $ fsPath : ps
|
||||
getChunkPaths (RcvFileChunk {chunkTmpPath = Nothing} : _cs) =
|
||||
throwError $ INTERNAL "no chunk path"
|
||||
throwE $ INTERNAL "no chunk path"
|
||||
|
||||
xftpDeleteRcvFile' :: AgentClient -> RcvFileId -> AM' ()
|
||||
xftpDeleteRcvFile' c rcvFileEntityId = xftpDeleteRcvFiles' c [rcvFileEntityId]
|
||||
@@ -379,7 +380,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
|
||||
prepareFile cfg f `catchAgentError` (sndWorkerInternalError c sndFileId sndFileEntityId prefixPath . show)
|
||||
prepareFile :: AgentConfig -> SndFile -> AM ()
|
||||
prepareFile _ SndFile {prefixPath = Nothing} =
|
||||
throwError $ INTERNAL "no prefix path"
|
||||
throwE $ INTERNAL "no prefix path"
|
||||
prepareFile cfg sndFile@SndFile {sndFileId, userId, prefixPath = Just ppath, status} = do
|
||||
SndFile {numRecipients, chunks} <-
|
||||
if status /= SFSEncrypted -- status is SFSNew or SFSEncrypting
|
||||
@@ -405,14 +406,14 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
|
||||
let CryptoFile {filePath} = srcFile
|
||||
fileName = takeFileName filePath
|
||||
fileSize <- liftIO $ fromInteger <$> CF.getFileContentsSize srcFile
|
||||
when (fileSize > maxFileSizeHard) $ throwError $ INTERNAL "max file size exceeded"
|
||||
when (fileSize > maxFileSizeHard) $ throwE $ INTERNAL "max file size exceeded"
|
||||
let fileHdr = smpEncode FileHeader {fileName, fileExtra = Nothing}
|
||||
fileSize' = fromIntegral (B.length fileHdr) + fileSize
|
||||
payloadSize = fileSize' + fileSizeLen + authTagSize
|
||||
chunkSizes <- case redirect of
|
||||
Nothing -> pure $ prepareChunkSizes payloadSize
|
||||
Just _ -> case singleChunkSize payloadSize of
|
||||
Nothing -> throwError $ INTERNAL "max file size exceeded for redirect"
|
||||
Nothing -> throwE $ INTERNAL "max file size exceeded for redirect"
|
||||
Just chunkSize -> pure [chunkSize]
|
||||
let encSize = sum $ map fromIntegral chunkSizes
|
||||
void $ liftError (INTERNAL . show) $ encryptFile srcFile fileHdr key nonce fileSize' encSize fsEncPath
|
||||
@@ -432,12 +433,12 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
|
||||
withRetryInterval (riFast ri) $ \_ loop -> do
|
||||
liftIO $ waitForUserNetwork c
|
||||
createWithNextSrv usedSrvs
|
||||
`catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
|
||||
`catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwE e) e
|
||||
where
|
||||
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"
|
||||
when deleted $ throwE $ INTERNAL "file deleted, aborting chunk creation"
|
||||
withNextSrv c userId usedSrvs [] $ \srvAuth -> do
|
||||
replica <- agentXFTPNewChunk c ch numRecipients' srvAuth
|
||||
pure (replica, srvAuth)
|
||||
@@ -479,7 +480,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
|
||||
uploadFileChunk AgentConfig {xftpMaxRecipientsPerRequest = maxRecipients} sndFileChunk@SndFileChunk {sndFileId, userId, chunkSpec = chunkSpec@XFTPChunkSpec {filePath}, digest = chunkDigest} replica = do
|
||||
replica'@SndFileChunkReplica {sndChunkReplicaId} <- addRecipients sndFileChunk replica
|
||||
fsFilePath <- lift $ toFSFilePath filePath
|
||||
unlessM (doesFileExist fsFilePath) $ throwError $ INTERNAL "encrypted file doesn't exist on upload"
|
||||
unlessM (doesFileExist fsFilePath) $ throwE $ INTERNAL "encrypted file doesn't exist on upload"
|
||||
let chunkSpec' = chunkSpec {filePath = fsFilePath} :: XFTPChunkSpec
|
||||
atomically $ assertAgentForeground c
|
||||
agentXFTPUploadChunk c userId chunkDigest replica' chunkSpec'
|
||||
@@ -499,7 +500,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
|
||||
where
|
||||
addRecipients :: SndFileChunk -> SndFileChunkReplica -> AM SndFileChunkReplica
|
||||
addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {rcvIdsKeys}
|
||||
| length rcvIdsKeys > numRecipients = throwError $ INTERNAL "too many recipients"
|
||||
| length rcvIdsKeys > numRecipients = throwE $ INTERNAL "too many recipients"
|
||||
| length rcvIdsKeys == numRecipients = pure cr
|
||||
| otherwise = do
|
||||
let numRecipients' = min (numRecipients - length rcvIdsKeys) maxRecipients
|
||||
@@ -507,22 +508,22 @@ runXFTPSndWorker c srv Worker {doWork} = do
|
||||
cr' <- withStore' c $ \db -> addSndChunkReplicaRecipients db cr $ L.toList rcvIdsKeys'
|
||||
addRecipients ch cr'
|
||||
sndFileToDescrs :: SndFile -> AM (ValidFileDescription 'FSender, [ValidFileDescription 'FRecipient])
|
||||
sndFileToDescrs SndFile {digest = Nothing} = throwError $ INTERNAL "snd file has no digest"
|
||||
sndFileToDescrs SndFile {chunks = []} = throwError $ INTERNAL "snd file has no chunks"
|
||||
sndFileToDescrs SndFile {digest = Nothing} = throwE $ INTERNAL "snd file has no digest"
|
||||
sndFileToDescrs SndFile {chunks = []} = throwE $ INTERNAL "snd file has no chunks"
|
||||
sndFileToDescrs SndFile {digest = Just digest, key, nonce, chunks = chunks@(fstChunk : _), redirect} = do
|
||||
let chunkSize = FileSize $ sndChunkSize fstChunk
|
||||
size = FileSize $ sum $ map (fromIntegral . sndChunkSize) chunks
|
||||
-- snd description
|
||||
sndDescrChunks <- mapM toSndDescrChunk chunks
|
||||
let fdSnd = FileDescription {party = SFSender, size, digest, key, nonce, chunkSize, chunks = sndDescrChunks, redirect = Nothing}
|
||||
validFdSnd <- either (throwError . INTERNAL) pure $ validateFileDescription fdSnd
|
||||
validFdSnd <- either (throwE . INTERNAL) pure $ validateFileDescription fdSnd
|
||||
-- rcv descriptions
|
||||
let fdRcv = FileDescription {party = SFRecipient, size, digest, key, nonce, chunkSize, chunks = [], redirect}
|
||||
fdRcvs = createRcvFileDescriptions fdRcv chunks
|
||||
validFdRcvs <- either (throwError . INTERNAL) pure $ mapM validateFileDescription fdRcvs
|
||||
validFdRcvs <- either (throwE . INTERNAL) pure $ mapM validateFileDescription fdRcvs
|
||||
pure (validFdSnd, validFdRcvs)
|
||||
toSndDescrChunk :: SndFileChunk -> AM FileChunk
|
||||
toSndDescrChunk SndFileChunk {replicas = []} = throwError $ INTERNAL "snd file chunk has no replicas"
|
||||
toSndDescrChunk SndFileChunk {replicas = []} = throwE $ INTERNAL "snd file chunk has no replicas"
|
||||
toSndDescrChunk ch@SndFileChunk {chunkNo, digest = chDigest, replicas = (SndFileChunkReplica {server, replicaId, replicaKey} : _)} = do
|
||||
let chunkSize = FileSize $ sndChunkSize ch
|
||||
replicas = [FileChunkReplica {server, replicaId, replicaKey}]
|
||||
|
||||
@@ -138,9 +138,9 @@ xftpClientHandshakeV1 serverVRange keyHash@(C.KeyHash kh) c@HTTP2Client {session
|
||||
liftTransportErr (TEHandshake PARSE) . smpDecode =<< liftTransportErr TEBadBlock (C.unPad shsBody)
|
||||
processServerHandshake :: XFTPServerHandshake -> ExceptT XFTPClientError IO (VersionRangeXFTP, C.PublicKeyX25519)
|
||||
processServerHandshake XFTPServerHandshake {xftpVersionRange, sessionId = serverSessId, authPubKey = serverAuth} = do
|
||||
unless (sessionId == serverSessId) $ throwError $ PCETransportError TEBadSession
|
||||
unless (sessionId == serverSessId) $ throwE $ PCETransportError TEBadSession
|
||||
case xftpVersionRange `compatibleVRange` serverVRange of
|
||||
Nothing -> throwError $ PCETransportError TEVersion
|
||||
Nothing -> throwE $ PCETransportError TEVersion
|
||||
Just (Compatible vr) ->
|
||||
fmap (vr,) . liftTransportErr (TEHandshake BAD_AUTH) $ do
|
||||
let (X.CertificateChain cert, exact) = serverAuth
|
||||
@@ -154,7 +154,7 @@ xftpClientHandshakeV1 serverVRange keyHash@(C.KeyHash kh) c@HTTP2Client {session
|
||||
chs' <- liftTransportErr TELargeMsg $ C.pad (smpEncode chs) xftpBlockSize
|
||||
let chsReq = H.requestBuilder "POST" "/" [] $ byteString chs'
|
||||
HTTP2Response {respBody = HTTP2Body {bodyHead}} <- liftError' xftpClientError $ sendRequest c chsReq Nothing
|
||||
unless (B.null bodyHead) $ throwError $ PCETransportError TEBadBlock
|
||||
unless (B.null bodyHead) $ throwE $ PCETransportError TEBadBlock
|
||||
liftTransportErr e = liftEitherWith (const $ PCETransportError e)
|
||||
|
||||
closeXFTPClient :: XFTPClient -> IO ()
|
||||
@@ -200,14 +200,14 @@ sendXFTPTransmission XFTPClient {config, thParams, http2Client} t chunkSpec_ = d
|
||||
let req = H.requestStreaming N.methodPost "/" [] streamBody
|
||||
reqTimeout = xftpReqTimeout config $ (\XFTPChunkSpec {chunkSize} -> chunkSize) <$> chunkSpec_
|
||||
HTTP2Response {respBody = body@HTTP2Body {bodyHead}} <- withExceptT xftpClientError . ExceptT $ sendRequest http2Client req (Just reqTimeout)
|
||||
when (B.length bodyHead /= xftpBlockSize) $ throwError $ PCEResponseError BLOCK
|
||||
when (B.length bodyHead /= xftpBlockSize) $ throwE $ PCEResponseError BLOCK
|
||||
-- TODO validate that the file ID is the same as in the request?
|
||||
(_, _, (_, _fId, respOrErr)) <- liftEither . first PCEResponseError $ xftpDecodeTransmission thParams bodyHead
|
||||
case respOrErr of
|
||||
Right r -> case protocolError r of
|
||||
Just e -> throwError $ PCEProtocolError e
|
||||
Just e -> throwE $ PCEProtocolError e
|
||||
_ -> pure (r, body)
|
||||
Left e -> throwError $ PCEResponseError e
|
||||
Left e -> throwE $ PCEResponseError e
|
||||
where
|
||||
streamBody :: (Builder -> IO ()) -> IO () -> IO ()
|
||||
streamBody send done = do
|
||||
@@ -250,7 +250,7 @@ downloadXFTPChunk g c@XFTPClient {config} rpKey fId chunkSpec@XFTPRcvChunkSpec {
|
||||
let dhSecret = C.dh' sDhKey rpDhKey
|
||||
cbState <- liftEither . first PCECryptoError $ LC.cbInit dhSecret cbNonce
|
||||
let t = chunkTimeout config chunkSize
|
||||
ExceptT (sequence <$> (t `timeout` (download cbState `catches` errors))) >>= maybe (throwError PCEResponseTimeout) pure
|
||||
ExceptT (sequence <$> (t `timeout` (download cbState `catches` errors))) >>= maybe (throwE PCEResponseTimeout) pure
|
||||
where
|
||||
errors =
|
||||
[ Handler $ \(_e :: H.HTTP2Error) -> pure $ Left PCENetworkError,
|
||||
@@ -260,8 +260,8 @@ downloadXFTPChunk g c@XFTPClient {config} rpKey fId chunkSpec@XFTPRcvChunkSpec {
|
||||
download cbState =
|
||||
runExceptT . withExceptT PCEResponseError $
|
||||
receiveEncFile chunkPart cbState chunkSpec `catchError` \e ->
|
||||
whenM (doesFileExist filePath) (removeFile filePath) >> throwError e
|
||||
_ -> throwError $ PCEResponseError NO_FILE
|
||||
whenM (doesFileExist filePath) (removeFile filePath) >> throwE e
|
||||
_ -> throwE $ PCEResponseError NO_FILE
|
||||
(r, _) -> throwE $ unexpectedResponse r
|
||||
|
||||
xftpReqTimeout :: XFTPClientConfig -> Maybe Word32 -> Int
|
||||
@@ -296,7 +296,7 @@ okResponse = \case
|
||||
-- TODO this currently does not check anything because response size is not set and bodyPart is always Just
|
||||
noFile :: HTTP2Body -> a -> ExceptT XFTPClientError IO a
|
||||
noFile HTTP2Body {bodyPart} a = case bodyPart of
|
||||
Just _ -> pure a -- throwError $ PCEResponseError HAS_FILE
|
||||
Just _ -> pure a -- throwE $ PCEResponseError HAS_FILE
|
||||
_ -> pure a
|
||||
|
||||
-- FACK :: FileCommand Recipient
|
||||
|
||||
@@ -11,6 +11,7 @@ import Control.Logger.Simple (logInfo)
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Trans (lift)
|
||||
import Control.Monad.Trans.Except
|
||||
import Data.Bifunctor (first)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Text (Text)
|
||||
@@ -108,7 +109,7 @@ getXFTPServerClient XFTPClientAgent {xftpClients, config} srv = do
|
||||
else atomically $ do
|
||||
putTMVar clientVar r
|
||||
TM.delete srv xftpClients
|
||||
throwError e
|
||||
throwE e
|
||||
tryConnectAsync :: ME ()
|
||||
tryConnectAsync = void . lift . async . runExceptT $ do
|
||||
withRetryInterval (reconnectInterval config) $ \_ loop -> void $ tryConnectClient loop
|
||||
|
||||
@@ -30,6 +30,7 @@ where
|
||||
import Control.Logger.Simple
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Trans.Except
|
||||
import Crypto.Random (ChaChaDRG)
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import Data.Bifunctor (first)
|
||||
@@ -292,7 +293,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
|
||||
encryptFileForUpload :: TVar ChaChaDRG -> String -> ExceptT CLIError IO (FilePath, FileDescription 'FRecipient, FileDescription 'FSender, [XFTPChunkSpec], Int64)
|
||||
encryptFileForUpload g fileName = do
|
||||
fileSize <- fromInteger <$> getFileSize filePath
|
||||
when (fileSize > maxFileSize) $ throwError $ CLIError $ "Files bigger than " <> maxFileSizeStr <> " are not supported"
|
||||
when (fileSize > maxFileSize) $ throwE $ CLIError $ "Files bigger than " <> maxFileSizeStr <> " are not supported"
|
||||
encPath <- getEncPath tempPath "xftp"
|
||||
key <- atomically $ C.randomSbKey g
|
||||
nonce <- atomically $ C.randomCbNonce g
|
||||
@@ -323,7 +324,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
|
||||
-- upload doesn't allow other requests within the same client until complete (but download does allow).
|
||||
logInfo $ "uploading " <> tshow (length chunks) <> " chunks..."
|
||||
(errs, rs) <- partitionEithers . concat <$> liftIO (pooledForConcurrentlyN 16 chunks' . mapM $ runExceptT . uploadFileChunk a)
|
||||
mapM_ throwError errs
|
||||
mapM_ throwE errs
|
||||
pure $ map snd (sortOn fst rs)
|
||||
where
|
||||
uploadFileChunk :: XFTPClientAgent -> (Int, XFTPChunkSpec, XFTPServerWithAuth) -> ExceptT CLIError IO (Int, SentFileChunk)
|
||||
@@ -437,12 +438,12 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
|
||||
srvChunks = groupAllOn srv chunks
|
||||
g <- liftIO C.newRandom
|
||||
(errs, rs) <- partitionEithers . concat <$> liftIO (pooledForConcurrentlyN 16 srvChunks $ mapM $ runExceptT . downloadFileChunk g a encPath size downloadedChunks)
|
||||
mapM_ throwError errs
|
||||
mapM_ throwE errs
|
||||
let chunkPaths = map snd $ sortOn fst rs
|
||||
encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
|
||||
when (encDigest /= unFileDigest digest) $ throwError $ CLIError "File digest mismatch"
|
||||
when (encDigest /= unFileDigest digest) $ throwE $ CLIError "File digest mismatch"
|
||||
encSize <- liftIO $ foldM (\s path -> (s +) . fromIntegral <$> getFileSize path) 0 chunkPaths
|
||||
when (FileSize encSize /= size) $ throwError $ CLIError "File size mismatch"
|
||||
when (FileSize encSize /= size) $ throwE $ 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
|
||||
@@ -464,20 +465,20 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
|
||||
printProgress "Downloaded" downloaded encSize
|
||||
when verbose $ putStrLn ""
|
||||
pure (chunkNo, chunkPath)
|
||||
downloadFileChunk _ _ _ _ _ _ = throwError $ CLIError "chunk has no replicas"
|
||||
downloadFileChunk _ _ _ _ _ _ = throwE $ CLIError "chunk has no replicas"
|
||||
getFilePath :: String -> ExceptT String IO FilePath
|
||||
getFilePath name =
|
||||
case filePath of
|
||||
Just path ->
|
||||
ifM (doesDirectoryExist path) (uniqueCombine path name) $
|
||||
ifM (doesFileExist path) (throwError "File already exists") (pure path)
|
||||
ifM (doesFileExist path) (throwE "File already exists") (pure path)
|
||||
_ -> (`uniqueCombine` name) . (</> "Downloads") =<< getHomeDirectory
|
||||
acknowledgeFileChunk :: XFTPClientAgent -> FileChunk -> ExceptT CLIError IO ()
|
||||
acknowledgeFileChunk a FileChunk {replicas = replica : _} = do
|
||||
let FileChunkReplica {server, replicaId, replicaKey} = replica
|
||||
c <- withRetry retryCount $ getXFTPServerClient a server
|
||||
withRetry retryCount $ ackXFTPChunk c replicaKey (unChunkReplicaId replicaId)
|
||||
acknowledgeFileChunk _ _ = throwError $ CLIError "chunk has no replicas"
|
||||
acknowledgeFileChunk _ _ = throwE $ CLIError "chunk has no replicas"
|
||||
|
||||
printProgress :: String -> Int64 -> Int64 -> IO ()
|
||||
printProgress s part total = printNoNewLine $ s <> " " <> show ((part * 100) `div` total) <> "%"
|
||||
@@ -503,7 +504,7 @@ cliDeleteFile DeleteOptions {fileDescription, retryCount, yes} = do
|
||||
let FileChunkReplica {server, replicaId, replicaKey} = replica
|
||||
withReconnect a server retryCount $ \c -> deleteXFTPChunk c replicaKey (unChunkReplicaId replicaId)
|
||||
logInfo $ "deleted chunk " <> tshow chunkNo <> " from " <> showServer server
|
||||
deleteFileChunk _ _ = throwError $ CLIError "chunk has no replicas"
|
||||
deleteFileChunk _ _ = throwE $ CLIError "chunk has no replicas"
|
||||
|
||||
cliFileDescrInfo :: InfoOptions -> ExceptT CLIError IO ()
|
||||
cliFileDescrInfo InfoOptions {fileDescription} = do
|
||||
@@ -533,7 +534,7 @@ getFileDescription path =
|
||||
getFileDescription' :: FilePartyI p => FilePath -> ExceptT CLIError IO (ValidFileDescription p)
|
||||
getFileDescription' path =
|
||||
getFileDescription path >>= \case
|
||||
AVFD fd -> either (throwError . CLIError) pure $ checkParty fd
|
||||
AVFD fd -> either (throwE . CLIError) pure $ checkParty fd
|
||||
|
||||
singleChunkSize :: Int64 -> Maybe Word32
|
||||
singleChunkSize size' =
|
||||
@@ -574,13 +575,13 @@ withReconnect a srv n run = withRetry n $ do
|
||||
c <- withRetry n $ getXFTPServerClient a srv
|
||||
withExceptT (CLIError . show) (run c) `catchError` \e -> do
|
||||
liftIO $ closeXFTPServerClient a srv
|
||||
throwError e
|
||||
throwE e
|
||||
|
||||
withRetry :: Show e => Int -> ExceptT e IO a -> ExceptT CLIError IO a
|
||||
withRetry retryCount = withRetry' retryCount . withExceptT (CLIError . show)
|
||||
where
|
||||
withRetry' :: Int -> ExceptT CLIError IO a -> ExceptT CLIError IO a
|
||||
withRetry' 0 _ = throwError $ CLIError "internal: no retry attempts"
|
||||
withRetry' 0 _ = throwE $ CLIError "internal: no retry attempts"
|
||||
withRetry' 1 a = a
|
||||
withRetry' n a =
|
||||
a `catchError` \e -> do
|
||||
|
||||
@@ -8,6 +8,7 @@ module Simplex.FileTransfer.Crypto where
|
||||
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Trans.Except
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import Data.Bifunctor (first)
|
||||
import qualified Data.ByteArray as BA
|
||||
@@ -48,17 +49,17 @@ encryptFile srcFile fileHdr key nonce fileSize' encSize encFile = do
|
||||
| otherwise = do
|
||||
let chSize = min len 65536
|
||||
ch <- liftIO $ get chSize
|
||||
when (B.length ch /= fromIntegral chSize) $ throwError $ FTCEFileIOError "encrypting file: unexpected EOF"
|
||||
when (B.length ch /= fromIntegral chSize) $ throwE $ FTCEFileIOError "encrypting file: unexpected EOF"
|
||||
let (ch', sb') = LC.sbEncryptChunk sb ch
|
||||
liftIO $ B.hPut w ch'
|
||||
encryptChunks_ get w (sb', len - chSize)
|
||||
|
||||
decryptChunks :: Int64 -> [FilePath] -> C.SbKey -> C.CbNonce -> (String -> ExceptT String IO CryptoFile) -> ExceptT FTCryptoError IO CryptoFile
|
||||
decryptChunks _ [] _ _ _ = throwError $ FTCEInvalidHeader "empty"
|
||||
decryptChunks _ [] _ _ _ = throwE $ FTCEInvalidHeader "empty"
|
||||
decryptChunks encSize (chPath : chPaths) key nonce getDestFile = case reverse chPaths of
|
||||
[] -> do
|
||||
(!authOk, !f) <- liftEither . first FTCECryptoError . LC.sbDecryptTailTag key nonce (encSize - authTagSize) =<< liftIO (LB.readFile chPath)
|
||||
unless authOk $ throwError FTCEInvalidAuthTag
|
||||
unless authOk $ throwE FTCEInvalidAuthTag
|
||||
(FileHeader {fileName}, !f') <- parseFileHeader f
|
||||
destFile <- withExceptT FTCEFileIOError $ getDestFile fileName
|
||||
CF.writeFile destFile f'
|
||||
@@ -73,7 +74,7 @@ decryptChunks encSize (chPath : chPaths) key nonce getDestFile = case reverse ch
|
||||
decryptLastChunk h state' expectedLen
|
||||
unless authOk $ do
|
||||
removeFile path
|
||||
throwError FTCEInvalidAuthTag
|
||||
throwE FTCEInvalidAuthTag
|
||||
pure destFile
|
||||
where
|
||||
decryptFirstChunk = do
|
||||
@@ -105,8 +106,8 @@ decryptChunks encSize (chPath : chPaths) key nonce getDestFile = case reverse ch
|
||||
parseFileHeader s = do
|
||||
let (hdrStr, s') = LB.splitAt 1024 s
|
||||
case A.parse smpP $ LB.toStrict hdrStr of
|
||||
A.Fail _ _ e -> throwError $ FTCEInvalidHeader e
|
||||
A.Partial _ -> throwError $ FTCEInvalidHeader "incomplete"
|
||||
A.Fail _ _ e -> throwE $ FTCEInvalidHeader e
|
||||
A.Partial _ -> throwE $ FTCEInvalidHeader "incomplete"
|
||||
A.Done rest hdr -> pure (hdr, LB.fromStrict rest <> s')
|
||||
|
||||
readChunks :: [FilePath] -> IO LB.ByteString
|
||||
|
||||
@@ -18,6 +18,7 @@ import Control.Logger.Simple
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Reader
|
||||
import Control.Monad.Trans.Except
|
||||
import Data.Bifunctor (first)
|
||||
import qualified Data.ByteString.Base64.URL as B64
|
||||
import Data.ByteString.Builder (Builder, byteString)
|
||||
@@ -136,7 +137,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
|
||||
either sendError pure r
|
||||
where
|
||||
processHello = do
|
||||
unless (B.null bodyHead) $ throwError HANDSHAKE
|
||||
unless (B.null bodyHead) $ throwE HANDSHAKE
|
||||
(k, pk) <- atomically . C.generateKeyPair =<< asks random
|
||||
atomically $ TM.insert sessionId (HandshakeSent pk) sessions
|
||||
let authPubKey = (chain, C.signX509 serverSignKey $ C.publicToX509 k)
|
||||
@@ -148,11 +149,11 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
|
||||
liftIO . sendResponse $ H.responseBuilder N.ok200 [] shs
|
||||
pure Nothing
|
||||
processClientHandshake pk = do
|
||||
unless (B.length bodyHead == xftpBlockSize) $ throwError HANDSHAKE
|
||||
unless (B.length bodyHead == xftpBlockSize) $ throwE HANDSHAKE
|
||||
body <- liftHS $ C.unPad bodyHead
|
||||
XFTPClientHandshake {xftpVersion = v, keyHash} <- liftHS $ smpDecode body
|
||||
kh <- asks serverIdentity
|
||||
unless (keyHash == kh) $ throwError HANDSHAKE
|
||||
unless (keyHash == kh) $ throwE HANDSHAKE
|
||||
case compatibleVRange' xftpServerVRange v of
|
||||
Just (Compatible vr) -> do
|
||||
let auth = THAuthServer {serverPrivKey = pk, sessSecret' = Nothing}
|
||||
@@ -163,7 +164,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
|
||||
#endif
|
||||
liftIO . sendResponse $ H.responseNoBody N.ok200 []
|
||||
pure Nothing
|
||||
Nothing -> throwError HANDSHAKE
|
||||
Nothing -> throwE HANDSHAKE
|
||||
sendError :: XFTPErrorType -> M (Maybe (THandleParams XFTPVersion 'TServer))
|
||||
sendError err = do
|
||||
runExceptT (encodeXftp err) >>= \case
|
||||
@@ -395,7 +396,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
|
||||
st <- asks store
|
||||
r <- runExceptT $ do
|
||||
sizes <- asks $ allowedChunkSizes . config
|
||||
unless (size file `elem` sizes) $ throwError SIZE
|
||||
unless (size file `elem` sizes) $ throwE SIZE
|
||||
ts <- liftIO getSystemTime
|
||||
-- TODO validate body empty
|
||||
sId <- ExceptT $ addFileRetry st file 3 ts
|
||||
|
||||
@@ -194,7 +194,7 @@ receiveFile_ :: (Handle -> Word32 -> IO (Either XFTPErrorType ())) -> XFTPRcvChu
|
||||
receiveFile_ receive XFTPRcvChunkSpec {filePath, chunkSize, chunkDigest} = do
|
||||
ExceptT $ withFile filePath WriteMode (`receive` chunkSize)
|
||||
digest' <- liftIO $ LC.sha256Hash <$> LB.readFile filePath
|
||||
when (digest' /= chunkDigest) $ throwError DIGEST
|
||||
when (digest' /= chunkDigest) $ throwE DIGEST
|
||||
|
||||
data XFTPErrorType
|
||||
= -- | incorrect block format, encoding or signature size
|
||||
|
||||
@@ -649,7 +649,7 @@ joinConnAsync c userId corrId enableNtfs cReqUri@CRInvitationUri {} cInfo pqSup
|
||||
connId <- withStore c $ \db -> createNewConn db g cData SCMInvitation
|
||||
enqueueCommand c corrId connId Nothing $ AClientCommand $ APC SAEConn $ JOIN enableNtfs (ACR sConnectionMode cReqUri) pqSupport subMode cInfo
|
||||
pure connId
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Nothing -> throwE $ AGENT A_VERSION
|
||||
joinConnAsync _c _userId _corrId _enableNtfs (CRContactUri _) _subMode _cInfo _pqEncryption =
|
||||
throwE $ CMD PROHIBITED "joinConnAsync"
|
||||
|
||||
@@ -668,7 +668,7 @@ acceptContactAsync' c corrId enableNtfs invId ownConnInfo pqSupport subMode = do
|
||||
withStore' c $ \db -> acceptInvitation db invId ownConnInfo
|
||||
joinConnAsync c userId corrId enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
|
||||
withStore' c (`unacceptInvitation` invId)
|
||||
throwError err
|
||||
throwE err
|
||||
_ -> throwE $ CMD PROHIBITED "acceptContactAsync"
|
||||
|
||||
ackMessageAsync' :: AgentClient -> ACorrId -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AM ()
|
||||
@@ -677,7 +677,7 @@ ackMessageAsync' c corrId connId msgId rcptInfo_ = do
|
||||
case cType of
|
||||
SCDuplex -> enqueueAck
|
||||
SCRcv -> enqueueAck
|
||||
SCSnd -> throwError $ CONN SIMPLEX
|
||||
SCSnd -> throwE $ CONN SIMPLEX
|
||||
SCContact -> throwE $ CMD PROHIBITED "ackMessageAsync: SCContact"
|
||||
SCNew -> throwE $ CMD PROHIBITED "ackMessageAsync: SCNew"
|
||||
where
|
||||
@@ -740,7 +740,7 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srv
|
||||
(SCMContact, CR.IKUsePQ) -> throwE $ CMD PROHIBITED "newRcvConnSrv"
|
||||
_ -> pure ()
|
||||
AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
|
||||
(rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwError e
|
||||
(rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwE e
|
||||
rq' <- withStore c $ \db -> updateNewConnRcv db connId rq
|
||||
lift . when (subMode == SMSubscribe) $ addNewQueueSubscription c rq' tSess sessId
|
||||
when enableNtfs $ do
|
||||
@@ -760,11 +760,11 @@ newConnToJoin c userId connId enableNtfs cReq pqSup = case cReq of
|
||||
CRInvitationUri {} ->
|
||||
lift (compatibleInvitationUri cReq) >>= \case
|
||||
Just (_, (Compatible (CR.E2ERatchetParams v _ _ _)), aVersion) -> create aVersion (Just v)
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Nothing -> throwE $ AGENT A_VERSION
|
||||
CRContactUri {} ->
|
||||
lift (compatibleContactUri cReq) >>= \case
|
||||
Just (_, aVersion) -> create aVersion Nothing
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Nothing -> throwE $ AGENT A_VERSION
|
||||
where
|
||||
create :: Compatible VersionSMPA -> Maybe CR.VersionE2E -> AM ConnId
|
||||
create (Compatible connAgentVersion) e2eV_ = do
|
||||
@@ -796,7 +796,7 @@ startJoinInvitation userId connId enableNtfs cReqUri pqSup =
|
||||
q <- lift $ newSndQueue userId "" qInfo
|
||||
let cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
|
||||
pure (cData, q, rc, e2eSndParams)
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Nothing -> throwE $ AGENT A_VERSION
|
||||
|
||||
connRequestPQSupport :: AgentClient -> PQSupport -> ConnectionRequestUri c -> IO (Maybe (VersionSMPA, PQSupport))
|
||||
connRequestPQSupport c pqSup cReq = withAgentEnv' c $ case cReq of
|
||||
@@ -846,14 +846,14 @@ joinConnSrv c userId connId hasNewConn enableNtfs inv@CRInvitationUri {} cInfo p
|
||||
Left e -> do
|
||||
-- possible improvement: recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
|
||||
void $ withStore' c $ \db -> deleteConn db Nothing connId'
|
||||
throwError e
|
||||
throwE e
|
||||
joinConnSrv c userId connId hasNewConn enableNtfs cReqUri@CRContactUri {} cInfo pqSup subMode srv =
|
||||
lift (compatibleContactUri cReqUri) >>= \case
|
||||
Just (qInfo, vrsn) -> do
|
||||
(connId', cReq) <- newConnSrv c userId connId hasNewConn enableNtfs SCMInvitation Nothing (CR.IKNoPQ pqSup) subMode srv
|
||||
void $ sendInvitation c userId qInfo vrsn cReq cInfo
|
||||
pure connId'
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Nothing -> throwE $ AGENT A_VERSION
|
||||
|
||||
joinConnSrvAsync :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> SMPServerWithAuth -> AM ()
|
||||
joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSupport subMode srv = do
|
||||
@@ -899,7 +899,7 @@ acceptContact' c connId enableNtfs invId ownConnInfo pqSupport subMode = withCon
|
||||
withStore' c $ \db -> acceptInvitation db invId ownConnInfo
|
||||
joinConn c userId connId False enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
|
||||
withStore' c (`unacceptInvitation` invId)
|
||||
throwError err
|
||||
throwE err
|
||||
_ -> throwE $ CMD PROHIBITED "acceptContact"
|
||||
|
||||
-- | Reject contact (RJCT command) in Reader monad
|
||||
@@ -916,8 +916,8 @@ subscribeConnection' c connId = toConnResult connId =<< subscribeConnections' c
|
||||
toConnResult :: ConnId -> Map ConnId (Either AgentErrorType ()) -> AM ()
|
||||
toConnResult connId rs = case M.lookup connId rs of
|
||||
Just (Right ()) -> when (M.size rs > 1) $ logError $ T.pack $ "too many results " <> show (M.size rs)
|
||||
Just (Left e) -> throwError e
|
||||
_ -> throwError $ INTERNAL $ "no result for connection " <> B.unpack connId
|
||||
Just (Left e) -> throwE e
|
||||
_ -> throwE $ INTERNAL $ "no result for connection " <> B.unpack connId
|
||||
|
||||
type QCmdResult = (QueueStatus, Either AgentErrorType ())
|
||||
|
||||
@@ -1006,7 +1006,7 @@ getConnectionMessage' c connId = do
|
||||
DuplexConnection _ (rq :| _) _ -> getQueueMessage c rq
|
||||
RcvConnection _ rq -> getQueueMessage c rq
|
||||
ContactConnection _ rq -> getQueueMessage c rq
|
||||
SndConnection _ _ -> throwError $ CONN SIMPLEX
|
||||
SndConnection _ _ -> throwE $ CONN SIMPLEX
|
||||
NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection"
|
||||
|
||||
getNotificationMessage' :: AgentClient -> C.CbNonce -> ByteString -> AM (NotificationInfo, [SMPMsgMeta])
|
||||
@@ -1146,7 +1146,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
|
||||
RcvConnection cData rq -> do
|
||||
secure rq senderKey
|
||||
mapM_ (connectReplyQueues c cData ownConnInfo) (L.nonEmpty $ smpReplyQueues senderConf)
|
||||
_ -> throwError $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
|
||||
_ -> throwE $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
|
||||
ICDuplexSecure _rId senderKey -> withServer' . tryWithLock "ICDuplexSecure" . withDuplexConn $ \(DuplexConnection cData (rq :| _) (sq :| _)) -> do
|
||||
secure rq senderKey
|
||||
void $ enqueueMessage c cData sq SMP.MsgFlags {notification = True} HELLO
|
||||
@@ -1182,8 +1182,8 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
|
||||
tryError (deleteQueue c rq') >>= \case
|
||||
Right () -> finalizeSwitch
|
||||
Left e
|
||||
| temporaryOrHostError e -> throwError e
|
||||
| otherwise -> finalizeSwitch >> throwError e
|
||||
| temporaryOrHostError e -> throwE e
|
||||
| otherwise -> finalizeSwitch >> throwE e
|
||||
where
|
||||
finalizeSwitch = do
|
||||
withStore' c $ \db -> deleteConnRcvQueue db rq'
|
||||
@@ -1229,7 +1229,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
|
||||
|
||||
enqueueMessages :: AgentClient -> ConnData -> NonEmpty SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption)
|
||||
enqueueMessages c cData sqs msgFlags aMessage = do
|
||||
when (ratchetSyncSendProhibited cData) $ throwError $ INTERNAL "enqueueMessages: ratchet is not synchronized"
|
||||
when (ratchetSyncSendProhibited cData) $ throwE $ INTERNAL "enqueueMessages: ratchet is not synchronized"
|
||||
enqueueMessages' c cData sqs msgFlags aMessage
|
||||
|
||||
enqueueMessages' :: AgentClient -> ConnData -> NonEmpty SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, CR.PQEncryption)
|
||||
@@ -1482,7 +1482,7 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
|
||||
case conn of
|
||||
DuplexConnection {} -> ack >> sendRcpt conn >> del
|
||||
RcvConnection {} -> ack >> del
|
||||
SndConnection {} -> throwError $ CONN SIMPLEX
|
||||
SndConnection {} -> throwE $ CONN SIMPLEX
|
||||
ContactConnection {} -> throwE $ CMD PROHIBITED "ackMessage: ContactConnection"
|
||||
NewConnection _ -> throwE $ CMD PROHIBITED "ackMessage: NewConnection"
|
||||
where
|
||||
@@ -1566,7 +1566,7 @@ abortConnectionSwitch' c connId =
|
||||
let rqs'' = updatedQs rq' rqs'
|
||||
conn' = DuplexConnection cData rqs'' sqs
|
||||
pure $ connectionStats conn'
|
||||
_ -> throwError $ INTERNAL "won't delete all rcv queues in connection"
|
||||
_ -> throwE $ INTERNAL "won't delete all rcv queues in connection"
|
||||
| otherwise -> throwE $ CMD PROHIBITED "abortConnectionSwitch: no rcv queues left"
|
||||
_ -> throwE $ CMD PROHIBITED "abortConnectionSwitch: not allowed"
|
||||
_ -> throwE $ CMD PROHIBITED "abortConnectionSwitch: not duplex"
|
||||
@@ -1596,7 +1596,7 @@ ackQueueMessage :: AgentClient -> RcvQueue -> SMP.MsgId -> AM ()
|
||||
ackQueueMessage c rq srvMsgId =
|
||||
sendAck c rq srvMsgId `catchAgentError` \case
|
||||
SMP _ SMP.NO_MSG -> pure ()
|
||||
e -> throwError e
|
||||
e -> throwE e
|
||||
|
||||
-- | Suspend SMP agent connection (OFF command) in Reader monad
|
||||
suspendConnection' :: AgentClient -> ConnId -> AM ()
|
||||
@@ -1606,7 +1606,7 @@ suspendConnection' c connId = withConnLock c connId "suspendConnection" $ do
|
||||
DuplexConnection _ rqs _ -> mapM_ (suspendQueue c) rqs
|
||||
RcvConnection _ rq -> suspendQueue c rq
|
||||
ContactConnection _ rq -> suspendQueue c rq
|
||||
SndConnection _ _ -> throwError $ CONN SIMPLEX
|
||||
SndConnection _ _ -> throwE $ CONN SIMPLEX
|
||||
NewConnection _ -> throwE $ CMD PROHIBITED "suspendConnection"
|
||||
|
||||
-- | Delete SMP agent connection (DEL command) in Reader monad
|
||||
@@ -1818,7 +1818,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
|
||||
ns <- asks ntfSupervisor
|
||||
tryReplace ns `catchAgentError` \e ->
|
||||
if temporaryOrHostError e
|
||||
then throwError e
|
||||
then throwE e
|
||||
else do
|
||||
withStore' c $ \db -> removeNtfToken db tkn
|
||||
atomically $ nsRemoveNtfToken ns
|
||||
@@ -1906,7 +1906,7 @@ toggleConnectionNtfs' c connId enable = do
|
||||
DuplexConnection cData _ _ -> toggle cData
|
||||
RcvConnection cData _ -> toggle cData
|
||||
ContactConnection cData _ -> toggle cData
|
||||
_ -> throwError $ CONN SIMPLEX
|
||||
_ -> throwE $ CONN SIMPLEX
|
||||
where
|
||||
toggle :: ConnData -> AM ()
|
||||
toggle cData
|
||||
@@ -1926,7 +1926,7 @@ deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
|
||||
atomically $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
|
||||
agentNtfDeleteToken c tknId tkn `catchAgentError` \case
|
||||
NTF _ AUTH -> pure ()
|
||||
e -> throwError e
|
||||
e -> throwE e
|
||||
withStore' c $ \db -> removeNtfToken db tkn
|
||||
atomically $ nsRemoveNtfToken ns
|
||||
|
||||
@@ -1946,8 +1946,8 @@ withToken c tkn@NtfToken {deviceToken, ntfMode} from_ (toStatus, toAction_) f =
|
||||
withStore' c $ \db -> removeNtfToken db tkn
|
||||
atomically $ nsRemoveNtfToken ns
|
||||
void $ registerNtfToken' c deviceToken ntfMode
|
||||
throwError e
|
||||
Left e -> throwError e
|
||||
throwE e
|
||||
Left e -> throwE e
|
||||
|
||||
initializeNtfSubs :: AgentClient -> AM ()
|
||||
initializeNtfSubs c = sendNtfConnCommands c NSCCreate
|
||||
@@ -2179,7 +2179,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
clientMsg@SMP.ClientMsgEnvelope {cmHeader = SMP.PubHeader phVer e2ePubKey_} <-
|
||||
parseMessage msgBody
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
unless (phVer `isCompatible` clientVRange) . throwError $ AGENT A_VERSION
|
||||
unless (phVer `isCompatible` clientVRange) . throwE $ AGENT A_VERSION
|
||||
case (e2eDhSecret, e2ePubKey_) of
|
||||
(Nothing, Just e2ePubKey) -> do
|
||||
let e2eDh = C.dh' e2ePubKey e2ePrivKey
|
||||
@@ -2275,7 +2275,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
checkDuplicateHash :: AgentErrorType -> ByteString -> AM ()
|
||||
checkDuplicateHash e encryptedMsgHash =
|
||||
unlessM (withStore' c $ \db -> checkRcvMsgHashExists db connId encryptedMsgHash) $
|
||||
throwError e
|
||||
throwE e
|
||||
updateTotalMsgCount :: STM ()
|
||||
updateTotalMsgCount =
|
||||
TM.lookup connId (msgCounts c) >>= \case
|
||||
@@ -2368,7 +2368,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
-- aVRange <- asks $ smpAgentVRange . config
|
||||
-- if agentVersion agentEnvelope `isCompatible` aVRange
|
||||
-- then pure (privHeader, agentEnvelope)
|
||||
-- else throwError $ AGENT A_VERSION
|
||||
-- else throwE $ AGENT A_VERSION
|
||||
pure (privHeader, agentEnvelope)
|
||||
|
||||
parseMessage :: Encoding a => ByteString -> AM a
|
||||
@@ -2381,12 +2381,12 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
let ConnData {pqSupport} = toConnData conn'
|
||||
unless
|
||||
(agentVersion `isCompatible` smpAgentVRange && smpClientVersion `isCompatible` smpClientVRange)
|
||||
(throwError $ AGENT A_VERSION)
|
||||
(throwE $ AGENT A_VERSION)
|
||||
case status of
|
||||
New -> case (conn', e2eEncryption) of
|
||||
-- party initiating connection
|
||||
(RcvConnection _ _, Just (CR.AE2ERatchetParams _ e2eSndParams@(CR.E2ERatchetParams e2eVersion _ _ _))) -> do
|
||||
unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwError $ AGENT A_VERSION)
|
||||
unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwE $ AGENT A_VERSION)
|
||||
(pk1, rcDHRs, pKem) <- withStore c (`getRatchetX3dhKeys` connId)
|
||||
rcParams <- liftError cryptoError $ CR.pqX3dhRcv pk1 rcDHRs pKem e2eSndParams
|
||||
let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eEncryptVRange}
|
||||
@@ -2482,7 +2482,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
qAddMsg :: SMP.MsgId -> NonEmpty (SMPQueueUri, Maybe SndQAddr) -> Connection 'CDuplex -> AM ()
|
||||
qAddMsg _ ((_, Nothing) :| _) _ = qError "adding queue without switching is not supported"
|
||||
qAddMsg srvMsgId ((qUri, Just addr) :| _) (DuplexConnection cData' rqs sqs) = do
|
||||
when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
|
||||
when (ratchetSyncSendProhibited cData') $ throwE $ AGENT (A_QUEUE "ratchet is not synchronized")
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
case qUri `compatibleVersion` clientVRange of
|
||||
Just qInfo@(Compatible sqInfo@SMPQueueInfo {queueAddress}) ->
|
||||
@@ -2509,14 +2509,14 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
_ -> qError "absent sender keys"
|
||||
_ -> qError "QADD: won't delete all snd queues in connection"
|
||||
_ -> qError "QADD: replaced queue address is not found in connection"
|
||||
_ -> throwError $ AGENT A_VERSION
|
||||
_ -> throwE $ AGENT A_VERSION
|
||||
|
||||
-- processed by queue recipient
|
||||
qKeyMsg :: SMP.MsgId -> NonEmpty (SMPQueueInfo, SndPublicAuthKey) -> Connection 'CDuplex -> AM ()
|
||||
qKeyMsg srvMsgId ((qInfo, senderKey) :| _) conn'@(DuplexConnection cData' rqs _) = do
|
||||
when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
|
||||
when (ratchetSyncSendProhibited cData') $ throwE $ AGENT (A_QUEUE "ratchet is not synchronized")
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
unless (qInfo `isCompatible` clientVRange) . throwError $ AGENT A_VERSION
|
||||
unless (qInfo `isCompatible` clientVRange) . throwE $ AGENT A_VERSION
|
||||
case findRQ (smpServer, senderId) rqs of
|
||||
Just rq'@RcvQueue {rcvId, e2ePrivKey = dhPrivKey, smpClientVersion = cVer, status = status'}
|
||||
| status' == New || status' == Confirmed -> do
|
||||
@@ -2536,7 +2536,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
qUseMsg :: SMP.MsgId -> NonEmpty ((SMPServer, SMP.SenderId), Bool) -> Connection 'CDuplex -> AM ()
|
||||
-- NOTE: does not yet support the change of the primary status during the rotation
|
||||
qUseMsg srvMsgId ((addr, _primary) :| _) (DuplexConnection cData' rqs sqs) = do
|
||||
when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
|
||||
when (ratchetSyncSendProhibited cData') $ throwE $ AGENT (A_QUEUE "ratchet is not synchronized")
|
||||
case findQ addr sqs of
|
||||
Just sq'@SndQueue {dbReplaceQueueId = Just replaceQId} -> do
|
||||
case find ((replaceQId ==) . dbQId) sqs of
|
||||
@@ -2555,7 +2555,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
_ -> qError "QUSE: switched queue address not found in connection"
|
||||
|
||||
qError :: String -> AM a
|
||||
qError = throwError . AGENT . A_QUEUE
|
||||
qError = throwE . AGENT . A_QUEUE
|
||||
|
||||
ereadyMsg :: CR.RatchetX448 -> Connection 'CDuplex -> AM ()
|
||||
ereadyMsg rcPrev (DuplexConnection cData'@ConnData {lastExternalSndId} _ sqs) = do
|
||||
@@ -2591,7 +2591,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) =
|
||||
unlessM ratchetExists $ do
|
||||
AgentConfig {e2eEncryptVRange} <- asks config
|
||||
unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwError $ AGENT A_VERSION)
|
||||
unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwE $ AGENT A_VERSION)
|
||||
keys <- getSendRatchetKeys
|
||||
let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eEncryptVRange}
|
||||
initRatchet rcVs keys
|
||||
@@ -2616,7 +2616,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
-- can communicate for other client to reset to RSRequired
|
||||
-- - need to add new AgentMsgEnvelope, AgentMessage, AgentMessageType
|
||||
-- - need to deduplicate on receiving side
|
||||
throwError $ AGENT (A_CRYPTO RATCHET_SYNC)
|
||||
throwE $ AGENT (A_CRYPTO RATCHET_SYNC)
|
||||
where
|
||||
sendReplyKey = do
|
||||
g <- asks random
|
||||
@@ -2671,7 +2671,7 @@ checkSQSwchStatus sq@SndQueue {sndSwchStatus} expected =
|
||||
|
||||
switchStatusError :: (SMPQueueRec q, Show a) => q -> a -> Maybe a -> AM ()
|
||||
switchStatusError q expected actual =
|
||||
throwError . INTERNAL $
|
||||
throwE . INTERNAL $
|
||||
("unexpected switch status, queueId=" <> show (queueId q))
|
||||
<> (", expected=" <> show expected)
|
||||
<> (", actual=" <> show actual)
|
||||
@@ -2680,7 +2680,7 @@ connectReplyQueues :: AgentClient -> ConnData -> ConnInfo -> NonEmpty SMPQueueIn
|
||||
connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo (qInfo :| _) = do
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
case qInfo `proveCompatible` clientVRange of
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Nothing -> throwE $ AGENT A_VERSION
|
||||
Just qInfo' -> do
|
||||
sq <- lift $ newSndQueue userId connId qInfo'
|
||||
sq' <- withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
|
||||
|
||||
@@ -586,7 +586,7 @@ instance ProtocolServerClient XFTPVersion XFTPErrorType FileResponse where
|
||||
|
||||
getSMPServerClient :: AgentClient -> SMPTransportSession -> AM SMPConnectedClient
|
||||
getSMPServerClient c@AgentClient {active, smpClients, workerSeq} tSess = do
|
||||
unlessM (readTVarIO active) . throwError $ INACTIVE
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess smpClients ts)
|
||||
>>= either newClient (waitForProtocolClient c tSess smpClients)
|
||||
@@ -597,7 +597,7 @@ getSMPServerClient c@AgentClient {active, smpClients, workerSeq} tSess = do
|
||||
|
||||
getSMPProxyClient :: AgentClient -> SMPTransportSession -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay)
|
||||
getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq} destSess@(userId, destSrv, qId) = do
|
||||
unlessM (readTVarIO active) . throwError $ INACTIVE
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
proxySrv <- getNextServer c userId [destSrv]
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getClientVar proxySrv ts) >>= \(tSess, auth, v) ->
|
||||
@@ -652,7 +652,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq
|
||||
smpConnectClient :: AgentClient -> SMPTransportSession -> TMap SMPServer ProxiedRelayVar -> SMPClientVar -> AM SMPConnectedClient
|
||||
smpConnectClient c@AgentClient {smpClients, msgQ} tSess@(_, srv, _) prs v =
|
||||
newProtocolClient c tSess smpClients connectClient v
|
||||
`catchAgentError` \e -> lift (resubscribeSMPSession c tSess) >> throwError e
|
||||
`catchAgentError` \e -> lift (resubscribeSMPSession c tSess) >> throwE e
|
||||
where
|
||||
connectClient :: SMPClientVar -> AM SMPConnectedClient
|
||||
connectClient v' = do
|
||||
@@ -748,7 +748,7 @@ reconnectSMPClient c tSess@(_, srv, _) qs = handleNotify $ do
|
||||
|
||||
getNtfServerClient :: AgentClient -> NtfTransportSession -> AM NtfClient
|
||||
getNtfServerClient c@AgentClient {active, ntfClients, workerSeq} tSess@(userId, srv, _) = do
|
||||
unlessM (readTVarIO active) . throwError $ INACTIVE
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess ntfClients ts)
|
||||
>>= either
|
||||
@@ -772,7 +772,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq} tSess@(userId,
|
||||
|
||||
getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
|
||||
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq} tSess@(userId, srv, _) = do
|
||||
unlessM (readTVarIO active) . throwError $ INACTIVE
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess xftpClients ts)
|
||||
>>= either
|
||||
@@ -988,7 +988,7 @@ withClient_ c tSess@(userId, srv, _) statCmd action = do
|
||||
logServerError cl e = do
|
||||
logServer "<--" c srv "" $ strEncode e
|
||||
stat cl $ strEncode e
|
||||
throwError e
|
||||
throwE e
|
||||
|
||||
withProxySession :: AgentClient -> SMPTransportSession -> SMP.SenderId -> ByteString -> ((SMPConnectedClient, ProxiedRelay) -> AM a) -> AM a
|
||||
withProxySession c destSess@(userId, destSrv, _) entId cmdStr action = do
|
||||
@@ -1007,7 +1007,7 @@ withProxySession c destSess@(userId, destSrv, _) entId cmdStr action = do
|
||||
logServerError cl e = do
|
||||
logServer ("<-- " <> proxySrv cl <> " <") c destSrv "" $ strEncode e
|
||||
stat cl $ strEncode e
|
||||
throwError e
|
||||
throwE e
|
||||
|
||||
withLogClient_ :: ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> AM a) -> AM a
|
||||
withLogClient_ c tSess@(_, srv, _) entId cmdStr action = do
|
||||
@@ -1192,7 +1192,7 @@ runXFTPServerTest c userId (ProtoServerWithAuth srv auth) = do
|
||||
liftError (testErr TSUploadFile) $ X.uploadXFTPChunk xftp spKey sId chunkSpec
|
||||
liftError (testErr TSDownloadFile) $ X.downloadXFTPChunk g xftp rpKey rId $ XFTPRcvChunkSpec rcvPath chSize digest
|
||||
rcvDigest <- liftIO $ C.sha256Hash <$> B.readFile rcvPath
|
||||
unless (digest == rcvDigest) $ throwError $ ProtocolTestFailure TSCompareFile $ XFTP (B.unpack $ strEncode srv) DIGEST
|
||||
unless (digest == rcvDigest) $ throwE $ ProtocolTestFailure TSCompareFile $ XFTP (B.unpack $ strEncode srv) DIGEST
|
||||
liftError (testErr TSDeleteFile) $ X.deleteXFTPChunk xftp spKey sId
|
||||
ok <- tcpTimeout xftpNetworkConfig `timeout` X.closeXFTPClient xftp
|
||||
incClientStat c userId xftp "XFTP_TEST" "OK"
|
||||
@@ -1486,7 +1486,7 @@ sendConfirmation c sq@SndQueue {userId, server, sndId, sndPublicKey = Just sndPu
|
||||
let clientMsg = SMP.ClientMessage (SMP.PHConfirmation sndPublicKey) agentConfirmation
|
||||
msg <- agentCbEncrypt sq e2ePubKey $ smpEncode clientMsg
|
||||
sendOrProxySMPMessage c userId server "<CONF>" Nothing sndId (MsgFlags {notification = True}) msg
|
||||
sendConfirmation _ _ _ = throwError $ INTERNAL "sendConfirmation called without snd_queue public key(s) in the database"
|
||||
sendConfirmation _ _ _ = throwE $ INTERNAL "sendConfirmation called without snd_queue public key(s) in the database"
|
||||
|
||||
sendInvitation :: AgentClient -> UserId -> Compatible SMPQueueInfo -> Compatible VersionSMPA -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> AM (Maybe SMPServer)
|
||||
sendInvitation c userId (Compatible (SMPQueueInfo v SMPQueueAddress {smpServer, senderId, dhPublicKey})) (Compatible agentVersion) connReq connInfo = do
|
||||
@@ -1657,7 +1657,7 @@ xftpRcvKeys n = do
|
||||
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"
|
||||
_ -> throwE $ INTERNAL "non-positive number of recipients"
|
||||
|
||||
xftpRcvIdsKeys :: NonEmpty ByteString -> NonEmpty C.AAuthKeyPair -> NonEmpty (ChunkReplicaId, C.APrivateAuthKey)
|
||||
xftpRcvIdsKeys rIds rKeys = L.map ChunkReplicaId rIds `L.zip` L.map snd rKeys
|
||||
@@ -1895,7 +1895,7 @@ withUserServers :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient
|
||||
withUserServers c userId action =
|
||||
atomically (TM.lookup userId $ userServers c) >>= \case
|
||||
Just srvs -> action srvs
|
||||
_ -> throwError $ INTERNAL "unknown userId - no user servers"
|
||||
_ -> throwE $ INTERNAL "unknown userId - no user servers"
|
||||
|
||||
withNextSrv :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> TVar [ProtocolServer p] -> [ProtocolServer p] -> (ProtoServerWithAuth p -> AM a) -> AM a
|
||||
withNextSrv c userId usedSrvs initUsed action = do
|
||||
|
||||
@@ -225,6 +225,7 @@ import Control.Logger.Simple
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.IO.Class
|
||||
import Control.Monad.Trans.Except
|
||||
import Crypto.Random (ChaChaDRG)
|
||||
import qualified Data.Aeson.TH as J
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
@@ -1045,7 +1046,7 @@ getWorkItem :: Show i => ByteString -> IO (Maybe i) -> (i -> IO (Either StoreErr
|
||||
getWorkItem itemName getId getItem markFailed =
|
||||
runExceptT $ handleErr "getId" getId >>= mapM tryGetItem
|
||||
where
|
||||
tryGetItem itemId = ExceptT (getItem itemId) `catchStoreErrors` \e -> mark itemId >> throwError e
|
||||
tryGetItem itemId = ExceptT (getItem itemId) `catchStoreErrors` \e -> mark itemId >> throwE e
|
||||
mark itemId = handleErr ("markFailed ID " <> bshow itemId) $ markFailed itemId
|
||||
catchStoreErrors = catchAllErrors (SEInternal . bshow)
|
||||
-- Errors caught by this function will suspend worker as if there is no more work,
|
||||
|
||||
@@ -933,8 +933,8 @@ forwardSMPMessage :: SMPClient -> CorrId -> VersionSMP -> C.PublicKeyX25519 -> E
|
||||
forwardSMPMessage c@ProtocolClient {thParams, client_ = PClient {clientCorrId = g}} fwdCorrId fwdVersion fwdKey fwdTransmission = do
|
||||
-- prepare params
|
||||
sessSecret <- case thAuth thParams of
|
||||
Nothing -> throwError $ PCETransportError TENoServerAuth
|
||||
Just THAuthClient {sessSecret} -> maybe (throwError $ PCETransportError TENoServerAuth) pure sessSecret
|
||||
Nothing -> throwE $ PCETransportError TENoServerAuth
|
||||
Just THAuthClient {sessSecret} -> maybe (throwE $ PCETransportError TENoServerAuth) pure sessSecret
|
||||
nonce <- liftIO . atomically $ C.randomCbNonce g
|
||||
-- wrap
|
||||
let fwdT = FwdTransmission {fwdCorrId, fwdVersion, fwdKey, fwdTransmission}
|
||||
|
||||
@@ -23,6 +23,7 @@ where
|
||||
import Control.Exception
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Trans.Except
|
||||
import Crypto.Random (ChaChaDRG)
|
||||
import qualified Data.Aeson.TH as J
|
||||
import qualified Data.ByteArray as BA
|
||||
@@ -56,10 +57,10 @@ readFile (CryptoFile path cfArgs) = do
|
||||
case cfArgs of
|
||||
Just (CFArgs (C.SbKey key) (C.CbNonce nonce)) -> do
|
||||
let len = LB.length s - fromIntegral C.authTagSize
|
||||
when (len < 0) $ throwError FTCEInvalidFileSize
|
||||
when (len < 0) $ throwE FTCEInvalidFileSize
|
||||
let (s', tag') = LB.splitAt len s
|
||||
(tag :| cs) <- liftEitherWith FTCECryptoError $ LC.secretBox LC.sbDecryptChunk key nonce s'
|
||||
unless (BA.constEq (LB.toStrict tag') tag) $ throwError FTCEInvalidAuthTag
|
||||
unless (BA.constEq (LB.toStrict tag') tag) $ throwE FTCEInvalidAuthTag
|
||||
pure $ LB.fromChunks cs
|
||||
Nothing -> pure s
|
||||
|
||||
@@ -96,7 +97,7 @@ hGetTag :: CryptoFileHandle -> ExceptT FTCryptoError IO ()
|
||||
hGetTag (CFHandle h sb_) = forM_ sb_ $ \sb -> do
|
||||
tag <- liftIO $ B.hGet h C.authTagSize
|
||||
tag' <- LC.sbAuth <$> readTVarIO sb
|
||||
unless (BA.constEq tag tag') $ throwError FTCEInvalidAuthTag
|
||||
unless (BA.constEq tag tag') $ throwE FTCEInvalidAuthTag
|
||||
|
||||
data FTCryptoError
|
||||
= FTCECryptoError C.CryptoError
|
||||
|
||||
@@ -447,7 +447,7 @@ pqX3dhRcv rpk1 rpk2 rpKem_ (E2ERatchetParams v sk1 sk2 sKem_) = do
|
||||
Just (PrivateRKParamsProposed ks@(_, pk)) -> do
|
||||
shared <- liftIO $ sntrup761Dec ct pk
|
||||
pure $ Just (ks, RatchetKEMAccepted k' shared ct)
|
||||
Nothing -> throwError CERatchetKEMState
|
||||
Nothing -> throwE CERatchetKEMState
|
||||
_ -> pure Nothing -- both parties can send "proposal" in case of ratchet renegotiation
|
||||
|
||||
pqX3dh :: DhAlgorithm a => (PublicKey a, PublicKey a) -> DhSecret a -> DhSecret a -> DhSecret a -> Maybe RatchetKEMAccepted -> RatchetInitParams
|
||||
|
||||
@@ -15,6 +15,7 @@ import Control.Logger.Simple
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.IO.Class
|
||||
import Control.Monad.Trans.Except
|
||||
import Crypto.Hash.Algorithms (SHA256 (..))
|
||||
import qualified Crypto.PubKey.ECC.ECDSA as EC
|
||||
import qualified Crypto.PubKey.ECC.Types as ECT
|
||||
@@ -353,18 +354,18 @@ apnsPushProviderClient c@APNSPushClient {nonceDrg, apnsCfg} tkn@NtfTknData {toke
|
||||
| status == Just N.ok200 = pure ()
|
||||
| status == Just N.badRequest400 =
|
||||
case reason' of
|
||||
"BadDeviceToken" -> throwError PPTokenInvalid
|
||||
"DeviceTokenNotForTopic" -> throwError PPTokenInvalid
|
||||
"TopicDisallowed" -> throwError PPPermanentError
|
||||
"BadDeviceToken" -> throwE PPTokenInvalid
|
||||
"DeviceTokenNotForTopic" -> throwE PPTokenInvalid
|
||||
"TopicDisallowed" -> throwE PPPermanentError
|
||||
_ -> err status reason'
|
||||
| status == Just N.forbidden403 = case reason' of
|
||||
"ExpiredProviderToken" -> throwError PPPermanentError -- there should be no point retrying it as the token was refreshed
|
||||
"InvalidProviderToken" -> throwError PPPermanentError
|
||||
"ExpiredProviderToken" -> throwE PPPermanentError -- there should be no point retrying it as the token was refreshed
|
||||
"InvalidProviderToken" -> throwE PPPermanentError
|
||||
_ -> err status reason'
|
||||
| status == Just N.gone410 = throwError PPTokenInvalid
|
||||
| status == Just N.serviceUnavailable503 = liftIO (disconnectApnsHTTP2Client c) >> throwError PPRetryLater
|
||||
| status == Just N.gone410 = throwE PPTokenInvalid
|
||||
| status == Just N.serviceUnavailable503 = liftIO (disconnectApnsHTTP2Client c) >> throwE PPRetryLater
|
||||
-- Just tooManyRequests429 -> TooManyRequests - too many requests for the same token
|
||||
| otherwise = err status reason'
|
||||
err :: Maybe Status -> Text -> ExceptT PushProviderError IO ()
|
||||
err s r = throwError $ PPResponseError s r
|
||||
err s r = throwE $ PPResponseError s r
|
||||
liftHTTPS2 a = ExceptT $ first PPConnection <$> a
|
||||
@@ -116,7 +116,7 @@ ntfServerHandshake serverSignKey c (k, pk) kh ntfVRange = do
|
||||
getHandshake th >>= \case
|
||||
NtfClientHandshake {ntfVersion = v, keyHash}
|
||||
| keyHash /= kh ->
|
||||
throwError $ TEHandshake IDENTITY
|
||||
throwE $ TEHandshake IDENTITY
|
||||
| otherwise ->
|
||||
case compatibleVRange' ntfVersionRange v of
|
||||
Just (Compatible vr) -> pure $ ntfThHandleServer th v vr pk
|
||||
@@ -128,7 +128,7 @@ ntfClientHandshake c keyHash ntfVRange = do
|
||||
let th@THandle {params = THandleParams {sessionId}} = ntfTHandle c
|
||||
NtfServerHandshake {sessionId = sessId, ntfVersionRange, authPubKey = sk'} <- getHandshake th
|
||||
if sessionId /= sessId
|
||||
then throwError TEBadSession
|
||||
then throwE TEBadSession
|
||||
else case ntfVersionRange `compatibleVRange` ntfVRange of
|
||||
Just (Compatible vr) -> do
|
||||
ck_ <- forM sk' $ \signedKey -> liftEitherWith (const $ TEHandshake BAD_AUTH) $ do
|
||||
|
||||
@@ -7,6 +7,7 @@ import qualified Control.Exception as E
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.IO.Unlift
|
||||
import Control.Monad.Trans.Except
|
||||
import Data.Bifunctor (first)
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
@@ -114,11 +115,11 @@ catchAllErrors' err action handler = tryAllErrors' err action >>= either handler
|
||||
{-# INLINE catchAllErrors' #-}
|
||||
|
||||
catchThrow :: MonadUnliftIO m => ExceptT e m a -> (E.SomeException -> e) -> ExceptT e m a
|
||||
catchThrow action err = catchAllErrors err action throwError
|
||||
catchThrow action err = catchAllErrors err action throwE
|
||||
{-# INLINE catchThrow #-}
|
||||
|
||||
allFinally :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> ExceptT e m b -> ExceptT e m a
|
||||
allFinally err action final = tryAllErrors err action >>= \r -> final >> either throwError pure r
|
||||
allFinally err action final = tryAllErrors err action >>= \r -> final >> either throwE pure r
|
||||
{-# INLINE allFinally #-}
|
||||
|
||||
eitherToMaybe :: Either a b -> Maybe b
|
||||
@@ -149,7 +150,7 @@ safeDecodeUtf8 = decodeUtf8With onError
|
||||
onError _ _ = Just '?'
|
||||
|
||||
timeoutThrow :: MonadUnliftIO m => e -> Int -> ExceptT e m a -> ExceptT e m a
|
||||
timeoutThrow e ms action = ExceptT (sequence <$> (ms `timeout` runExceptT action)) >>= maybe (throwError e) pure
|
||||
timeoutThrow e ms action = ExceptT (sequence <$> (ms `timeout` runExceptT action)) >>= maybe (throwE e) pure
|
||||
|
||||
threadDelay' :: Int64 -> IO ()
|
||||
threadDelay' = loop
|
||||
|
||||
@@ -30,6 +30,7 @@ import Control.Logger.Simple
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.IO.Class
|
||||
import Control.Monad.Trans.Except
|
||||
import Crypto.Random (ChaChaDRG)
|
||||
import qualified Data.Aeson as J
|
||||
import Data.ByteString (ByteString)
|
||||
@@ -106,9 +107,9 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
|
||||
action <- liftIO $ runClient c r hostKeys
|
||||
-- wait for the port to make invitation
|
||||
portNum <- atomically $ readTMVar startedPort
|
||||
signedInv@RCSignedInvitation {invitation} <- maybe (throwError RCETLSStartFailed) (liftIO . mkInvitation hostKeys address) portNum
|
||||
signedInv@RCSignedInvitation {invitation} <- maybe (throwE RCETLSStartFailed) (liftIO . mkInvitation hostKeys address) portNum
|
||||
when multicast $ case knownHost of
|
||||
Nothing -> throwError RCENewController
|
||||
Nothing -> throwE RCENewController
|
||||
Just KnownHostPairing {hostDhPubKey} -> do
|
||||
ann <- liftIO . async . runExceptT $ announceRC drg 60 idPrivKey hostDhPubKey hostKeys invitation
|
||||
atomically $ putTMVar announcer ann
|
||||
@@ -117,7 +118,7 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
|
||||
findCtrlAddress :: ExceptT RCErrorType IO (NonEmpty RCCtrlAddress)
|
||||
findCtrlAddress = do
|
||||
found' <- liftIO $ getLocalAddress rcAddrPrefs_
|
||||
maybe (throwError RCENoLocalAddress) pure $ L.nonEmpty found'
|
||||
maybe (throwE RCENoLocalAddress) pure $ L.nonEmpty found'
|
||||
mkClient :: IO RCHClient_
|
||||
mkClient = do
|
||||
startedPort <- newEmptyTMVarIO
|
||||
@@ -211,10 +212,10 @@ prepareHostSession
|
||||
let sharedKey = C.dh' dhPubKey dhPrivKey
|
||||
helloBody <- liftEitherWith (const RCEDecrypt) $ C.cbDecrypt sharedKey nonce encBody
|
||||
hostHello@RCHostHello {v, ca, kem = kemPubKey} <- liftEitherWith RCESyntax $ J.eitherDecodeStrict helloBody
|
||||
unless (ca == tlsHostFingerprint) $ throwError RCEIdentity
|
||||
unless (ca == tlsHostFingerprint) $ throwE RCEIdentity
|
||||
(kemCiphertext, kemSharedKey) <- liftIO $ sntrup761Enc drg kemPubKey
|
||||
let hybridKey = kemHybridSecret dhPubKey dhPrivKey kemSharedKey
|
||||
unless (isCompatible v supportedRCPVRange) $ throwError RCEVersion
|
||||
unless (isCompatible v supportedRCPVRange) $ throwE RCEVersion
|
||||
let keys = HostSessKeys {hybridKey, idPrivKey, sessPrivKey}
|
||||
knownHost' <- updateKnownHost ca dhPubKey
|
||||
let ctrlHello = RCCtrlHello {}
|
||||
@@ -227,7 +228,7 @@ prepareHostSession
|
||||
updateKnownHost :: C.KeyHash -> C.PublicKeyX25519 -> ExceptT RCErrorType IO KnownHostPairing
|
||||
updateKnownHost ca hostDhPubKey = case knownHost_ of
|
||||
Just h -> do
|
||||
unless (hostFingerprint h == tlsHostFingerprint) . throwError $
|
||||
unless (hostFingerprint h == tlsHostFingerprint) . throwE $
|
||||
RCEInternal "TLS host CA is different from host pairing, should be caught in TLS handshake"
|
||||
pure (h :: KnownHostPairing) {hostDhPubKey}
|
||||
Nothing -> pure KnownHostPairing {hostFingerprint = ca, hostDhPubKey}
|
||||
@@ -257,7 +258,7 @@ connectRCCtrl drg (RCVerifiedInvitation inv@RCInvitation {ca, idkey}) pairing_ h
|
||||
pure RCCtrlPairing {caKey, caCert, ctrlFingerprint = ca, idPubKey = idkey, dhPrivKey, prevDhPrivKey = Nothing}
|
||||
updateCtrlPairing :: RCCtrlPairing -> ExceptT RCErrorType IO RCCtrlPairing
|
||||
updateCtrlPairing pairing@RCCtrlPairing {ctrlFingerprint, idPubKey, dhPrivKey = currDhPrivKey} = do
|
||||
unless (ca == ctrlFingerprint && idPubKey == idkey) $ throwError RCEIdentity
|
||||
unless (ca == ctrlFingerprint && idPubKey == idkey) $ throwE RCEIdentity
|
||||
(_, dhPrivKey) <- atomically $ C.generateKeyPair drg
|
||||
pure pairing {dhPrivKey, prevDhPrivKey = Just currDhPrivKey}
|
||||
|
||||
@@ -278,7 +279,7 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
|
||||
clientCredentials <-
|
||||
liftIO (genTLSCredentials drg caKey caCert) >>= \case
|
||||
TLS.Credentials (creds : _) -> pure $ Just creds
|
||||
_ -> throwError $ RCEInternal "genTLSCredentials must generate credentials"
|
||||
_ -> throwE $ RCEInternal "genTLSCredentials must generate credentials"
|
||||
let clientConfig = defaultTransportClientConfig {clientCredentials}
|
||||
ExceptT . runTransportClient clientConfig Nothing host (show port) (Just ca) $ \tls@TLS {tlsBuffer, tlsContext} -> runExceptT $ do
|
||||
-- pump socket to detect connection problems
|
||||
@@ -307,7 +308,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) >> throwE e
|
||||
|
||||
sendRCPacket :: Encoding a => TLS -> a -> ExceptT RCErrorType IO ()
|
||||
sendRCPacket tls pkt = do
|
||||
@@ -317,7 +318,7 @@ sendRCPacket tls pkt = do
|
||||
receiveRCPacket :: Encoding a => TLS -> ExceptT RCErrorType IO a
|
||||
receiveRCPacket tls = do
|
||||
b <- liftIO $ cGet tls xrcpBlockSize
|
||||
when (B.length b /= xrcpBlockSize) $ throwError RCEBlockSize
|
||||
when (B.length b /= xrcpBlockSize) $ throwE RCEBlockSize
|
||||
b' <- liftEitherWith (const RCEBlockSize) $ C.unPad b
|
||||
liftEitherWith RCESyntax $ smpDecode b'
|
||||
|
||||
@@ -329,7 +330,7 @@ prepareHostHello
|
||||
hostAppInfo = do
|
||||
logDebug "Preparing session"
|
||||
case compatibleVersion v supportedRCPVRange of
|
||||
Nothing -> throwError RCEVersion
|
||||
Nothing -> throwE RCEVersion
|
||||
Just (Compatible v') -> do
|
||||
nonce <- liftIO . atomically $ C.randomCbNonce drg
|
||||
(kemPubKey, kemPrivKey) <- liftIO $ sntrup761Keypair drg
|
||||
@@ -355,7 +356,7 @@ prepareCtrlSession
|
||||
pure CtrlSessKeys {hybridKey, idPubKey, sessPubKey = skey}
|
||||
RCCtrlEncError {nonce, encMessage} -> do
|
||||
message <- liftEitherWith (const RCEDecrypt) $ C.cbDecrypt sharedKey nonce encMessage
|
||||
throwError $ RCECtrlError $ T.unpack $ safeDecodeUtf8 message
|
||||
throwE $ RCECtrlError $ T.unpack $ safeDecodeUtf8 message
|
||||
|
||||
-- * Multicast discovery
|
||||
|
||||
@@ -382,7 +383,7 @@ discoverRCCtrl subscribers pairings =
|
||||
r@(_, RCVerifiedInvitation RCInvitation {host}) <- findRCCtrlPairing pairings encInvitation
|
||||
case source of
|
||||
SockAddrInet _ ha | THIPv4 (hostAddressToTuple ha) == host -> pure ()
|
||||
_ -> throwError RCEInvitation
|
||||
_ -> throwE RCEInvitation
|
||||
pure r
|
||||
where
|
||||
loop :: ExceptT RCErrorType IO a -> ExceptT RCErrorType IO a
|
||||
@@ -392,8 +393,8 @@ findRCCtrlPairing :: NonEmpty RCCtrlPairing -> RCEncInvitation -> ExceptT RCErro
|
||||
findRCCtrlPairing pairings RCEncInvitation {dhPubKey, nonce, encInvitation} = do
|
||||
(pairing, signedInvStr) <- liftEither $ decrypt (L.toList pairings)
|
||||
signedInv <- liftEitherWith RCESyntax $ strDecode signedInvStr
|
||||
inv@(RCVerifiedInvitation RCInvitation {dh = invDh}) <- maybe (throwError RCEInvitation) pure $ verifySignedInvitation signedInv
|
||||
unless (invDh == dhPubKey) $ throwError RCEInvitation
|
||||
inv@(RCVerifiedInvitation RCInvitation {dh = invDh}) <- maybe (throwE RCEInvitation) pure $ verifySignedInvitation signedInv
|
||||
unless (invDh == dhPubKey) $ throwE RCEInvitation
|
||||
pure (pairing, inv)
|
||||
where
|
||||
decrypt :: [RCCtrlPairing] -> Either RCErrorType (RCCtrlPairing, ByteString)
|
||||
@@ -433,7 +434,7 @@ rcEncryptBody drg hybridKey s = do
|
||||
rcDecryptBody :: KEMHybridSecret -> C.CbNonce -> LazyByteString -> ExceptT RCErrorType IO LazyByteString
|
||||
rcDecryptBody hybridKey nonce ct = do
|
||||
let len = LB.length ct - 16
|
||||
when (len < 0) $ throwError RCEDecrypt
|
||||
when (len < 0) $ throwE RCEDecrypt
|
||||
(ok, s) <- liftEitherWith (const RCEDecrypt) $ LC.kcbDecryptTailTag hybridKey nonce len ct
|
||||
unless ok $ throwError RCEDecrypt
|
||||
unless ok $ throwE RCEDecrypt
|
||||
pure s
|
||||
@@ -2134,7 +2134,7 @@ testSwitchAsync servers = do
|
||||
withB = withAgent 2 agentCfg servers testDB2
|
||||
|
||||
withAgent :: HasCallStack => Int -> AgentConfig -> InitialAgentServers -> FilePath -> (HasCallStack => AgentClient -> IO a) -> IO a
|
||||
withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disposeAgentClient
|
||||
withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) (\a -> disposeAgentClient a >> threadDelay 100000)
|
||||
|
||||
sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO ()
|
||||
sessionSubscribe withC connIds a =
|
||||
|
||||
Reference in new issue
Block a user