core: http transport for remote session (#3178)

* Wire some of the session endpoints * Start sending remote commands * Expand remote controller - Fix queues for pumping to remote - Add 3-way test - WIP: Add TTY wrapper for remote hosts - Stop remote controller w/o ids to match starting * Fix view events * Drop notifications, add message test * refactor, receive test * hunt down stray asyncs * Take discovery sockets in brackets --------- Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>
2026-10-23 10:43:05 +00:00 · 2023-10-07 14:23:24 +01:00
1 parent 3ac342782b
commit 91561da351
@@ -4,12 +4,15 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}

 module Simplex.Chat.Remote where

+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Class
+import Control.Monad.Reader (asks)
 import Control.Monad.STM (retry)
 import Crypto.Random (getRandomBytes)
 import qualified Data.Aeson as J
@@ -18,9 +21,13 @@ import Data.ByteString (ByteString)
 import qualified Data.ByteString.Base64.URL as B64U
 import qualified Data.ByteString.Char8 as B
 import Data.List.NonEmpty (NonEmpty (..))
+import Data.Maybe (fromMaybe)
 import qualified Data.Map.Strict as M
+import qualified Data.Text as T
 import qualified Network.HTTP.Types as HTTP
+import qualified Network.HTTP.Types.Status as Status
 import qualified Network.HTTP2.Client as HTTP2Client
+import qualified Network.HTTP2.Server as HTTP2Server
 import Network.Socket (SockAddr (..), hostAddressToTuple)
 import Simplex.Chat.Controller
 import qualified Simplex.Chat.Remote.Discovery as Discovery
@@ -36,7 +43,7 @@ import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..))
 import Simplex.Messaging.Transport.HTTP2.Client (HTTP2Client)
 import qualified Simplex.Messaging.Transport.HTTP2.Client as HTTP2
 import qualified Simplex.Messaging.Transport.HTTP2.Server as HTTP2
-import Simplex.Messaging.Util (bshow)
+import Simplex.Messaging.Util (bshow, ifM, tshow)
 import System.Directory (getFileSize)
 import UnliftIO

@@ -54,32 +61,67 @@ withRemoteHost remoteHostId action =

 startRemoteHost :: (ChatMonad m) => RemoteHostId -> m ChatResponse
 startRemoteHost remoteHostId = do
-  M.lookup remoteHostId <$> chatReadVar remoteHostSessions >>= \case
+  asks remoteHostSessions >>= atomically . TM.lookup remoteHostId >>= \case
    Just _ -> throwError $ ChatErrorRemoteHost remoteHostId RHBusy
-    Nothing -> withRemoteHost remoteHostId run
-  where
-    run RemoteHost {storePath, caKey, caCert} = do
-      announcer <- async $ do
-        cleanup <- toIO $ closeRemoteHostSession remoteHostId >>= toView
-        let parent = (C.signatureKeyPair caKey, caCert)
-        sessionCreds <- liftIO $ genCredentials (Just parent) (0, 24) "Session"
-        let (fingerprint, credentials) = tlsCredentials $ sessionCreds :| [parent]
-        Discovery.announceRevHTTP2 cleanup fingerprint credentials >>= \case
-          Left todo'err -> liftIO cleanup -- TODO: log error
-          Right ctrlClient -> do
-            chatModifyVar remoteHostSessions $ M.insert remoteHostId RemoteHostSessionStarted {storePath, ctrlClient}
-            -- TODO: start streaming outputQ
-            toView CRRemoteHostConnected {remoteHostId}
+    Nothing -> withRemoteHost remoteHostId $ \rh -> do
+      announcer <- async $ run rh
      chatModifyVar remoteHostSessions $ M.insert remoteHostId RemoteHostSessionStarting {announcer}
      pure CRRemoteHostStarted {remoteHostId}
+  where
+    cleanup finished = do
+      logInfo "Remote host http2 client fininshed"
+      atomically $ writeTVar finished True
+      closeRemoteHostSession remoteHostId >>= toView
+    run RemoteHost {storePath, caKey, caCert} = do
+      finished <- newTVarIO False
+      let parent = (C.signatureKeyPair caKey, caCert)
+      sessionCreds <- liftIO $ genCredentials (Just parent) (0, 24) "Session"
+      let (fingerprint, credentials) = tlsCredentials $ sessionCreds :| [parent]
+      Discovery.announceRevHTTP2 (cleanup finished) fingerprint credentials >>= \case
+        Left h2ce -> do
+          logError $ "Failed to set up remote host connection: " <> tshow h2ce
+          cleanup finished
+        Right ctrlClient -> do
+          chatModifyVar remoteHostSessions $ M.insert remoteHostId RemoteHostSessionStarted {storePath, ctrlClient}
+          chatWriteVar currentRemoteHost $ Just remoteHostId
+          sendHello ctrlClient >>= \case
+            Left h2ce -> do
+              logError $ "Failed to send initial remote host request: " <> tshow h2ce
+              cleanup finished
+            Right HTTP2.HTTP2Response {respBody = HTTP2Body {bodyHead}} -> do
+              logDebug $ "Got initial from remote host: " <> tshow bodyHead
+              _ <- asks outputQ >>= async . pollRemote finished ctrlClient "/recv" (Nothing, Just remoteHostId,)
+              toView CRRemoteHostConnected {remoteHostId}
+
+sendHello :: (ChatMonad m) => HTTP2Client -> m (Either HTTP2.HTTP2ClientError HTTP2.HTTP2Response)
+sendHello http = liftIO (HTTP2.sendRequestDirect http req Nothing)
+  where
+    req = HTTP2Client.requestNoBody "GET" "/" mempty
+
+pollRemote :: (ChatMonad m, J.FromJSON a) => TVar Bool -> HTTP2Client -> ByteString -> (a -> b) -> TBQueue b -> m ()
+pollRemote finished http path f queue = loop
+  where
+    loop = do
+      liftIO (HTTP2.sendRequestDirect http req Nothing) >>= \case
+        Left e -> logError $ "pollRemote: " <> tshow (path, e)
+        Right HTTP2.HTTP2Response {respBody = HTTP2Body {bodyHead}} ->
+          case J.eitherDecodeStrict' bodyHead of
+            Left e -> logError $ "pollRemote/decode: " <> tshow (path, e)
+            Right o -> atomically $ writeTBQueue queue (f o)
+      readTVarIO finished >>= (`unless` loop)
+    req = HTTP2Client.requestNoBody "GET" path mempty

 closeRemoteHostSession :: (ChatMonad m) => RemoteHostId -> m ChatResponse
 closeRemoteHostSession remoteHostId = withRemoteHostSession remoteHostId $ \session -> do
-  case session of
-    RemoteHostSessionStarting {announcer} -> cancel announcer
-    RemoteHostSessionStarted {ctrlClient} -> liftIO (HTTP2.closeHTTP2Client ctrlClient)
+  liftIO $ cancelRemoteHostSession session
+  chatWriteVar currentRemoteHost Nothing
  chatModifyVar remoteHostSessions $ M.delete remoteHostId
-  pure CRRemoteHostStopped { remoteHostId }
+  pure CRRemoteHostStopped {remoteHostId}
+
+cancelRemoteHostSession :: MonadUnliftIO m => RemoteHostSession -> m ()
+cancelRemoteHostSession = \case
+  RemoteHostSessionStarting {announcer} -> cancel announcer
+  RemoteHostSessionStarted {ctrlClient} -> liftIO $ HTTP2.closeHTTP2Client ctrlClient

 createRemoteHost :: (ChatMonad m) => m ChatResponse
 createRemoteHost = do
@@ -87,10 +129,7 @@ createRemoteHost = do
  ((_, caKey), caCert) <- liftIO $ genCredentials Nothing (-25, 24 * 365) displayName
  storePath <- liftIO randomStorePath
  remoteHostId <- withStore' $ \db -> insertRemoteHost db storePath displayName caKey caCert
-  let oobData =
-        RemoteCtrlOOB
-          { caFingerprint = C.certificateFingerprint caCert
-          }
+  let oobData = RemoteCtrlOOB {caFingerprint = C.certificateFingerprint caCert}
  pure CRRemoteHostCreated {remoteHostId, oobData}

 -- | Generate a random 16-char filepath without / in it by using base64url encoding.
@@ -113,41 +152,40 @@ deleteRemoteHost remoteHostId = withRemoteHost remoteHostId $ \rh -> do
  pure CRRemoteHostDeleted {remoteHostId}

 processRemoteCommand :: (ChatMonad m) => RemoteHostSession -> (ByteString, ChatCommand) -> m ChatResponse
-processRemoteCommand RemoteHostSessionStarting {} _ = error "TODO: sending remote commands before session started"
-processRemoteCommand RemoteHostSessionStarted {ctrlClient} (s, cmd) =
+processRemoteCommand RemoteHostSessionStarting {} _ = pure . CRChatError Nothing . ChatError $ CEInternalError "sending remote commands before session started"
+processRemoteCommand RemoteHostSessionStarted {ctrlClient} (s, cmd) = do
+  logDebug $ "processRemoteCommand: " <> T.pack (show s)
  -- XXX: intercept and filter some commands
  -- TODO: store missing files on remote host
  relayCommand ctrlClient s

 relayCommand :: (ChatMonad m) => HTTP2Client -> ByteString -> m ChatResponse
 relayCommand http s =
-  postBytestring Nothing http "/relay" mempty s >>= \case
-    Left e -> error "TODO: http2chatError"
+  postBytestring Nothing http "/send" mempty s >>= \case
+    Left e -> err $ "relayCommand/post: " <> show e
    Right HTTP2.HTTP2Response {respBody = HTTP2Body {bodyHead}} -> do
-      remoteChatResponse <-
-        if iTax
-          then case J.eitherDecodeStrict bodyHead of -- XXX: large JSONs can overflow into buffered chunks
-            Left e -> error "TODO: json2chatError" e
-            Right (raw :: J.Value) -> case J.fromJSON (sum2tagged raw) of
-              J.Error e -> error "TODO: json2chatError" e
-              J.Success cr -> pure cr
-          else case J.eitherDecodeStrict bodyHead of -- XXX: large JSONs can overflow into buffered chunks
-            Left e -> error "TODO: json2chatError" e
-            Right cr -> pure cr
+      logDebug $ "Got /send response: " <> T.pack (show bodyHead)
+      remoteChatResponse <- case J.eitherDecodeStrict bodyHead of -- XXX: large JSONs can overflow into buffered chunks
+        Left e -> err $ "relayCommand/decodeValue: " <> show e
+        Right json -> case J.fromJSON $ toTaggedJSON json of
+          J.Error e -> err $ "relayCommand/fromJSON: " <> show e
+          J.Success cr -> pure cr
      case remoteChatResponse of
        -- TODO: intercept file responses and fetch files when needed
        -- XXX: is that even possible, to have a file response to a command?
        _ -> pure remoteChatResponse
  where
-    iTax = True -- TODO: get from RemoteHost
+    err = pure . CRChatError Nothing . ChatError . CEInternalError
+    toTaggedJSON :: J.Value -> J.Value
+    toTaggedJSON = id -- owsf2tagged TODO: get from RemoteHost
    -- XXX: extract to http2 transport
-    postBytestring timeout c path hs body = liftIO $ HTTP2.sendRequest c req timeout
+    postBytestring timeout' c path hs body = liftIO $ HTTP2.sendRequestDirect c req timeout'
      where
        req = HTTP2Client.requestBuilder "POST" path hs (Binary.fromByteString body)

 -- | Convert swift single-field sum encoding into tagged/discriminator-field
-sum2tagged :: J.Value -> J.Value
-sum2tagged = \case
+owsf2tagged :: J.Value -> J.Value
+owsf2tagged = \case
  J.Object todo'convert -> J.Object todo'convert
  skip -> skip

@@ -161,13 +199,13 @@ storeRemoteFile http localFile = do
  where
    postFile timeout c path hs file = liftIO $ do
      fileSize <- fromIntegral <$> getFileSize file
-      HTTP2.sendRequest c (req fileSize) timeout
+      HTTP2.sendRequestDirect c (req fileSize) timeout
      where
-        req size = HTTP2Client.requestFile "POST" path hs (HTTP2Client.FileSpec file 0 size)
+        req size = HTTP2Client.requestFile "PUT" path hs (HTTP2Client.FileSpec file 0 size)

 fetchRemoteFile :: (ChatMonad m) => HTTP2Client -> FilePath -> FileTransferId -> m ChatResponse
 fetchRemoteFile http storePath remoteFileId = do
-  liftIO (HTTP2.sendRequest http req Nothing) >>= \case
+  liftIO (HTTP2.sendRequestDirect http req Nothing) >>= \case
    Left e -> error "TODO: http2chatError"
    Right HTTP2.HTTP2Response {respBody} -> do
      error "TODO: stream body into a local file" -- XXX: consult headers for a file name?
@@ -175,47 +213,84 @@ fetchRemoteFile http storePath remoteFileId = do
    req = HTTP2Client.requestNoBody "GET" path mempty
    path = "/fetch/" <> bshow remoteFileId

-processControllerRequest :: (ChatMonad m) => RemoteCtrlId -> HTTP2.HTTP2Request -> m ()
-processControllerRequest rc req = error "TODO: processControllerRequest"
+processControllerRequest :: forall m . (ChatMonad m) => (ByteString -> m ChatResponse) -> HTTP2.HTTP2Request -> m ()
+processControllerRequest execChatCommand HTTP2.HTTP2Request {request, reqBody = HTTP2Body {bodyHead}, sendResponse} = do
+  logDebug $ "Remote controller request: " <> T.pack (show $ method <> " " <> path)
+  res <- tryChatError $ case (method, path) 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 "" $ HTTP2Server.requestMethod request
+    path = fromMaybe "" $ HTTP2Server.requestPath request
+    getHello = respond "OK"
+    sendCommand = execChatCommand bodyHead >>= respondJSON
+    recvMessage = chatReadVar remoteCtrlSession >>= \case
+      Nothing -> respondWith Status.internalServerError500 "session not active"
+      Just rcs -> atomically (readTBQueue $ remoteOutputQ rcs) >>= respondJSON
+    storeFile = respondWith Status.notImplemented501 "TODO: storeFile"
+    fetchFile = respondWith Status.notImplemented501 "TODO: fetchFile"
+
+    respondJSON :: J.ToJSON a => a -> m ()
+    respondJSON = respond . Binary.fromLazyByteString . J.encode
+
+    respond = respondWith Status.ok200
+    respondWith status = liftIO . sendResponse . HTTP2Server.responseBuilder status []

 -- * ChatRequest handlers

-startRemoteCtrl :: (ChatMonad m) => m ChatResponse
-startRemoteCtrl =
+startRemoteCtrl :: (ChatMonad m) => (ByteString -> m ChatResponse) -> m ChatResponse
+startRemoteCtrl execChatCommand =
  chatReadVar remoteCtrlSession >>= \case
    Just _busy -> throwError $ ChatErrorRemoteCtrl RCEBusy
    Nothing -> do
-      accepted <- newEmptyTMVarIO
+      size <- asks $ tbqSize . config
+      remoteOutputQ <- newTBQueueIO size
+      remoteNotifyQ <- newTBQueueIO size
      discovered <- newTVarIO mempty
      discoverer <- async $ discoverRemoteCtrls discovered
+      accepted <- newEmptyTMVarIO
      supervisor <- async $ do
        remoteCtrlId <- atomically (readTMVar accepted)
        withRemoteCtrl remoteCtrlId $ \RemoteCtrl {displayName, fingerprint} -> do
          source <- atomically $ TM.lookup fingerprint discovered >>= maybe retry pure
          toView $ CRRemoteCtrlConnecting {remoteCtrlId, displayName}
          atomically $ writeTVar discovered mempty -- flush unused sources
-          server <- async $ Discovery.connectRevHTTP2 source fingerprint (processControllerRequest remoteCtrlId)
+          server <- async $ Discovery.connectRevHTTP2 source fingerprint (processControllerRequest execChatCommand)
          chatModifyVar remoteCtrlSession $ fmap $ \s -> s {hostServer = Just server}
          toView $ CRRemoteCtrlConnected {remoteCtrlId, displayName}
          _ <- waitCatch server
          chatWriteVar remoteCtrlSession Nothing
-          toView $ CRRemoteCtrlStopped {remoteCtrlId}
-      chatWriteVar remoteCtrlSession $ Just RemoteCtrlSession {discoverer, supervisor, hostServer = Nothing, discovered, accepted}
+          toView $ CRRemoteCtrlStopped Nothing
+      chatWriteVar remoteCtrlSession $ Just RemoteCtrlSession {discoverer, supervisor, hostServer = Nothing, discovered, accepted, remoteOutputQ, remoteNotifyQ}
      pure $ CRRemoteCtrlStarted Nothing

 discoverRemoteCtrls :: (ChatMonad m) => TM.TMap C.KeyHash TransportHost -> m ()
-discoverRemoteCtrls discovered = Discovery.openListener >>= go
+discoverRemoteCtrls discovered = Discovery.withListener go
  where
    go sock =
      Discovery.recvAnnounce sock >>= \case
-        (SockAddrInet _port addr, invite) -> case strDecode invite of
+        (SockAddrInet _sockPort sockAddr, invite) -> case strDecode invite of
          Left _ -> go sock -- ignore malformed datagrams
          Right fingerprint -> do
-            atomically $ TM.insert fingerprint (THIPv4 $ hostAddressToTuple addr) discovered
+            let addr = THIPv4 (hostAddressToTuple sockAddr)
+            ifM
+              (atomically $ TM.member fingerprint discovered)
+              (logDebug $ "Fingerprint announce already knwon: " <> T.pack (show (addr, invite)))
+              (do
+                logInfo $ "New fingerprint announce: " <> T.pack (show (addr, invite))
+                atomically $ TM.insert fingerprint addr discovered
+              )
            withStore' (`getRemoteCtrlByFingerprint` fingerprint) >>= \case
-              Nothing -> toView $ CRRemoteCtrlAnnounce fingerprint -- unknown controller, ui action required
+              Nothing -> toView $ CRRemoteCtrlAnnounce fingerprint -- unknown controller, ui "register" action required
              Just found@RemoteCtrl {remoteCtrlId, accepted=storedChoice} -> case storedChoice of
-                Nothing -> toView $ CRRemoteCtrlFound found -- first-time controller, ui action required
+                Nothing -> toView $ CRRemoteCtrlFound found -- first-time controller, ui "accept" action required
                Just False -> pure () -- skipping a rejected item
                Just True -> chatReadVar remoteCtrlSession >>= \case
                  Nothing -> toView . CRChatError Nothing . ChatError $ CEInternalError "Remote host found without running a session"
@@ -258,20 +333,28 @@ rejectRemoteCtrl remoteCtrlId = do
      cancel supervisor
  pure $ CRRemoteCtrlRejected {remoteCtrlId}

-stopRemoteCtrl :: (ChatMonad m) => RemoteCtrlId -> m ChatResponse
-stopRemoteCtrl remoteCtrlId =
+stopRemoteCtrl :: (ChatMonad m) => m ChatResponse
+stopRemoteCtrl =
  chatReadVar remoteCtrlSession >>= \case
    Nothing -> throwError $ ChatErrorRemoteCtrl RCEInactive
-    Just RemoteCtrlSession {discoverer, supervisor, hostServer} -> do
-      cancel discoverer -- may be gone by now
-      case hostServer of
-        Just host -> cancel host -- supervisor will clean up
-        Nothing -> do
-          cancel supervisor -- supervisor is blocked until session progresses
-          chatWriteVar remoteCtrlSession Nothing
-          toView $ CRRemoteCtrlStopped {remoteCtrlId}
+    Just rcs -> do
+      cancelRemoteCtrlSession rcs $ do
+        chatWriteVar remoteCtrlSession Nothing
+        toView $ CRRemoteCtrlStopped Nothing
      pure $ CRCmdOk Nothing

+cancelRemoteCtrlSession_ :: MonadUnliftIO m => RemoteCtrlSession -> m ()
+cancelRemoteCtrlSession_ rcs = cancelRemoteCtrlSession rcs $ pure ()
+
+cancelRemoteCtrlSession :: MonadUnliftIO m => RemoteCtrlSession -> m () -> m ()
+cancelRemoteCtrlSession RemoteCtrlSession {discoverer, supervisor, hostServer} cleanup = do
+  cancel discoverer -- may be gone by now
+  case hostServer of
+    Just host -> cancel host -- supervisor will clean up
+    Nothing -> do
+      cancel supervisor -- supervisor is blocked until session progresses
+      cleanup
+
 deleteRemoteCtrl :: (ChatMonad m) => RemoteCtrlId -> m ChatResponse
 deleteRemoteCtrl remoteCtrlId =
  chatReadVar remoteCtrlSession >>= \case