xftp: agent API for xftp commands and events (#684)

* xftp: agent API for xftp commands and events

* fix tests

* fix tests 2

* xftp: update agent send api

* update API to make temp path optional

* revert tmp path changes (fixes send)

---------

Co-authored-by: spacedandy <8711996+spaced4ndy@users.noreply.github.com>
This commit is contained in:
Evgeny Poberezkinandspacedandy authored and GitHub committed 2023-03-13 09:04:23 +00:00
1 parent 3fe1f3c822
commit ddc2da8fe4
15 files changed
+218 -172

No files matched your search

+28 -29
View File
@@ -29,6 +29,7 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
import Data.Bifunctor (first)
import Data.ByteString (ByteString)
import qualified Data.ByteString.Base64.URL as U
import qualified Data.ByteString.Char8 as B
import qualified Data.ByteString.Lazy.Char8 as LB
import Data.Int (Int64)
import Data.List (isSuffixOf, partition)
@@ -36,7 +37,7 @@ import Data.List.NonEmpty (nonEmpty)
import qualified Data.List.NonEmpty as L
import Simplex.FileTransfer.Client.Main (CLIError, SendOptions (..), cliSendFile)
import Simplex.FileTransfer.Description
import Simplex.FileTransfer.Protocol (FileParty (..))
import Simplex.FileTransfer.Protocol (FileParty (..), FilePartyI)
import Simplex.FileTransfer.Transport (XFTPRcvChunkSpec (..))
import Simplex.FileTransfer.Types
import Simplex.FileTransfer.Util (removePath, uniqueCombine)
@@ -44,25 +45,26 @@ import Simplex.Messaging.Agent.Client
import Simplex.Messaging.Agent.Env.SQLite
import Simplex.Messaging.Agent.Protocol
import Simplex.Messaging.Agent.RetryInterval
import Simplex.Messaging.Agent.Store
import Simplex.Messaging.Agent.Store.SQLite
import qualified Simplex.Messaging.Crypto.Lazy as LC
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (XFTPServer)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Util (tshow)
import Simplex.Messaging.Util (liftIOEither, tshow)
import System.FilePath (takeFileName, (</>))
import UnliftIO
import UnliftIO.Concurrent
import UnliftIO.Directory
import qualified UnliftIO.Exception as E
receiveFile :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> FilePath -> m RcvFileId
receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) xftpPath = do
receiveFile :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe FilePath -> m RcvFileId
receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) xftpWorkPath = do
g <- asks idsDrg
encPath <- uniqueCombine xftpPath "xftp.encrypted"
workPath <- maybe getTemporaryDirectory pure xftpWorkPath
encPath <- uniqueCombine workPath "xftp.encrypted"
createDirectory encPath
fId <- withStore c $ \db -> createRcvFile db g userId fd xftpPath encPath
fId <- withStore c $ \db -> createRcvFile db g userId fd workPath encPath
forM_ chunks downloadChunk
pure fId
where
@@ -186,9 +188,10 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do
pure $ path : ps
getChunkPaths (RcvFileChunk {chunkTmpPath = Nothing} : _cs) =
throwError $ INTERNAL "no chunk path"
-- TODO refactor with decrypt in CLI, streaming decryption
decrypt :: Int64 -> [FilePath] -> m FilePath
decrypt encSize chunkPaths = do
lazyChunks <- readChunks chunkPaths
lazyChunks <- liftIO $ readChunks chunkPaths
(authOk, f) <- liftEither . first cryptoError $ LC.sbDecryptTailTag key nonce (encSize - authTagSize) lazyChunks
let (fileHdr, f') = LB.splitAt 1024 f
-- withFile encPath ReadMode $ \r -> do
@@ -205,30 +208,25 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do
removeFile path
throwError $ INTERNAL "Error decrypting file: incorrect auth tag"
pure path
readChunks :: [FilePath] -> m LB.ByteString
readChunks =
foldM
( \s path -> do
chunk <- liftIO $ LB.readFile path
pure $ s <> chunk
)
LB.empty
readChunks :: [FilePath] -> IO LB.ByteString
readChunks = foldM (\s path -> (s <>) <$> LB.readFile path) ""
sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId
sendFileExperimental AgentClient {subQ} _userId numRecipients xftpPath filePath = do
sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> Maybe FilePath -> m SndFileId
sendFileExperimental AgentClient {subQ} _userId filePath numRecipients xftpWorkPath = do
g <- asks idsDrg
sndFileEntityId <- liftIO $ randomId g 12
void $ forkIO $ sendCLI sndFileEntityId
pure sndFileEntityId
sndFileId <- liftIO $ randomId g 12
void $ forkIO $ sendCLI sndFileId
pure sndFileId
where
randomId :: TVar ChaChaDRG -> Int -> IO ByteString
randomId gVar n = U.encode <$> (atomically . stateTVar gVar $ randomBytesGenerate n)
sendCLI :: SndFileId -> m ()
sendCLI sndFileEntityId = do
sendCLI sndFileId = do
let fileName = takeFileName filePath
outputDir <- uniqueCombine xftpPath (fileName <> ".descr")
workPath <- maybe getTemporaryDirectory pure xftpWorkPath
outputDir <- uniqueCombine workPath $ fileName <> ".descr"
createDirectory outputDir
let tempPath = xftpPath </> "snd"
let tempPath = workPath </> "snd"
createDirectoryIfMissing False tempPath
let sendOptions =
SendOptions
@@ -242,20 +240,21 @@ sendFileExperimental AgentClient {subQ} _userId numRecipients xftpPath filePath
}
liftCLI $ cliSendFile sendOptions
(sndDescr, rcvDescrs) <- readDescrs outputDir fileName
notify sndFileEntityId $ SFDONE sndDescr rcvDescrs
notify sndFileId $ SFDONE sndDescr rcvDescrs
liftCLI :: ExceptT CLIError IO () -> m ()
liftCLI = either (throwError . INTERNAL . show) pure <=< liftIO . runExceptT
readDescrs :: FilePath -> FilePath -> m (String, [String])
readDescrs :: FilePath -> FilePath -> m (ValidFileDescription 'FSender, [ValidFileDescription 'FRecipient])
readDescrs outDir fileName = do
let descrDir = outDir </> (fileName <> ".xftp")
files <- listDirectory descrDir
let (sdFiles, rdFiles) = partition ("snd.xftp.private" `isSuffixOf`) files
sdFile = maybe "" (\l -> descrDir </> L.head l) (nonEmpty sdFiles)
rdFiles' = map (descrDir </>) rdFiles
-- TODO map files to contents
pure (sdFile, rdFiles')
(,) <$> readDescr sdFile <*> mapM readDescr rdFiles'
readDescr :: FilePartyI p => FilePath -> m (ValidFileDescription p)
readDescr f = liftIOEither $ first INTERNAL . strDecode <$> B.readFile f
notify :: forall e. AEntityI e => SndFileId -> ACommand 'Agent e -> m ()
notify sndFileEntityId cmd = atomically $ writeTBQueue subQ ("", sndFileEntityId, APC (sAEntity @e) cmd)
notify sndFileId cmd = atomically $ writeTBQueue subQ ("", sndFileId, APC (sAEntity @e) cmd)
-- _sendFile :: AgentMonad m => AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId
_sendFile :: AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId
+3 -4
View File
@@ -458,7 +458,7 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
throwError $ CLIError "Error decrypting file: incorrect auth tag"
pure path
readChunks :: [FilePath] -> IO LB.ByteString
readChunks = foldM (\s path -> (s <>) <$> LB.readFile path) LB.empty
readChunks = foldM (\s path -> (s <>) <$> LB.readFile path) ""
{-# NOINLINE readChunks #-}
getFilePath :: String -> ExceptT CLIError IO FilePath
getFilePath name =
@@ -523,9 +523,8 @@ strEnc :: StrEncoding a => a -> String
strEnc = B.unpack . strEncode
getFileDescription :: FilePath -> ExceptT CLIError IO AValidFileDescription
getFileDescription path = do
fd <- ExceptT $ first (CLIError . ("Failed to parse file description: " <>)) . strDecode <$> B.readFile path
liftEither . first CLIError $ validateFileDescription fd
getFileDescription path =
ExceptT $ first (CLIError . ("Failed to parse file description: " <>)) . strDecode <$> B.readFile path
getFileDescription' :: FilePartyI p => FilePath -> ExceptT CLIError IO (ValidFileDescription p)
getFileDescription' path =
+15
View File
@@ -27,6 +27,7 @@ module Simplex.FileTransfer.Description
validateFileDescription,
groupReplicasByServer,
replicaServer,
fdSeparator,
kb,
mb,
gb,
@@ -75,6 +76,7 @@ data FileDescription (p :: FileParty) = FileDescription
data AFileDescription = forall p. FilePartyI p => AFD (FileDescription p)
newtype ValidFileDescription p = ValidFD (FileDescription p)
deriving (Eq, Show)
pattern ValidFileDescription :: FileDescription p -> ValidFileDescription p
pattern ValidFileDescription fd = ValidFD fd
@@ -83,6 +85,9 @@ pattern ValidFileDescription fd = ValidFD fd
data AValidFileDescription = forall p. FilePartyI p => AVFD (ValidFileDescription p)
fdSeparator :: IsString s => s
fdSeparator = "################################\n"
newtype FileDigest = FileDigest {unFileDigest :: ByteString}
deriving (Eq, Show)
@@ -170,6 +175,16 @@ data FileServerReplica = FileServerReplica
}
deriving (Show)
instance FilePartyI p => StrEncoding (ValidFileDescription p) where
strEncode (ValidFD fd) = strEncode fd
strDecode s = strDecode s >>= (\(AVFD fd) -> checkParty fd)
strP = strDecode <$?> A.takeByteString
instance StrEncoding AValidFileDescription where
strEncode (AVFD fd) = strEncode fd
strDecode = validateFileDescription <=< strDecode
strP = strDecode <$?> A.takeByteString
instance FilePartyI p => StrEncoding (FileDescription p) where
strEncode = Y.encode . encodeFileDescription
strDecode s = strDecode s >>= (\(AFD fd) -> checkParty fd)
+3 -3
View File
@@ -339,12 +339,12 @@ toggleConnectionNtfs :: AgentErrorMonad m => AgentClient -> ConnId -> Bool -> m
toggleConnectionNtfs c = withAgentEnv c .: toggleConnectionNtfs' c
-- | Receive XFTP file
xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> FilePath -> m RcvFileId
xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe FilePath -> m RcvFileId
xftpReceiveFile c = withAgentEnv c .:. receiveFile c
-- | Send XFTP file
xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId
xftpSendFile c userId n xftpPath filePath = withAgentEnv c $ sendFileExperimental c userId n xftpPath filePath
xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> FilePath -> Int -> Maybe FilePath -> m SndFileId
xftpSendFile c = withAgentEnv c .:: sendFileExperimental c
-- | Activate operations
activateAgent :: AgentErrorMonad m => AgentClient -> m ()
+2 -2
View File
@@ -441,8 +441,8 @@ reconnectSMPClient c tSess@(_, srv, _) =
let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
liftIO $ mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
mapM_ (throwError . snd) $ listToMaybe tempErrs
notifySub :: ConnId -> ACommand 'Agent 'AEConn -> IO ()
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC SAEConn cmd)
notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
getNtfServerClient :: forall m. AgentMonad m => AgentClient -> NtfTransportSession -> m NtfClient
getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = do
@@ -44,7 +44,6 @@ import Simplex.FileTransfer.Client (XFTPClientConfig (..), defaultXFTPClientConf
import Simplex.FileTransfer.Types (DBSndFileId)
import Simplex.Messaging.Agent.Protocol
import Simplex.Messaging.Agent.RetryInterval
import Simplex.Messaging.Agent.Store (UserId)
import Simplex.Messaging.Agent.Store.SQLite
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
import Simplex.Messaging.Client
+1 -1
View File
@@ -14,7 +14,7 @@ createLock :: STM Lock
createLock = newEmptyTMVar
{-# INLINE createLock #-}
withLock :: MonadUnliftIO m => TMVar String -> String -> m a -> m a
withLock :: MonadUnliftIO m => Lock -> String -> m a -> m a
withLock lock name =
E.bracket_
(atomically $ putTMVar lock name)
+34 -18
View File
@@ -104,6 +104,7 @@ module Simplex.Messaging.Agent.Protocol
MsgIntegrity (..),
MsgErrorType (..),
QueueStatus (..),
UserId,
ACorrId,
AgentMsgId,
NotificationsMode (..),
@@ -163,7 +164,8 @@ import Database.SQLite.Simple.FromField
import Database.SQLite.Simple.ToField
import GHC.Generics (Generic)
import Generic.Random (genericArbitraryU)
import Simplex.FileTransfer.Protocol (XFTPErrorType)
import Simplex.FileTransfer.Description
import Simplex.FileTransfer.Protocol (FileParty (..), XFTPErrorType)
import Simplex.Messaging.Agent.QueryString
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Crypto.Ratchet (E2ERatchetParams, E2ERatchetParamsUri)
@@ -224,6 +226,8 @@ type ATransmission p = (ACorrId, EntityId, APartyCmd p)
-- | SMP agent protocol transmission or transmission error.
type ATransmissionOrError p = (ACorrId, EntityId, Either AgentErrorType (APartyCmd p))
type UserId = Int64
type ACorrId = ByteString
-- | SMP agent protocol participants.
@@ -311,8 +315,8 @@ data ACommand (p :: AParty) (e :: AEntity) where
END :: ACommand Agent AEConn
CONNECT :: AProtocolType -> TransportHost -> ACommand Agent AENone
DISCONNECT :: AProtocolType -> TransportHost -> ACommand Agent AENone
DOWN :: SMPServer -> [ConnId] -> ACommand Agent AEConn
UP :: SMPServer -> [ConnId] -> ACommand Agent AEConn
DOWN :: SMPServer -> [ConnId] -> ACommand Agent AENone
UP :: SMPServer -> [ConnId] -> ACommand Agent AENone
SWITCH :: QueueDirection -> SwitchPhase -> ConnectionStats -> ACommand Agent AEConn
SEND :: MsgFlags -> MsgBody -> ACommand Client AEConn
MID :: AgentMsgId -> ACommand Agent AEConn
@@ -335,7 +339,8 @@ data ACommand (p :: AParty) (e :: AEntity) where
RFPROG :: Int -> Int -> ACommand Agent AERcvFile
RFDONE :: FilePath -> ACommand Agent AERcvFile
RFERR :: AgentErrorType -> ACommand Agent AERcvFile
SFDONE :: String -> [String] -> ACommand Agent AESndFile
SFPROG :: Int -> Int -> ACommand Agent AESndFile
SFDONE :: ValidFileDescription 'FSender -> [ValidFileDescription 'FRecipient] -> ACommand Agent AESndFile
deriving instance Eq (ACommand p e)
@@ -367,8 +372,8 @@ data ACommandTag (p :: AParty) (e :: AEntity) where
END_ :: ACommandTag Agent AEConn
CONNECT_ :: ACommandTag Agent AENone
DISCONNECT_ :: ACommandTag Agent AENone
DOWN_ :: ACommandTag Agent AEConn
UP_ :: ACommandTag Agent AEConn
DOWN_ :: ACommandTag Agent AENone
UP_ :: ACommandTag Agent AENone
SWITCH_ :: ACommandTag Agent AEConn
SEND_ :: ACommandTag Client AEConn
MID_ :: ACommandTag Agent AEConn
@@ -391,6 +396,7 @@ data ACommandTag (p :: AParty) (e :: AEntity) where
RFDONE_ :: ACommandTag Agent AERcvFile
RFPROG_ :: ACommandTag Agent AERcvFile
RFERR_ :: ACommandTag Agent AERcvFile
SFPROG_ :: ACommandTag Agent AESndFile
SFDONE_ :: ACommandTag Agent AESndFile
deriving instance Eq (ACommandTag p e)
@@ -439,6 +445,7 @@ aCommandTag = \case
RFPROG {} -> RFPROG_
RFDONE {} -> RFDONE_
RFERR {} -> RFERR_
SFPROG {} -> SFPROG_
SFDONE {} -> SFDONE_
data QueueDirection = QDRcv | QDSnd
@@ -1305,10 +1312,10 @@ instance StrEncoding ACmdTag where
"CON" -> ct CON_
"SUB" -> t SUB_
"END" -> ct END_
"CONNECT" -> at SAENone CONNECT_
"DISCONNECT" -> at SAENone DISCONNECT_
"DOWN" -> ct DOWN_
"UP" -> ct UP_
"CONNECT" -> nt CONNECT_
"DISCONNECT" -> nt DISCONNECT_
"DOWN" -> nt DOWN_
"UP" -> nt UP_
"SWITCH" -> ct SWITCH_
"SEND" -> t SEND_
"MID" -> ct MID_
@@ -1321,20 +1328,23 @@ instance StrEncoding ACmdTag where
"DEL" -> t DEL_
"DEL_RCVQ" -> ct DEL_RCVQ_
"DEL_CONN" -> ct DEL_CONN_
"DEL_USER" -> at SAENone DEL_USER_
"DEL_USER" -> nt DEL_USER_
"CHK" -> t CHK_
"STAT" -> ct STAT_
"OK" -> ct OK_
"ERR" -> ct ERR_
"SUSPENDED" -> at SAENone SUSPENDED_
"SUSPENDED" -> nt SUSPENDED_
"RFPROG" -> at SAERcvFile RFPROG_
"RFDONE" -> at SAERcvFile RFDONE_
"RFERR" -> at SAERcvFile RFERR_
"SFPROG" -> at SAESndFile SFPROG_
"SFDONE" -> at SAESndFile SFDONE_
_ -> fail "bad ACmdTag"
where
t = pure . ACmdTag SClient SAEConn
at e = pure . ACmdTag SAgent e
ct = at SAEConn
nt = at SAENone
instance APartyI p => StrEncoding (APartyCmdTag p) where
strEncode (APCT _ cmd) = strEncode cmd
@@ -1379,6 +1389,7 @@ instance (APartyI p, AEntityI e) => StrEncoding (ACommandTag p e) where
RFPROG_ -> "RFPROG"
RFDONE_ -> "RFDONE"
RFERR_ -> "RFERR"
SFPROG_ -> "SFPROG"
SFDONE_ -> "SFDONE"
strP = (\(APCT _ t) -> checkEntity t) <$?> strP
@@ -1438,14 +1449,19 @@ commandP binaryP =
RFPROG_ -> s (RFPROG <$> A.decimal <* A.space <*> A.decimal)
RFDONE_ -> s (RFDONE <$> strP)
RFERR_ -> s (RFERR <$> strP)
SFDONE_ -> s (SFDONE <$> strP <*> rcvDescrs)
SFPROG_ -> s (SFPROG <$> A.decimal <* A.space <*> A.decimal)
SFDONE_ -> s (sfDone . safeDecodeUtf8 <$?> binaryP)
where
s :: Parser a -> Parser a
s p = A.space *> p
connections :: Parser [ConnId]
connections = strP `A.sepBy'` A.char ','
rcvDescrs :: Parser [String]
rcvDescrs = strP `A.sepBy'` A.char ',' -- TODO consider separator
sfDone :: Text -> Either String (ACommand 'Agent 'AESndFile)
sfDone t =
let ds = T.splitOn fdSeparator t
in case ds of
[] -> Left "no sender file description"
sd : rds -> SFDONE <$> strDecode (encodeUtf8 sd) <*> mapM (strDecode . encodeUtf8) rds
msgMetaP = do
integrity <- strP
recipient <- " R=" *> partyMeta A.decimal
@@ -1497,7 +1513,8 @@ serializeCommand = \case
RFPROG rcvd total -> s (RFPROG_, rcvd, total)
RFDONE fPath -> s (RFDONE_, fPath)
RFERR e -> s (RFERR_, e)
SFDONE sd rds -> B.unwords [s SFDONE_, s sd, rcvDescrs rds]
SFPROG sent total -> s (SFPROG_, sent, total)
SFDONE sd rds -> B.unwords [s SFDONE_, serializeBinary (sfDone sd rds)]
where
s :: StrEncoding a => a -> ByteString
s = strEncode
@@ -1505,8 +1522,7 @@ serializeCommand = \case
showTs = B.pack . formatISO8601Millis
connections :: [ConnId] -> ByteString
connections = B.intercalate "," . map strEncode
rcvDescrs :: [String] -> ByteString
rcvDescrs = B.intercalate "," . map strEncode -- TODO consider separator
sfDone sd rds = B.intercalate fdSeparator $ strEncode sd : map strEncode rds
serializeMsgMeta :: MsgMeta -> ByteString
serializeMsgMeta MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} =
B.unwords
-2
View File
@@ -264,8 +264,6 @@ data PendingCommand = PendingCommand
data AgentCmdType = ACClient | ACInternal
type UserId = Int64
instance StrEncoding AgentCmdType where
strEncode = \case
ACClient -> "CLIENT"
+2 -2
View File
@@ -16,8 +16,8 @@ import Control.Concurrent.STM
import qualified Data.Map.Strict as M
import Data.Set (Set)
import qualified Data.Set as S
import Simplex.Messaging.Agent.Protocol (ConnId)
import Simplex.Messaging.Agent.Store (RcvQueue (..), UserId)
import Simplex.Messaging.Agent.Protocol (ConnId, UserId)
import Simplex.Messaging.Agent.Store (RcvQueue (..))
import Simplex.Messaging.Protocol (RecipientId, SMPServer)
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM