mirror of
https://github.com/simplex-chat/simplex-chat.git
synced 2024-12-17 17:20:21 +01:00
remote protocol (#3225)
* draft remote protocol types and external api * types (it compiles) * add error * move remote controller from http to remote host client protocol * refactor (doesnt compile) * fix compile * Connect remote session * WIP: wire in remote protocol * add commands and events * cleanup * fix desktop shutdown * prepare for testing remote files * Add file IO * update simplexmq to master with http2 to 4.1.4 * use json transcoder * update simplexmq * collapse RemoteHostSession states * fold RemoteHello back into the protocol command move http-command-response-http wrapper to protocol * use sendRemoteCommand with optional attachments use streaming request/response * ditch lazy body streaming * fix formatting * put body builder/processor closer together * wrap handleRemoteCommand around sending files * handle ChatError's too * remove binary, use 32-bit encoding for JSON bodies * enable tests * refactor * refactor request handling * return ChatError * Flatten remote host --------- Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>
This commit is contained in:
Alexander Bondarenko
committed by
GitHub
GitHub
parent
0444367002
commit
0d1a080a6e
+186
-329
@@ -5,7 +5,6 @@
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE PatternSynonyms #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE TupleSections #-}
|
||||
|
||||
@@ -20,148 +19,136 @@ import Control.Monad.IO.Class
|
||||
import Control.Monad.Reader (asks)
|
||||
import Control.Monad.STM (retry)
|
||||
import Crypto.Random (getRandomBytes)
|
||||
import Data.Aeson ((.=))
|
||||
import qualified Data.Aeson as J
|
||||
import qualified Data.Aeson.Key as JK
|
||||
import qualified Data.Aeson.KeyMap as JM
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import qualified Data.Binary.Builder as Binary
|
||||
import Data.ByteString (ByteString, hPut)
|
||||
import Data.ByteString (ByteString)
|
||||
import qualified Data.ByteString.Base64.URL as B64U
|
||||
import Data.ByteString.Builder (Builder)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Int (Int64)
|
||||
import Data.List.NonEmpty (NonEmpty (..))
|
||||
import qualified Data.Map.Strict as M
|
||||
import Data.Maybe (fromMaybe)
|
||||
import Data.Text (Text)
|
||||
import qualified Data.Text as T
|
||||
import Data.Text.Encoding (decodeUtf8, encodeUtf8)
|
||||
import qualified Network.HTTP.Types as HTTP
|
||||
import qualified Network.HTTP.Types.Status as Status
|
||||
import qualified Network.HTTP2.Client as HC
|
||||
import qualified Network.HTTP2.Server as HS
|
||||
import Data.Text.Encoding (encodeUtf8)
|
||||
import Data.Word (Word32)
|
||||
import Network.HTTP2.Server (responseStreaming)
|
||||
import qualified Network.HTTP.Types as N
|
||||
import Network.Socket (SockAddr (..), hostAddressToTuple)
|
||||
import Simplex.Chat.Controller
|
||||
import Simplex.Chat.Messages (AChatItem (..), CIFile (..), CIFileStatus (..), ChatItem (..), chatNameStr)
|
||||
import Simplex.Chat.Messages.CIContent (MsgDirection (..), SMsgDirection (..))
|
||||
import qualified Simplex.Chat.Remote.Discovery as Discovery
|
||||
import Simplex.Chat.Remote.Protocol
|
||||
import Simplex.Chat.Remote.Types
|
||||
import Simplex.Chat.Store.Files (getRcvFileTransfer)
|
||||
import Simplex.Chat.Store.Profiles (getUser)
|
||||
import Simplex.Chat.Store.Remote
|
||||
import Simplex.Chat.Store.Shared (StoreError (..))
|
||||
import Simplex.Chat.Types
|
||||
import Simplex.FileTransfer.Util (uniqueCombine)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Crypto.File (CryptoFile (..))
|
||||
import Simplex.Messaging.Encoding.String (StrEncoding (..))
|
||||
import Simplex.Messaging.Parsers (pattern SingleFieldJSONTag, pattern TaggedObjectJSONData, pattern TaggedObjectJSONTag)
|
||||
import Simplex.Messaging.Transport.HTTP2.Server (HTTP2Request (..))
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
import Simplex.Messaging.Transport.Client (TransportHost (..))
|
||||
import Simplex.Messaging.Transport.Credentials (genCredentials, tlsCredentials)
|
||||
import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..), defaultHTTP2BufferSize)
|
||||
import Simplex.Messaging.Transport.HTTP2.Client (HTTP2Client, HTTP2ClientError, HTTP2Response (..))
|
||||
import qualified Simplex.Messaging.Transport.HTTP2.Client as HTTP2
|
||||
import qualified Simplex.Messaging.Transport.HTTP2.Server as HTTP2
|
||||
import Simplex.Messaging.Util (bshow, ifM, liftEitherError, liftEitherWith, tshow, ($>>=))
|
||||
import System.FilePath (isPathSeparator, takeFileName, (</>))
|
||||
import Simplex.Messaging.Transport.HTTP2.File (hSendFile)
|
||||
import Simplex.Messaging.Util (ifM, liftEitherError, liftEitherWith, liftError, liftIOEither, tryAllErrors, tshow, ($>>=))
|
||||
import System.FilePath ((</>))
|
||||
import UnliftIO
|
||||
import UnliftIO.Directory (createDirectoryIfMissing, getFileSize)
|
||||
|
||||
-- * Desktop side
|
||||
|
||||
getRemoteHostSession :: ChatMonad m => RemoteHostId -> m RemoteHostSession
|
||||
getRemoteHostSession rhId = chatReadVar remoteHostSessions >>= maybe err pure . M.lookup rhId
|
||||
where
|
||||
err = throwError $ ChatErrorRemoteHost rhId RHMissing
|
||||
getRemoteHostSession rhId = withRemoteHostSession rhId $ \_ s -> pure $ Right s
|
||||
|
||||
checkNoRemoteHostSession :: ChatMonad m => RemoteHostId -> m ()
|
||||
checkNoRemoteHostSession rhId = chatReadVar remoteHostSessions >>= maybe (pure ()) err . M.lookup rhId
|
||||
withRemoteHostSession :: ChatMonad m => RemoteHostId -> (TM.TMap RemoteHostId RemoteHostSession -> RemoteHostSession -> STM (Either ChatError a)) -> m a
|
||||
withRemoteHostSession rhId = withRemoteHostSession_ rhId missing
|
||||
where
|
||||
err _ = throwError $ ChatErrorRemoteHost rhId RHBusy
|
||||
missing _ = pure . Left $ ChatErrorRemoteHost rhId RHMissing
|
||||
|
||||
withNoRemoteHostSession :: ChatMonad m => RemoteHostId -> (TM.TMap RemoteHostId RemoteHostSession -> STM (Either ChatError a)) -> m a
|
||||
withNoRemoteHostSession rhId action = withRemoteHostSession_ rhId action busy
|
||||
where
|
||||
busy _ _ = pure . Left $ ChatErrorRemoteHost rhId RHBusy
|
||||
|
||||
-- | Atomically process controller state wrt. specific remote host session
|
||||
withRemoteHostSession_ :: ChatMonad m => RemoteHostId -> (TM.TMap RemoteHostId RemoteHostSession -> STM (Either ChatError a)) -> (TM.TMap RemoteHostId RemoteHostSession -> RemoteHostSession -> STM (Either ChatError a)) -> m a
|
||||
withRemoteHostSession_ rhId missing present = do
|
||||
sessions <- asks remoteHostSessions
|
||||
liftIOEither . atomically $ TM.lookup rhId sessions >>= maybe (missing sessions) (present sessions)
|
||||
|
||||
startRemoteHost :: ChatMonad m => RemoteHostId -> m ()
|
||||
startRemoteHost rhId = do
|
||||
checkNoRemoteHostSession rhId
|
||||
rh <- withStore (`getRemoteHost` rhId)
|
||||
announcer <- async $ do
|
||||
finished <- newTVarIO False
|
||||
http <- start rh finished `onChatError` cleanup finished
|
||||
run rh finished http
|
||||
chatModifyVar remoteHostSessions $ M.insert rhId RemoteHostSessionStarting {announcer}
|
||||
tasks <- startRemoteHostSession rh
|
||||
logInfo $ "Remote host session starting for " <> tshow rhId
|
||||
asyncRegistered tasks $ run rh tasks `catchAny` \err -> do
|
||||
logError $ "Remote host session startup failed for " <> tshow rhId <> ": " <> tshow err
|
||||
cancelTasks tasks
|
||||
chatModifyVar remoteHostSessions $ M.delete rhId
|
||||
throwError $ fromMaybe (mkChatError err) $ fromException err
|
||||
-- logInfo $ "Remote host session starting for " <> tshow rhId
|
||||
where
|
||||
cleanup finished = do
|
||||
logInfo "Remote host http2 client fininshed"
|
||||
atomically $ writeTVar finished True
|
||||
-- TODO why this is not an error?
|
||||
M.lookup rhId <$> chatReadVar remoteHostSessions >>= \case
|
||||
Nothing -> logInfo $ "Session already closed for remote host " <> tshow rhId
|
||||
Just _ -> closeRemoteHostSession rhId >> toView (CRRemoteHostStopped rhId)
|
||||
start rh@RemoteHost {storePath, caKey, caCert} finished = do
|
||||
let parent = (C.signatureKeyPair caKey, caCert)
|
||||
sessionCreds <- liftIO $ genCredentials (Just parent) (0, 24) "Session"
|
||||
let (fingerprint, credentials) = tlsCredentials $ sessionCreds :| [parent]
|
||||
u <- askUnliftIO
|
||||
ctrlClient <- liftHTTP2 $ Discovery.announceRevHTTP2 fingerprint credentials $ unliftIO u (cleanup finished) -- >>= \case
|
||||
chatModifyVar remoteHostSessions $ M.insert rhId RemoteHostSessionStarted {storePath, ctrlClient}
|
||||
chatWriteVar currentRemoteHost $ Just rhId
|
||||
HTTP2Response {respBody = HTTP2Body {bodyHead}} <- sendHello ctrlClient
|
||||
run :: ChatMonad m => RemoteHost -> Tasks -> m ()
|
||||
run rh@RemoteHost {storePath} tasks = do
|
||||
(fingerprint, credentials) <- liftIO $ genSessionCredentials rh
|
||||
cleanupIO <- toIO $ do
|
||||
logNote $ "Remote host session stopping for " <> tshow rhId
|
||||
cancelTasks tasks -- cancel our tasks anyway
|
||||
chatModifyVar currentRemoteHost $ \cur -> if cur == Just rhId then Nothing else cur -- only wipe the closing RH
|
||||
withRemoteHostSession rhId $ \sessions _ -> Right <$> TM.delete rhId sessions
|
||||
toView (CRRemoteHostStopped rhId) -- only signal "stopped" when the session is unregistered cleanly
|
||||
-- block until some client is connected or an error happens
|
||||
logInfo $ "Remote host session connecting for " <> tshow rhId
|
||||
httpClient <- liftEitherError (ChatErrorRemoteCtrl . RCEHTTP2Error . show) $ Discovery.announceRevHTTP2 tasks fingerprint credentials cleanupIO
|
||||
logInfo $ "Remote host session connected for " <> tshow rhId
|
||||
rcName <- chatReadVar localDeviceName
|
||||
-- TODO what sets session active?
|
||||
toView CRRemoteHostConnected {remoteHost = remoteHostInfo rh True rcName}
|
||||
pure ctrlClient
|
||||
run RemoteHost {storePath} finished ctrlClient = do
|
||||
-- test connection and establish a protocol layer
|
||||
remoteHostClient <- liftRH rhId $ createRemoteHostClient httpClient rcName
|
||||
-- set up message polling
|
||||
oq <- asks outputQ
|
||||
let toViewRemote = atomically . writeTBQueue oq . (Nothing,Just rhId,)
|
||||
-- TODO remove REST
|
||||
void . async $ pollRemote finished ctrlClient "/recv" $ handleFile >=> toViewRemote
|
||||
asyncRegistered tasks . forever $ do
|
||||
liftRH rhId (remoteRecv remoteHostClient 1000000) >>= mapM_ (atomically . writeTBQueue oq . (Nothing,Just rhId,))
|
||||
-- update session state
|
||||
logInfo $ "Remote host session started for " <> tshow rhId
|
||||
chatModifyVar remoteHostSessions $ M.adjust (\rhs -> rhs {remoteHostClient = Just remoteHostClient}) rhId
|
||||
chatWriteVar currentRemoteHost $ Just rhId
|
||||
toView $ CRRemoteHostConnected RemoteHostInfo
|
||||
{ remoteHostId = rhId,
|
||||
storePath = storePath,
|
||||
displayName = remoteDeviceName remoteHostClient,
|
||||
remoteCtrlOOB = RemoteCtrlOOB {fingerprint, displayName=rcName},
|
||||
sessionActive = True
|
||||
}
|
||||
|
||||
genSessionCredentials RemoteHost {caKey, caCert} = do
|
||||
sessionCreds <- genCredentials (Just parent) (0, 24) "Session"
|
||||
pure . tlsCredentials $ sessionCreds :| [parent]
|
||||
where
|
||||
-- TODO move to view / terminal
|
||||
handleFile = \case
|
||||
cr@CRRcvFileComplete {user, chatItem = AChatItem c SMDRcv i ci@ChatItem {file = Just ciFile@CIFile {fileStatus = CIFSRcvComplete}}} -> do
|
||||
maybe cr update <$> handleRcvFileComplete ctrlClient storePath user ciFile
|
||||
where
|
||||
update localFile = cr {chatItem = AChatItem c SMDRcv i ci {file = Just localFile}}
|
||||
cr -> pure cr
|
||||
parent = (C.signatureKeyPair caKey, caCert)
|
||||
|
||||
sendHello :: ChatMonad m => HTTP2Client -> m HTTP2Response
|
||||
sendHello http = liftHTTP2 $ HTTP2.sendRequestDirect http req Nothing
|
||||
where
|
||||
req = HC.requestNoBody "GET" "/" mempty
|
||||
|
||||
-- TODO how (on what condition) it would stop polling?
|
||||
-- TODO add JSON translation
|
||||
pollRemote :: ChatMonad m => TVar Bool -> HTTP2Client -> ByteString -> (ChatResponse -> m ()) -> m ()
|
||||
pollRemote finished http path action = loop `catchChatError` \e -> action (CRChatError Nothing e) >> loop
|
||||
where
|
||||
loop = do
|
||||
-- TODO this will never load full body
|
||||
HTTP2Response {respBody = HTTP2Body {bodyHead}} <- liftHTTP2 $ HTTP2.sendRequestDirect http req Nothing
|
||||
json <- liftEitherWith (ChatErrorRemoteCtrl . RCEInvalidResponse) $ J.eitherDecodeStrict' bodyHead -- of
|
||||
action json
|
||||
readTVarIO finished >>= (`unless` loop)
|
||||
req = HC.requestNoBody "GET" path mempty
|
||||
-- | Atomically check/register session and prepare its task list
|
||||
startRemoteHostSession :: ChatMonad m => RemoteHost -> m Tasks
|
||||
startRemoteHostSession RemoteHost {remoteHostId, storePath} = withNoRemoteHostSession remoteHostId $ \sessions -> do
|
||||
remoteHostTasks <- newTVar []
|
||||
TM.insert remoteHostId RemoteHostSession {remoteHostTasks, storePath, remoteHostClient = Nothing} sessions
|
||||
pure $ Right remoteHostTasks
|
||||
|
||||
closeRemoteHostSession :: ChatMonad m => RemoteHostId -> m ()
|
||||
closeRemoteHostSession remoteHostId = do
|
||||
session <- getRemoteHostSession remoteHostId
|
||||
logInfo $ "Closing remote host session for " <> tshow remoteHostId
|
||||
liftIO $ cancelRemoteHostSession session
|
||||
chatWriteVar currentRemoteHost Nothing
|
||||
chatModifyVar remoteHostSessions $ M.delete remoteHostId
|
||||
closeRemoteHostSession rhId = do
|
||||
logNote $ "Closing remote host session for " <> tshow rhId
|
||||
chatModifyVar currentRemoteHost $ \cur -> if cur == Just rhId then Nothing else cur -- only wipe the closing RH
|
||||
session <- withRemoteHostSession rhId $ \sessions rhs -> Right rhs <$ TM.delete rhId sessions
|
||||
cancelRemoteHostSession session
|
||||
|
||||
cancelRemoteHostSession :: MonadUnliftIO m => RemoteHostSession -> m ()
|
||||
cancelRemoteHostSession = \case
|
||||
RemoteHostSessionStarting {announcer} -> cancel announcer
|
||||
RemoteHostSessionStarted {ctrlClient} -> liftIO $ HTTP2.closeHTTP2Client ctrlClient
|
||||
cancelRemoteHostSession RemoteHostSession {remoteHostTasks, remoteHostClient} = do
|
||||
cancelTasks remoteHostTasks
|
||||
mapM_ closeRemoteHostClient remoteHostClient
|
||||
|
||||
createRemoteHost :: ChatMonad m => m RemoteHostInfo
|
||||
createRemoteHost = do
|
||||
let rhName = "TODO" -- you don't have remote host name here, it will be passed from remote host
|
||||
((_, caKey), caCert) <- liftIO $ genCredentials Nothing (-25, 24 * 365) rhName
|
||||
((_, caKey), caCert) <- liftIO $ genCredentials Nothing (-25, 24 * 365) "Host"
|
||||
storePath <- liftIO randomStorePath
|
||||
remoteHostId <- withStore' $ \db -> insertRemoteHost db storePath rhName caKey caCert
|
||||
rcName <- chatReadVar localDeviceName
|
||||
let remoteCtrlOOB = RemoteCtrlOOB {fingerprint = C.certificateFingerprint caCert, displayName = rcName}
|
||||
pure RemoteHostInfo {remoteHostId, storePath, displayName = rhName, remoteCtrlOOB, sessionActive = False}
|
||||
let remoteName = "" -- will be passed from remote host in hello
|
||||
remoteHostId <- withStore' $ \db -> insertRemoteHost db storePath remoteName caKey caCert
|
||||
localName <- chatReadVar localDeviceName
|
||||
let remoteCtrlOOB = RemoteCtrlOOB {fingerprint = C.certificateFingerprint caCert, displayName = localName}
|
||||
pure RemoteHostInfo {remoteHostId, storePath, displayName = remoteName, remoteCtrlOOB, sessionActive = False}
|
||||
|
||||
-- | Generate a random 16-char filepath without / in it by using base64url encoding.
|
||||
randomStorePath :: IO FilePath
|
||||
@@ -191,241 +178,111 @@ deleteRemoteHost rhId = do
|
||||
Nothing -> logWarn "Local file store not available while deleting remote host"
|
||||
withStore' (`deleteRemoteHostRecord` rhId)
|
||||
|
||||
processRemoteCommand :: ChatMonad m => RemoteHostSession -> (ByteString, ChatCommand) -> m ChatResponse
|
||||
processRemoteCommand RemoteHostSessionStarting {} _ = pure $ chatCmdError Nothing "remote command sent before session started"
|
||||
processRemoteCommand RemoteHostSessionStarted {ctrlClient} (s, cmd) =
|
||||
uploadFile cmd >>= relayCommand ctrlClient
|
||||
where
|
||||
fileCmd cmdPfx cn hostPath = utf8String $ unwords [cmdPfx, chatNameStr cn, hostPath]
|
||||
uploadFile = \case
|
||||
SendFile cn ctrlPath -> fileCmd "/file" cn <$> storeRemoteFile ctrlClient ctrlPath
|
||||
SendImage cn ctrlPath -> fileCmd "/image" cn <$> storeRemoteFile ctrlClient ctrlPath
|
||||
-- TODO APISendMessage should only be used with host path already, and UI has to upload file first.
|
||||
-- The problem is that we cannot have different file names in host and controller, because it simply won't be able to show files.
|
||||
-- So we need to ask the host to store files BEFORE storing them in the app storage and use host names in the command and to store the file locally if it has to be shown,
|
||||
-- or don't even store it if it's not image/video.
|
||||
-- The current approach won't work.
|
||||
-- It also does not account for local file encryption.
|
||||
-- Also, local file encryption setting should be tracked in the controller, as otherwise host won't be able to decide what to do having received the upload command.
|
||||
APISendMessage {composedMessage = cm@ComposedMessage {fileSource = Just CryptoFile {filePath = ctrlPath, cryptoArgs}}} -> do
|
||||
hostPath <- storeRemoteFile ctrlClient ctrlPath
|
||||
let cm' = cm {fileSource = Just CryptoFile {filePath = hostPath, cryptoArgs}} :: ComposedMessage
|
||||
-- TODO we shouldn't manipulate JSON like that
|
||||
pure $ B.takeWhile (/= '{') s <> B.toStrict (J.encode cm')
|
||||
_ -> pure s
|
||||
processRemoteCommand :: ChatMonad m => RemoteHostId -> RemoteHostSession -> ByteString -> m ChatResponse
|
||||
processRemoteCommand remoteHostId RemoteHostSession {remoteHostClient = Just rhc} s = liftRH remoteHostId $ remoteSend rhc s
|
||||
processRemoteCommand _ _ _ = pure $ chatCmdError Nothing "remote command sent before session started"
|
||||
|
||||
relayCommand :: ChatMonad m => HTTP2Client -> ByteString -> m ChatResponse
|
||||
relayCommand http s = do
|
||||
-- TODO ExceptT
|
||||
let timeout' = Nothing
|
||||
HTTP2Response {respBody = HTTP2Body {bodyHead}} <-
|
||||
liftHTTP2 $ HTTP2.sendRequestDirect http req timeout'
|
||||
-- TODO: large JSONs can overflow into buffered chunks
|
||||
json <- liftEitherWith (ChatErrorRemoteCtrl . RCEInvalidResponse) $ J.eitherDecodeStrict' bodyHead
|
||||
case J.fromJSON $ toTaggedJSON json of
|
||||
J.Error e -> err $ show e
|
||||
J.Success cr -> pure cr
|
||||
where
|
||||
err = pure . CRChatError Nothing . ChatErrorRemoteCtrl . RCEInvalidResponse
|
||||
toTaggedJSON :: J.Value -> J.Value
|
||||
toTaggedJSON = id -- owsf2tagged TODO: get from RemoteHost
|
||||
req = HC.requestBuilder "POST" "/send" mempty (Binary.fromByteString s)
|
||||
liftRH :: ChatMonad m => RemoteHostId -> ExceptT RemoteProtocolError IO a -> m a
|
||||
liftRH rhId = liftError (ChatErrorRemoteHost rhId . RHProtocolError)
|
||||
|
||||
-- TODO fileName is just metadata that does not determine the actual file location for UI, or whether it is encrypted or not
|
||||
-- fileSource is the actual file location (with information whether it is locally encrypted)
|
||||
handleRcvFileComplete :: ChatMonad m => HTTP2Client -> FilePath -> User -> CIFile 'MDRcv -> m (Maybe (CIFile 'MDRcv))
|
||||
handleRcvFileComplete http storePath remoteUser f@CIFile {fileId, fileName} =
|
||||
chatReadVar filesFolder >>= \case
|
||||
Just baseDir -> do
|
||||
let hostStore = baseDir </> storePath
|
||||
createDirectoryIfMissing True hostStore
|
||||
-- TODO the problem here is that the name may turn out to be different and nothing will work
|
||||
-- file processing seems to work "accidentally", not "by design"
|
||||
localPath <- uniqueCombine hostStore fileName
|
||||
fetchRemoteFile http remoteUser fileId localPath
|
||||
pure $ Just (f {fileName = localPath} :: CIFile 'MDRcv)
|
||||
-- TODO below will not work with CLI, it should store file to download folder when not specified
|
||||
-- It should not load all files when received, instead it should only load files received with /fr commands
|
||||
Nothing -> Nothing <$ logError "Local file store not available while fetching remote file"
|
||||
-- * Mobile side
|
||||
|
||||
-- | Convert swift single-field sum encoding into tagged/discriminator-field
|
||||
owsf2tagged :: J.Value -> J.Value
|
||||
owsf2tagged = fst . convert
|
||||
where
|
||||
convert val = case val of
|
||||
J.Object o
|
||||
| JM.size o == 2 ->
|
||||
case JM.toList o of
|
||||
[OwsfTag, o'] -> tagged o'
|
||||
[o', OwsfTag] -> tagged o'
|
||||
_ -> props
|
||||
| otherwise -> props
|
||||
where
|
||||
props = (J.Object $ fmap owsf2tagged o, False)
|
||||
J.Array a -> (J.Array $ fmap owsf2tagged a, False)
|
||||
_ -> (val, False)
|
||||
-- `tagged` converts the pair of single-field object encoding to tagged encoding.
|
||||
-- It sets innerTag returned by `convert` to True to prevent the tag being overwritten.
|
||||
tagged (k, v) = (J.Object pairs, True)
|
||||
where
|
||||
(v', innerTag) = convert v
|
||||
pairs = case v' of
|
||||
-- `innerTag` indicates that internal object already has tag,
|
||||
-- so the current tag cannot be inserted into it.
|
||||
J.Object o
|
||||
| innerTag -> pair
|
||||
| otherwise -> JM.insert TaggedObjectJSONTag tag o
|
||||
_ -> pair
|
||||
tag = J.String $ JK.toText k
|
||||
pair = JM.fromList [TaggedObjectJSONTag .= tag, TaggedObjectJSONData .= v']
|
||||
|
||||
pattern OwsfTag :: (JK.Key, J.Value)
|
||||
pattern OwsfTag = (SingleFieldJSONTag, J.Bool True)
|
||||
|
||||
storeRemoteFile :: ChatMonad m => HTTP2Client -> FilePath -> m FilePath
|
||||
storeRemoteFile http localFile = do
|
||||
fileSize <- liftIO $ fromIntegral <$> getFileSize localFile
|
||||
-- TODO configure timeout
|
||||
let timeout' = Nothing
|
||||
r@HTTP2Response {respBody = HTTP2Body {bodyHead}} <-
|
||||
liftHTTP2 $ HTTP2.sendRequestDirect http (req fileSize) timeout'
|
||||
responseStatusOK r
|
||||
-- TODO what if response doesn't fit in the head?
|
||||
-- it'll be solved when processing moved to POST with Command/Response types
|
||||
pure $ B.unpack bodyHead
|
||||
where
|
||||
-- TODO local file encryption?
|
||||
uri = "/store?" <> HTTP.renderSimpleQuery False [("file_name", utf8String $ takeFileName localFile)]
|
||||
req size = HC.requestFile "PUT" uri mempty (HC.FileSpec localFile 0 size)
|
||||
|
||||
liftHTTP2 :: ChatMonad m => IO (Either HTTP2ClientError a) -> m a
|
||||
liftHTTP2 = liftEitherError $ ChatErrorRemoteCtrl . RCEHTTP2Error . show
|
||||
|
||||
responseStatusOK :: ChatMonad m => HTTP2Response -> m ()
|
||||
responseStatusOK HTTP2Response {response} = do
|
||||
let s = HC.responseStatus response
|
||||
unless (s == Just Status.ok200) $
|
||||
throwError $ ChatErrorRemoteCtrl $ RCEHTTP2RespStatus $ Status.statusCode <$> s
|
||||
|
||||
fetchRemoteFile :: ChatMonad m => HTTP2Client -> User -> Int64 -> FilePath -> m ()
|
||||
fetchRemoteFile http User {userId = remoteUserId} remoteFileId localPath = do
|
||||
r@HTTP2Response {respBody} <- liftHTTP2 $ HTTP2.sendRequestDirect http req Nothing
|
||||
responseStatusOK r
|
||||
writeBodyToFile localPath respBody
|
||||
where
|
||||
req = HC.requestNoBody "GET" path mempty
|
||||
path = "/fetch?" <> HTTP.renderSimpleQuery False [("user_id", bshow remoteUserId), ("file_id", bshow remoteFileId)]
|
||||
|
||||
-- XXX: extract to Transport.HTTP2 ?
|
||||
writeBodyToFile :: MonadUnliftIO m => FilePath -> HTTP2Body -> m ()
|
||||
writeBodyToFile path HTTP2Body {bodyHead, bodySize, bodyPart} = do
|
||||
logInfo $ "Receiving " <> tshow bodySize <> " bytes to " <> tshow path
|
||||
liftIO . withFile path WriteMode $ \h -> do
|
||||
hPut h bodyHead
|
||||
mapM_ (hPutBodyChunks h) bodyPart
|
||||
|
||||
hPutBodyChunks :: Handle -> (Int -> IO ByteString) -> IO ()
|
||||
hPutBodyChunks h getChunk = do
|
||||
chunk <- getChunk defaultHTTP2BufferSize
|
||||
unless (B.null chunk) $ do
|
||||
hPut h chunk
|
||||
hPutBodyChunks h getChunk
|
||||
|
||||
-- TODO command/response pattern, remove REST conventions
|
||||
processControllerRequest :: forall m. ChatMonad m => (ByteString -> m ChatResponse) -> HTTP2.HTTP2Request -> m ()
|
||||
processControllerRequest execChatCommand HTTP2.HTTP2Request {request, reqBody, sendResponse} = do
|
||||
logDebug $ "Remote controller request: " <> tshow (method <> " " <> path)
|
||||
res <- tryChatError $ case (method, ps) of
|
||||
("GET", []) -> getHello
|
||||
("POST", ["send"]) -> sendCommand
|
||||
("GET", ["recv"]) -> recvMessage
|
||||
("PUT", ["store"]) -> storeFile
|
||||
("GET", ["fetch"]) -> fetchFile
|
||||
unexpected -> respondWith Status.badRequest400 $ "unexpected method/path: " <> Binary.putStringUtf8 (show unexpected)
|
||||
case res of
|
||||
Left e -> logError $ "Error handling remote controller request: (" <> tshow (method <> " " <> path) <> "): " <> tshow e
|
||||
Right () -> logDebug $ "Remote controller request: " <> tshow (method <> " " <> path) <> " OK"
|
||||
where
|
||||
method = fromMaybe "" $ HS.requestMethod request
|
||||
path = fromMaybe "/" $ HS.requestPath request
|
||||
(ps, query) = HTTP.decodePath path
|
||||
getHello = respond "OK"
|
||||
sendCommand = execChatCommand (bodyHead reqBody) >>= respondJSON
|
||||
recvMessage =
|
||||
chatReadVar remoteCtrlSession >>= \case
|
||||
Nothing -> respondWith Status.internalServerError500 "session not active"
|
||||
Just rcs -> atomically (readTBQueue $ remoteOutputQ rcs) >>= respondJSON
|
||||
-- TODO liftEither storeFileQuery
|
||||
storeFile = case storeFileQuery of
|
||||
Left err -> respondWith Status.badRequest400 (Binary.putStringUtf8 err)
|
||||
Right fileName -> do
|
||||
baseDir <- fromMaybe "." <$> chatReadVar filesFolder
|
||||
localPath <- uniqueCombine baseDir fileName
|
||||
logDebug $ "Storing controller file to " <> tshow (baseDir, localPath)
|
||||
writeBodyToFile localPath reqBody
|
||||
let storeRelative = takeFileName localPath
|
||||
respond $ Binary.putStringUtf8 storeRelative
|
||||
where
|
||||
storeFileQuery = parseField "file_name" $ A.many1 (A.satisfy $ not . isPathSeparator)
|
||||
-- TODO move to ExceptT monad, catch errors in one place, convert errors to responses
|
||||
fetchFile = case fetchFileQuery of
|
||||
Left err -> respondWith Status.badRequest400 (Binary.putStringUtf8 err)
|
||||
Right (userId, fileId) -> do
|
||||
logInfo $ "Fetching file " <> tshow fileId <> " from user " <> tshow userId
|
||||
x <- withStore' $ \db -> runExceptT $ do
|
||||
user <- getUser db userId
|
||||
getRcvFileTransfer db user fileId
|
||||
-- TODO this error handling is very ad-hoc, there is no separation between Chat errors and responses
|
||||
case x of
|
||||
Right RcvFileTransfer {fileStatus = RFSComplete RcvFileInfo {filePath}} -> do
|
||||
baseDir <- fromMaybe "." <$> chatReadVar filesFolder
|
||||
let fullPath = baseDir </> filePath
|
||||
size <- fromInteger <$> getFileSize fullPath
|
||||
liftIO . sendResponse . HS.responseFile Status.ok200 mempty $ HS.FileSpec fullPath 0 size
|
||||
Right _ -> respondWith Status.internalServerError500 "The requested file is not complete"
|
||||
Left SEUserNotFound {} -> respondWith Status.notFound404 "User not found"
|
||||
Left SERcvFileNotFound {} -> respondWith Status.notFound404 "File not found"
|
||||
_ -> respondWith Status.internalServerError500 "Store error"
|
||||
where
|
||||
fetchFileQuery =
|
||||
(,)
|
||||
<$> parseField "user_id" A.decimal
|
||||
<*> parseField "file_id" A.decimal
|
||||
|
||||
parseField :: ByteString -> A.Parser a -> Either String a
|
||||
parseField field p = maybe (Left $ "missing " <> B.unpack field) (A.parseOnly $ p <* A.endOfInput) (join $ lookup field query)
|
||||
|
||||
respondJSON :: (J.ToJSON a) => a -> m ()
|
||||
respondJSON = respond . Binary.fromLazyByteString . J.encode
|
||||
|
||||
respond = respondWith Status.ok200
|
||||
respondWith status = liftIO . sendResponse . HS.responseBuilder status []
|
||||
|
||||
-- * ChatRequest handlers
|
||||
|
||||
startRemoteCtrl :: ChatMonad m => (ByteString -> m ChatResponse) -> m ()
|
||||
startRemoteCtrl :: forall m . ChatMonad m => (ByteString -> m ChatResponse) -> m ()
|
||||
startRemoteCtrl execChatCommand = do
|
||||
checkNoRemoteCtrlSession
|
||||
logInfo "Starting remote host"
|
||||
checkNoRemoteCtrlSession -- tiny race with the final @chatWriteVar@ until the setup finishes and supervisor spawned
|
||||
discovered <- newTVarIO mempty
|
||||
discoverer <- async $ discoverRemoteCtrls discovered -- TODO extract to a controller service singleton
|
||||
size <- asks $ tbqSize . config
|
||||
remoteOutputQ <- newTBQueueIO size
|
||||
discovered <- newTVarIO mempty
|
||||
discoverer <- async $ discoverRemoteCtrls discovered
|
||||
accepted <- newEmptyTMVarIO
|
||||
supervisor <- async $ runSupervisor discovered accepted
|
||||
supervisor <- async $ runHost discovered accepted $ handleRemoteCommand execChatCommand remoteOutputQ
|
||||
chatWriteVar remoteCtrlSession $ Just RemoteCtrlSession {discoverer, supervisor, hostServer = Nothing, discovered, accepted, remoteOutputQ}
|
||||
|
||||
-- | Track remote host lifecycle in controller session state and signal UI on its progress
|
||||
runHost :: ChatMonad m => TM.TMap C.KeyHash TransportHost -> TMVar RemoteCtrlId -> (HTTP2Request -> m ()) -> m ()
|
||||
runHost discovered accepted handleHttp = do
|
||||
remoteCtrlId <- atomically (readTMVar accepted) -- wait for ???
|
||||
rc@RemoteCtrl {fingerprint} <- withStore (`getRemoteCtrl` remoteCtrlId)
|
||||
source <- atomically $ TM.lookup fingerprint discovered >>= maybe retry pure -- wait for location of the matching fingerprint
|
||||
toView $ CRRemoteCtrlConnecting $ remoteCtrlInfo rc False
|
||||
atomically $ writeTVar discovered mempty -- flush unused sources
|
||||
server <- async $ Discovery.connectRevHTTP2 source fingerprint handleHttp -- spawn server for remote protocol commands
|
||||
chatModifyVar remoteCtrlSession $ fmap $ \s -> s {hostServer = Just server}
|
||||
toView $ CRRemoteCtrlConnected $ remoteCtrlInfo rc True
|
||||
_ <- waitCatch server -- wait for the server to finish
|
||||
chatWriteVar remoteCtrlSession Nothing
|
||||
toView CRRemoteCtrlStopped
|
||||
|
||||
handleRemoteCommand :: forall m . ChatMonad m => (ByteString -> m ChatResponse) -> TBQueue ChatResponse -> HTTP2Request -> m ()
|
||||
handleRemoteCommand execChatCommand remoteOutputQ HTTP2Request {request, reqBody, sendResponse} = do
|
||||
logDebug "handleRemoteCommand"
|
||||
liftRC (tryRemoteError parseRequest) >>= \case
|
||||
Right (getNext, rc) -> processCommand getNext rc `catchAny` (reply . RRProtocolError . RPEException . tshow)
|
||||
Left e -> reply $ RRProtocolError e
|
||||
where
|
||||
runSupervisor discovered accepted = do
|
||||
remoteCtrlId <- atomically (readTMVar accepted)
|
||||
rc@RemoteCtrl {fingerprint} <- withStore (`getRemoteCtrl` remoteCtrlId)
|
||||
source <- atomically $ TM.lookup fingerprint discovered >>= maybe retry pure
|
||||
toView $ CRRemoteCtrlConnecting $ remoteCtrlInfo rc False
|
||||
atomically $ writeTVar discovered mempty -- flush unused sources
|
||||
server <- async $ Discovery.connectRevHTTP2 source fingerprint (processControllerRequest execChatCommand)
|
||||
chatModifyVar remoteCtrlSession $ fmap $ \s -> s {hostServer = Just server}
|
||||
toView $ CRRemoteCtrlConnected $ remoteCtrlInfo rc True
|
||||
_ <- waitCatch server
|
||||
chatWriteVar remoteCtrlSession Nothing
|
||||
toView CRRemoteCtrlStopped
|
||||
parseRequest :: ExceptT RemoteProtocolError IO (GetChunk, RemoteCommand)
|
||||
parseRequest = do
|
||||
(header, getNext) <- parseHTTP2Body request reqBody
|
||||
(getNext,) <$> liftEitherWith (RPEInvalidJSON . T.pack) (J.eitherDecodeStrict' header)
|
||||
processCommand :: GetChunk -> RemoteCommand -> m ()
|
||||
processCommand getNext = \case
|
||||
RCHello {deviceName = desktopName} -> handleHello desktopName >>= reply
|
||||
RCSend {command} -> handleSend execChatCommand command >>= reply
|
||||
RCRecv {wait = time} -> handleRecv time remoteOutputQ >>= reply
|
||||
RCStoreFile {fileSize, encrypt} -> handleStoreFile fileSize encrypt getNext >>= reply
|
||||
RCGetFile {filePath} -> handleGetFile filePath replyWith
|
||||
reply :: RemoteResponse -> m ()
|
||||
reply = (`replyWith` \_ -> pure ())
|
||||
replyWith :: Respond m
|
||||
replyWith rr attach =
|
||||
liftIO . sendResponse . responseStreaming N.status200 [] $ \send flush -> do
|
||||
send $ sizePrefixedEncode rr
|
||||
attach send
|
||||
flush
|
||||
|
||||
type GetChunk = Int -> IO ByteString
|
||||
|
||||
type SendChunk = Builder -> IO ()
|
||||
|
||||
type Respond m = RemoteResponse -> (SendChunk -> IO ()) -> m ()
|
||||
|
||||
liftRC :: ChatMonad m => ExceptT RemoteProtocolError IO a -> m a
|
||||
liftRC = liftError (ChatErrorRemoteCtrl . RCEProtocolError)
|
||||
|
||||
tryRemoteError :: ExceptT RemoteProtocolError IO a -> ExceptT RemoteProtocolError IO (Either RemoteProtocolError a)
|
||||
tryRemoteError = tryAllErrors (RPEException . tshow)
|
||||
{-# INLINE tryRemoteError #-}
|
||||
|
||||
handleHello :: ChatMonad m => Text -> m RemoteResponse
|
||||
handleHello desktopName = do
|
||||
logInfo $ "Hello from " <> tshow desktopName
|
||||
mobileName <- chatReadVar localDeviceName
|
||||
pure RRHello {encoding = localEncoding, deviceName = mobileName}
|
||||
|
||||
handleSend :: ChatMonad m => (ByteString -> m ChatResponse) -> Text -> m RemoteResponse
|
||||
handleSend execChatCommand command = do
|
||||
logDebug $ "Send: " <> tshow command
|
||||
-- execChatCommand checks for remote-allowed commands
|
||||
-- convert errors thrown in ChatMonad into error responses to prevent aborting the protocol wrapper
|
||||
RRChatResponse <$> execChatCommand (encodeUtf8 command) `catchError` (pure . CRChatError Nothing)
|
||||
|
||||
handleRecv :: MonadUnliftIO m => Int -> TBQueue ChatResponse -> m RemoteResponse
|
||||
handleRecv time events = do
|
||||
logDebug $ "Recv: " <> tshow time
|
||||
RRChatEvent <$> (timeout time . atomically $ readTBQueue events)
|
||||
|
||||
handleStoreFile :: ChatMonad m => Word32 -> Maybe Bool -> GetChunk -> m RemoteResponse
|
||||
handleStoreFile _fileSize _encrypt _getNext = error "TODO" <$ logError "TODO: handleStoreFile"
|
||||
|
||||
handleGetFile :: ChatMonad m => FilePath -> Respond m -> m ()
|
||||
handleGetFile path reply = do
|
||||
logDebug $ "GetFile: " <> tshow path
|
||||
withFile path ReadMode $ \h -> do
|
||||
fileSize' <- hFileSize h
|
||||
when (fileSize' > toInteger (maxBound :: Word32)) $ throwIO RPEFileTooLarge
|
||||
let fileSize = fromInteger fileSize'
|
||||
reply RRFile {fileSize} $ \send -> hSendFile h send fileSize
|
||||
|
||||
-- TODO the problem with this code was that it wasn't clear where the recursion can happen,
|
||||
-- by splitting receiving and processing to two functions it becomes clear
|
||||
|
||||
Reference in New Issue
Block a user