diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Revision history for erebos +## 0.1.4 -- 2024-06-11 + +* Added `/conversations` command to list and select conversations +* Added `/details` command for info about selected conversation +* Handle peer reconnection after its restart +* Support non-interactive mode without tty + ## 0.1.3 -- 2024-05-05 * Enable/disable network services by command-line parameters diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -36,10 +36,38 @@ After the initial setup, the erebos tool presents interactive prompt for messages and commands. All commands start with the slash (`/`) character, -followed by command name and parameters (if any) separated by spaces. When a -peer or contact is selected, message to send him can be entered directly on the -command prompt. +followed by command name and parameters (if any) separated by spaces. When +a conversation is selected, message to send there is entered directly on +the command prompt. +The session can be terminated either by end-of-input (typically `Ctrl-d`) or +using the `/quit` command. + +### Example + +Start `erebos` CLI and create new identity: +``` +Name: Some Name +Device: First device +Some Name / First device +> +``` + +Add public peer: +``` +> /peer-add-public +[1] PEER NEW [37.221.243.57 29665] +[1] PEER UPD discovery1.erebosprotocol.net [37.221.243.57 29665] +``` + +Select the peer and send it a message, the public server just responds with automatic echo message: +``` +> /1 +discovery1.erebosprotocol.net> hello +[18:55] Some Name: hello +[18:55] discovery1.erebosprotocol.net: Echo: hello +``` + ### Messaging `/peers` @@ -49,16 +77,22 @@ `/contacts` List known contacts (see below). +`/conversations` +List started conversations with contacts or other peers. + `/` -Select contact or peer `` based on previous `/contacts` or `/peers` -output list. +Select conversation, contact or peer `` based on the last +`/conversations`, `/contacts` or `/peers` output list. `` -Send `` to selected contact. +Send `` to selected conversation. `/history` -Show message history for selected contact or peer. +Show message history of the selected conversation. +`/details` +Show information about the selected conversations, contact or peer. + ### Add contacts To ensure the identity of the contact and prevent man-in-the-middle attack, @@ -119,8 +153,18 @@ `/peer-add []` Manually add network peer with given hostname or IP address. +`/peer-add-public` +Add known public network peer(s). + +`/peer-drop` +Drop the currently selected peer. Afterwards, the connection can be +re-established by either side. + `/update-identity` Interactively update current identity information + +`/quit` +Quit the erebos tool. Storage diff --git a/erebos.cabal b/erebos.cabal --- a/erebos.cabal +++ b/erebos.cabal @@ -1,7 +1,7 @@ Cabal-Version: 2.2 Name: erebos -Version: 0.1.3 +Version: 0.1.4 Synopsis: Decentralized messaging and synchronization Description: Library and simple CLI interface implementing the Erebos identity @@ -97,6 +97,7 @@ Erebos.Channel Erebos.Chatroom Erebos.Contact + Erebos.Conversation Erebos.Identity Erebos.Message Erebos.Network diff --git a/main/Main.hs b/main/Main.hs --- a/main/Main.hs +++ b/main/Main.hs @@ -20,9 +20,10 @@ import Data.List import Data.Maybe import Data.Ord -import qualified Data.Text as T -import qualified Data.Text.Encoding as T -import qualified Data.Text.IO as T +import Data.Text (Text) +import Data.Text qualified as T +import Data.Text.Encoding qualified as T +import Data.Text.IO qualified as T import Data.Time.LocalTime import Data.Typeable @@ -36,12 +37,13 @@ import Erebos.Attach import Erebos.Contact +import Erebos.Conversation #ifdef ENABLE_ICE_SUPPORT import Erebos.Discovery import Erebos.ICE #endif import Erebos.Identity -import Erebos.Message +import Erebos.Message hiding (formatMessage) import Erebos.Network import Erebos.PubKey import Erebos.Service @@ -57,6 +59,7 @@ data Options = Options { optServer :: ServerOptions , optServices :: [ServiceOption] + , optDmBotEcho :: Maybe Text , optShowHelp :: Bool , optShowVersion :: Bool } @@ -72,6 +75,7 @@ defaultOptions = Options { optServer = defaultServerOptions , optServices = availableServices + , optDmBotEcho = Nothing , optShowHelp = False , optShowVersion = False } @@ -100,6 +104,9 @@ , Option ['s'] ["silent"] (NoArg (so $ \opts -> opts { serverLocalDiscovery = False })) "do not send announce packets for local discovery" + , Option [] ["dm-bot-echo"] + (ReqArg (\prefix -> \opts -> opts { optDmBotEcho = Just (T.pack prefix) }) "") + "automatically reply to direct messages with the same text prefixed with " , Option ['h'] ["help"] (NoArg $ \opts -> opts { optShowHelp = True }) "show this help and exit" @@ -184,7 +191,7 @@ serviceDesc ServiceOption {..} = padService (" " <> soptName) <> soptDescription padTo n str = str <> replicate (n - length str) ' ' - padOpt = padTo 28 + padOpt = padTo 37 padService = padTo 16 if | optShowHelp opts -> putStr $ usageInfo header options <> unlines @@ -213,16 +220,66 @@ erebosHead <- liftIO $ loadLocalStateHead st outputStrLn $ T.unpack $ displayIdentity $ headLocalIdentity erebosHead - haveTerminalUI >>= \case True -> return () - False -> error "Requires terminal" + tui <- haveTerminalUI extPrint <- getExternalPrint let extPrintLn str = extPrint $ case reverse str of ('\n':_) -> str _ -> str ++ "\n"; + let getInputLinesTui eprompt = do + prompt <- case eprompt of + Left cstate -> do + pname <- case csContext cstate of + NoContext -> return "" + SelectedPeer peer -> peerIdentity peer >>= return . \case + PeerIdentityFull pid -> maybe "" T.unpack $ idName $ finalOwner pid + PeerIdentityRef wref _ -> "<" ++ BC.unpack (showRefDigest $ wrDigest wref) ++ ">" + PeerIdentityUnknown _ -> "" + SelectedContact contact -> return $ T.unpack $ contactName contact + SelectedConversation conv -> return $ T.unpack $ conversationName conv + return $ pname ++ "> " + Right prompt -> return prompt + Just input <- lift $ getInputLine prompt + case reverse input of + _ | all isSpace input -> getInputLinesTui eprompt + '\\':rest -> (reverse ('\n':rest) ++) <$> getInputLinesTui (Right ">> ") + _ -> return input + + getInputCommandTui cstate = do + input <- getInputLinesTui cstate + let (CommandM cmd, line) = case input of + '/':rest -> let (scmd, args) = dropWhile isSpace <$> span (\c -> isAlphaNum c || c == '-') rest + in if not (null scmd) && all isDigit scmd + then (cmdSelectContext, scmd) + else (fromMaybe (cmdUnknown scmd) $ lookup scmd commands, args) + _ -> (cmdSend, input) + return (cmd, line) + + getInputLinesPipe = do + lift (getInputLine "") >>= \case + Just input -> return input + Nothing -> liftIO $ forever $ threadDelay 100000000 + + getInputCommandPipe _ = do + input <- getInputLinesPipe + let (scmd, args) = dropWhile isSpace <$> span (\c -> isAlphaNum c || c == '-') input + let (CommandM cmd, line) = (fromMaybe (cmdUnknown scmd) $ lookup scmd commands, args) + return (cmd, line) + + let getInputCommand = if tui then getInputCommandTui . Left + else getInputCommandPipe + _ <- liftIO $ do tzone <- getCurrentTimeZone - watchReceivedMessages erebosHead $ - extPrintLn . formatMessage tzone . fromStored + watchReceivedMessages erebosHead $ \smsg -> do + let msg = fromStored smsg + extPrintLn $ formatDirectMessage tzone msg + case optDmBotEcho opts of + Nothing -> return () + Just prefix -> do + res <- runExceptT $ flip runReaderT erebosHead $ sendDirectMessage (msgFrom msg) (prefix <> msgText msg) + case res of + Right reply -> extPrintLn $ formatDirectMessage tzone $ fromStored reply + Left err -> extPrintLn $ "Failed to send dm echo: " <> err server <- liftIO $ do startServer (optServer opts) erebosHead extPrintLn $ @@ -235,58 +292,47 @@ peer <- getNextPeerChange server peerIdentity peer >>= \case pid@(PeerIdentityFull _) -> do + dropped <- isPeerDropped peer let shown = showPeer pid $ peerAddress peer - let update [] = ([(peer, shown)], Nothing) - update ((p,s):ps) | p == peer = ((peer, shown) : ps, Just s) - | otherwise = first ((p,s):) $ update ps + let update [] = ([(peer, shown)], (Nothing, "NEW")) + update ((p,s):ps) + | p == peer && dropped = (ps, (Nothing, "DEL")) + | p == peer = ((peer, shown) : ps, (Just s, "UPD")) + | otherwise = first ((p,s):) $ update ps let ctxUpdate n [] = ([SelectedPeer peer], n) ctxUpdate n (ctx:ctxs) | SelectedPeer p <- ctx, p == peer = (ctx:ctxs, n) | otherwise = first (ctx:) $ ctxUpdate (n + 1) ctxs - op <- modifyMVar peers (return . update) + (op, updateType) <- modifyMVar peers (return . update) + let updateType' = if dropped then "DEL" else updateType idx <- modifyMVar contextOptions (return . ctxUpdate (1 :: Int)) - when (Just shown /= op) $ extPrint $ "[" <> show idx <> "] PEER " <> shown + when (Just shown /= op) $ extPrintLn $ "[" <> show idx <> "] PEER " <> updateType' <> " " <> shown _ -> return () - let getInputLines prompt = do - Just input <- lift $ getInputLine prompt - case reverse input of - _ | all isSpace input -> getInputLines prompt - '\\':rest -> (reverse ('\n':rest) ++) <$> getInputLines ">> " - _ -> return input - let process :: CommandState -> MaybeT (InputT IO) CommandState process cstate = do - pname <- case csContext cstate of - NoContext -> return "" - SelectedPeer peer -> peerIdentity peer >>= return . \case - PeerIdentityFull pid -> maybe "" T.unpack $ idName $ finalOwner pid - PeerIdentityRef wref _ -> "<" ++ BC.unpack (showRefDigest $ wrDigest wref) ++ ">" - PeerIdentityUnknown _ -> "" - SelectedContact contact -> return $ T.unpack $ contactName contact - input <- getInputLines $ pname ++ "> " - let (CommandM cmd, line) = case input of - '/':rest -> let (scmd, args) = dropWhile isSpace <$> span (\c -> isAlphaNum c || c == '-') rest - in if not (null scmd) && all isDigit scmd - then (cmdSelectContext $ read scmd, args) - else (fromMaybe (cmdUnknown scmd) $ lookup scmd commands, args) - _ -> (cmdSend, input) + (cmd, line) <- getInputCommand cstate h <- liftIO (reloadHead $ csHead cstate) >>= \case Just h -> return h - Nothing -> do lift $ lift $ extPrint "current head deleted" + Nothing -> do lift $ lift $ extPrintLn "current head deleted" mzero res <- liftIO $ runExceptT $ flip execStateT cstate { csHead = h } $ runReaderT cmd CommandInput { ciServer = server , ciLine = line , ciPrint = extPrintLn - , ciPeers = liftIO $ readMVar peers + , ciPeers = liftIO $ modifyMVar peers $ \ps -> do + ps' <- filterM (fmap not . isPeerDropped . fst) ps + return (ps', ps') , ciContextOptions = liftIO $ readMVar contextOptions , ciSetContextOptions = \ctxs -> liftIO $ modifyMVar_ contextOptions $ const $ return ctxs } case res of - Right cstate' -> return cstate' - Left err -> do lift $ lift $ extPrint $ "Error: " ++ err - return cstate + Right cstate' + | csQuit cstate' -> mzero + | otherwise -> return cstate' + Left err -> do + lift $ lift $ extPrintLn $ "Error: " ++ err + return cstate let loop (Just cstate) = runMaybeT (process cstate) >>= loop loop Nothing = return () @@ -297,6 +343,7 @@ , csIceSessions = [] #endif , csIcePeer = Nothing + , csQuit = False } @@ -316,11 +363,13 @@ , csIceSessions :: [IceSession] #endif , csIcePeer :: Maybe Peer + , csQuit :: Bool } data CommandContext = NoContext | SelectedPeer Peer | SelectedContact Contact + | SelectedConversation Conversation newtype CommandM a = CommandM (ReaderT CommandInput (StateT CommandState (ExceptT String IO)) a) deriving (Functor, Applicative, Monad, MonadReader CommandInput, MonadState CommandState, MonadError String) @@ -353,21 +402,24 @@ SelectedPeer peer -> return peer _ -> throwError "no peer selected" -getSelectedIdentity :: CommandM ComposedIdentity -getSelectedIdentity = gets csContext >>= \case +getSelectedConversation :: CommandM Conversation +getSelectedConversation = gets csContext >>= \case SelectedPeer peer -> peerIdentity peer >>= \case - PeerIdentityFull pid -> return $ toComposedIdentity pid + PeerIdentityFull pid -> directMessageConversation $ finalOwner pid _ -> throwError "incomplete peer identity" SelectedContact contact -> case contactIdentity contact of - Just cid -> return cid + Just cid -> directMessageConversation cid Nothing -> throwError "contact without erebos identity" - _ -> throwError "no contact or peer selected" + SelectedConversation conv -> reloadConversation conv + _ -> throwError "no contact, peer or conversation selected" commands :: [(String, Command)] commands = [ ("history", cmdHistory) , ("peers", cmdPeers) , ("peer-add", cmdPeerAdd) + , ("peer-add-public", cmdPeerAddPublic) + , ("peer-drop", cmdPeerDrop) , ("send", cmdSend) , ("update-identity", cmdUpdateIdentity) , ("attach", cmdAttach) @@ -377,6 +429,8 @@ , ("contact-add", cmdContactAdd) , ("contact-accept", cmdContactAccept) , ("contact-reject", cmdContactReject) + , ("conversations", cmdConversations) + , ("details", cmdDetails) #ifdef ENABLE_ICE_SUPPORT , ("discovery-init", cmdDiscoveryInit) , ("discovery", cmdDiscovery) @@ -386,6 +440,8 @@ , ("ice-connect", cmdIceConnect) , ("ice-send", cmdIceSend) #endif + , ("select", cmdSelectContext) + , ("quit", cmdQuit) ] commandCompletion :: CompletionFunc IO @@ -417,6 +473,17 @@ addr:_ <- liftIO $ getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just hostname) (Just port) liftIO $ serverPeer server (addrAddress addr) +cmdPeerAddPublic :: Command +cmdPeerAddPublic = do + server <- asks ciServer + addr:_ <- liftIO $ getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just "discovery1.erebosprotocol.net") (Just (show discoveryPort)) + void $ liftIO $ serverPeer server (addrAddress addr) + +cmdPeerDrop :: Command +cmdPeerDrop = do + dropPeer =<< getSelectedPeer + modify $ \s -> s { csContext = NoContext } + showPeer :: PeerIdentity -> PeerAddress -> String showPeer pidentity paddr = let name = case pidentity of @@ -425,30 +492,29 @@ PeerIdentityFull pid -> T.unpack $ displayIdentity pid in name ++ " [" ++ show paddr ++ "]" -cmdSelectContext :: Int -> Command -cmdSelectContext n = join (asks ciContextOptions) >>= \ctxs -> if - | n > 0, (ctx : _) <- drop (n - 1) ctxs -> modify $ \s -> s { csContext = ctx } - | otherwise -> throwError "invalid index" +cmdSelectContext :: Command +cmdSelectContext = do + n <- read <$> asks ciLine + join (asks ciContextOptions) >>= \ctxs -> if + | n > 0, (ctx : _) <- drop (n - 1) ctxs -> modify $ \s -> s { csContext = ctx } + | otherwise -> throwError "invalid index" cmdSend :: Command cmdSend = void $ do text <- asks ciLine - powner <- finalOwner <$> getSelectedIdentity - smsg <- sendDirectMessage powner $ T.pack text + conv <- getSelectedConversation + msg <- sendMessage conv $ T.pack text tzone <- liftIO $ getCurrentTimeZone - liftIO $ putStrLn $ formatMessage tzone $ fromStored smsg + liftIO $ putStrLn $ formatMessage tzone msg cmdHistory :: Command cmdHistory = void $ do - ehead <- gets csHead - powner <- finalOwner <$> getSelectedIdentity - - case find (sameIdentity powner . msgPeer) $ - toThreadList $ lookupSharedValue $ lsShared $ headObject ehead of - Just thread -> do + conv <- getSelectedConversation + case conversationHistory conv of + thread@(_:_) -> do tzone <- liftIO $ getCurrentTimeZone - liftIO $ mapM_ (putStrLn . formatMessage tzone) $ reverse $ take 50 $ threadToList thread - Nothing -> do + liftIO $ mapM_ (putStrLn . formatMessage tzone) $ reverse $ take 50 thread + [] -> do liftIO $ putStrLn $ "" cmdUpdateIdentity :: Command @@ -493,6 +559,74 @@ cmdContactReject :: Command cmdContactReject = contactReject =<< getSelectedPeer +cmdConversations :: Command +cmdConversations = do + conversations <- lookupConversations + set <- asks ciSetContextOptions + set $ map SelectedConversation conversations + forM_ (zip [1..] conversations) $ \(i :: Int, conv) -> do + liftIO $ putStrLn $ "[" ++ show i ++ "] " ++ T.unpack (conversationName conv) + +cmdDetails :: Command +cmdDetails = do + gets csContext >>= \case + SelectedPeer peer -> do + liftIO $ putStr $ unlines + [ "Network peer:" + , " " <> show (peerAddress peer) + ] + peerIdentity peer >>= \case + PeerIdentityUnknown _ -> liftIO $ do + putStrLn $ "unknown identity" + PeerIdentityRef wref _ -> liftIO $ do + putStrLn $ "Identity ref:" + putStrLn $ " " <> BC.unpack (showRefDigest $ wrDigest wref) + PeerIdentityFull pid -> printContactOrIdentityDetails pid + + SelectedContact contact -> do + printContactDetails contact + + SelectedConversation conv -> do + case conversationPeer conv of + Just pid -> printContactOrIdentityDetails pid + Nothing -> liftIO $ putStrLn $ "(conversation without peer)" + + NoContext -> liftIO $ putStrLn "nothing selected" + where + printContactOrIdentityDetails cid = do + contacts <- fromSetBy (comparing contactName) . lookupSharedValue . lsShared . fromStored <$> getLocalHead + case find (maybe False (sameIdentity cid) . contactIdentity) contacts of + Just contact -> printContactDetails contact + Nothing -> printIdentityDetails cid + + printContactDetails contact = liftIO $ do + putStrLn $ "Contact:" + prefix <- case contactCustomName contact of + Just name -> do + putStrLn $ " " <> T.unpack name + return $ Just "alias of" + Nothing -> do + return $ Nothing + + case contactIdentity contact of + Just cid -> do + printIdentityDetailsBody prefix cid + Nothing -> do + putStrLn $ " (without erebos identity)" + + printIdentityDetails identity = liftIO $ do + putStrLn $ "Identity:" + printIdentityDetailsBody Nothing identity + + printIdentityDetailsBody prefix identity = do + forM_ (zip (False : repeat True) $ unfoldOwners identity) $ \(owned, cpid) -> do + putStrLn $ unwords $ concat + [ [ " " ] + , if owned then [ "owned by" ] else maybeToList prefix + , [ maybe "" T.unpack (idName cpid) ] + , map (BC.unpack . showRefDigest . refDigest . storedRef) $ idExtDataF cpid + ] + #ifdef ENABLE_ICE_SUPPORT cmdDiscoveryInit :: Command @@ -566,3 +700,6 @@ liftIO $ serverPeerIce server s #endif + +cmdQuit :: Command +cmdQuit = modify $ \s -> s { csQuit = True } diff --git a/main/Test.hs b/main/Test.hs --- a/main/Test.hs +++ b/main/Test.hs @@ -246,6 +246,8 @@ , ("start-server", cmdStartServer) , ("stop-server", cmdStopServer) , ("peer-add", cmdPeerAdd) + , ("peer-drop", cmdPeerDrop) + , ("peer-list", cmdPeerList) , ("test-message-send", cmdTestMessageSend) , ("shared-state-get", cmdSharedStateGet) , ("shared-state-wait", cmdSharedStateWait) @@ -269,6 +271,7 @@ , ("chatroom-list-local", cmdChatroomListLocal) , ("chatroom-watch-local", cmdChatroomWatchLocal) , ("chatroom-set-name", cmdChatroomSetName) + , ("chatroom-message-send", cmdChatroomMessageSend) ] cmdStore :: Command @@ -409,6 +412,29 @@ addr:_ <- liftIO $ getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just host) (Just port) void $ liftIO $ serverPeer rsServer (addrAddress addr) +cmdPeerDrop :: Command +cmdPeerDrop = do + [spidx] <- asks tiParams + peer <- getPeer spidx + liftIO $ dropPeer peer + +cmdPeerList :: Command +cmdPeerList = do + Just RunningServer {..} <- gets tsServer + peers <- liftIO $ getCurrentPeerList rsServer + tpeers <- liftIO $ readMVar rsPeers + forM_ peers $ \peer -> do + Just (n, _) <- return $ find ((peer==).snd) . snd $ tpeers + mbpid <- peerIdentity peer + cmdOut $ unwords $ concat + [ [ "peer-list-item", show n ] + , [ "addr", show (peerAddress peer) ] + , case mbpid of PeerIdentityFull pid -> ("id":) $ map (maybe "" T.unpack . idName) (unfoldOwners pid) + _ -> [] + ] + cmdOut "peer-list-done" + + cmdTestMessageSend :: Command cmdTestMessageSend = do [spidx, tref] <- asks tiParams @@ -628,14 +654,30 @@ Just diff -> forM_ diff $ \case AddedChatroom room -> outLine out $ unwords $ "chatroom-watched-added" : chatroomInfo room RemovedChatroom room -> outLine out $ unwords $ "chatroom-watched-removed" : chatroomInfo room - UpdatedChatroom oldroom room -> outLine out $ unwords $ concat - [ [ "chatroom-watched-updated" ], chatroomInfo room - , [ "old" ], map (show . refDigest . storedRef) (roomStateData oldroom) - , [ "new" ], map (show . refDigest . storedRef) (roomStateData room) - ] + UpdatedChatroom oldroom room -> do + when (any (not . null . rsdRoom . fromStored) (roomStateData room)) $ do + outLine out $ unwords $ concat + [ [ "chatroom-watched-updated" ], chatroomInfo room + , [ "old" ], map (show . refDigest . storedRef) (roomStateData oldroom) + , [ "new" ], map (show . refDigest . storedRef) (roomStateData room) + ] + when (any (not . null . rsdMessages . fromStored) (roomStateData room)) $ do + forM_ (getMessagesSinceState room oldroom) $ \msg -> do + outLine out $ unwords $ concat + [ [ "chatroom-message-new" ] + , [ show . refDigest . storedRef . head . filterAncestors . concatMap storedRoots . toComponents $ room ] + , [ "from", maybe "" T.unpack $ idName $ cmsgFrom msg ] + , maybe [] (("text":) . (:[]) . T.unpack) $ cmsgText msg + ] chatroomInfo :: ChatroomState -> [String] chatroomInfo room = [ show . refDigest . storedRef . head . filterAncestors . concatMap storedRoots . toComponents $ room , maybe "" T.unpack $ roomName =<< roomStateRoom room ] + +cmdChatroomMessageSend :: Command +cmdChatroomMessageSend = do + [cid, msg] <- asks tiParams + to <- getChatroomStateData cid + void $ chatroomMessageByStateData to msg diff --git a/src/Erebos/Chatroom.hs b/src/Erebos/Chatroom.hs --- a/src/Erebos/Chatroom.hs +++ b/src/Erebos/Chatroom.hs @@ -10,10 +10,15 @@ listChatrooms, findChatroomByRoomData, findChatroomByStateData, + getMessagesSinceState, ChatroomSetChange(..), watchChatrooms, + ChatMessage, cmsgFrom, cmsgReplyTo, cmsgTime, cmsgText, cmsgLeave, + ChatMessageData(..), + chatroomMessageByStateData, + ChatroomService(..), ) where @@ -27,8 +32,11 @@ import Data.Maybe import Data.Monoid import Data.Ord +import Data.Set qualified as S import Data.Text (Text) +import Data.Time +import Erebos.Identity import Erebos.PubKey import Erebos.Service import Erebos.Set @@ -87,34 +95,129 @@ in All $ all (fromStored sdata `isSignedBy`) required +data ChatMessageData = ChatMessageData + { mdPrev :: [Stored (Signed ChatMessageData)] + , mdRoom :: [Stored (Signed ChatroomData)] + , mdFrom :: ComposedIdentity + , mdReplyTo :: Maybe (Stored (Signed ChatMessageData)) + , mdTime :: ZonedTime + , mdText :: Maybe Text + , mdLeave :: Bool + } + +data ChatMessage = ChatMessage + { cmsgData :: Stored (Signed ChatMessageData) + } + +cmsgFrom :: ChatMessage -> ComposedIdentity +cmsgFrom = mdFrom . fromSigned . cmsgData + +cmsgReplyTo :: ChatMessage -> Maybe ChatMessage +cmsgReplyTo = fmap ChatMessage . mdReplyTo . fromSigned . cmsgData + +cmsgTime :: ChatMessage -> ZonedTime +cmsgTime = mdTime . fromSigned . cmsgData + +cmsgText :: ChatMessage -> Maybe Text +cmsgText = mdText . fromSigned . cmsgData + +cmsgLeave :: ChatMessage -> Bool +cmsgLeave = mdLeave . fromSigned . cmsgData + +instance Storable ChatMessageData where + store' ChatMessageData {..} = storeRec $ do + mapM_ (storeRef "SPREV") mdPrev + mapM_ (storeRef "room") mdRoom + mapM_ (storeRef "from") $ idExtDataF mdFrom + storeMbRef "reply-to" mdReplyTo + storeDate "time" mdTime + storeMbText "text" mdText + when mdLeave $ storeEmpty "leave" + + load' = loadRec $ do + mdPrev <- loadRefs "SPREV" + mdRoom <- loadRefs "room" + mdFrom <- loadIdentity "from" + mdReplyTo <- loadMbRef "reply-to" + mdTime <- loadDate "time" + mdText <- loadMbText "text" + mdLeave <- isJust <$> loadMbEmpty "leave" + return ChatMessageData {..} + +threadToList :: [Stored (Signed ChatMessageData)] -> [ChatMessage] +threadToList thread = helper S.empty $ thread + where + helper :: S.Set (Stored (Signed ChatMessageData)) -> [Stored (Signed ChatMessageData)] -> [ChatMessage] + helper seen msgs + | msg : msgs' <- filter (`S.notMember` seen) $ reverse $ sortBy (comparing cmpView) msgs = + messageFromData msg : helper (S.insert msg seen) (msgs' ++ mdPrev (fromSigned msg)) + | otherwise = [] + cmpView msg = (zonedTimeToUTC $ mdTime $ fromSigned msg, msg) + + messageFromData :: Stored (Signed ChatMessageData) -> ChatMessage + messageFromData sdata = ChatMessage { cmsgData = sdata } + +chatroomMessageByStateData + :: (MonadStorage m, MonadHead LocalState m, MonadError String m) + => Stored ChatroomStateData -> Text -> m () +chatroomMessageByStateData lookupData msg = void $ findAndUpdateChatroomState $ \cstate -> do + guard $ any (lookupData `precedesOrEquals`) $ roomStateData cstate + Just $ do + self <- finalOwner . localIdentity . fromStored <$> getLocalHead + secret <- loadKey $ idKeyMessage self + time <- liftIO getZonedTime + mdata <- mstore =<< sign secret =<< mstore ChatMessageData + { mdPrev = roomStateMessageData cstate + , mdRoom = [] + , mdFrom = self + , mdReplyTo = Nothing + , mdTime = time + , mdText = Just msg + , mdLeave = False + } + mergeSorted . (:[]) <$> mstore ChatroomStateData + { rsdPrev = roomStateData cstate + , rsdRoom = [] + , rsdMessages = [ mdata ] + } + + data ChatroomStateData = ChatroomStateData { rsdPrev :: [Stored ChatroomStateData] , rsdRoom :: [Stored (Signed ChatroomData)] + , rsdMessages :: [Stored (Signed ChatMessageData)] } data ChatroomState = ChatroomState { roomStateData :: [Stored ChatroomStateData] , roomStateRoom :: Maybe Chatroom + , roomStateMessageData :: [Stored (Signed ChatMessageData)] + , roomStateMessages :: [ChatMessage] } instance Storable ChatroomStateData where store' ChatroomStateData {..} = storeRec $ do forM_ rsdPrev $ storeRef "PREV" forM_ rsdRoom $ storeRef "room" + forM_ rsdMessages $ storeRef "msg" load' = loadRec $ do rsdPrev <- loadRefs "PREV" rsdRoom <- loadRefs "room" + rsdMessages <- loadRefs "msg" return ChatroomStateData {..} instance Mergeable ChatroomState where type Component ChatroomState = ChatroomStateData - mergeSorted cdata = ChatroomState - { roomStateData = cdata - , roomStateRoom = either (const Nothing) Just $ runExcept $ - validateChatroom $ concat $ findProperty ((\case [] -> Nothing; xs -> Just xs) . rsdRoom) cdata - } + mergeSorted roomStateData = + let roomStateRoom = either (const Nothing) Just $ runExcept $ + validateChatroom $ concat $ findProperty ((\case [] -> Nothing; xs -> Just xs) . rsdRoom) roomStateData + roomStateMessageData = filterAncestors $ concat $ flip findProperty roomStateData $ \case + ChatroomStateData {..} | null rsdMessages -> Nothing + | otherwise -> Just rsdMessages + roomStateMessages = threadToList $ concatMap (rsdMessages . fromStored) roomStateData + in ChatroomState {..} toComponents = roomStateData @@ -129,6 +232,7 @@ cstate <- mergeSorted . (:[]) <$> mstore ChatroomStateData { rsdPrev = [] , rsdRoom = [ rdata ] + , rsdMessages = [] } updateLocalHead $ updateSharedState $ \rooms -> do @@ -174,6 +278,7 @@ mergeSorted . (:[]) <$> mstore ChatroomStateData { rsdPrev = roomStateData cstate , rsdRoom = [ rdata ] + , rsdMessages = [] } @@ -193,7 +298,13 @@ findChatroomByStateData :: MonadHead LocalState m => Stored ChatroomStateData -> m (Maybe ChatroomState) findChatroomByStateData cdata = findChatroom $ any (cdata `precedesOrEquals`) . roomStateData +getMessagesSinceState :: ChatroomState -> ChatroomState -> [ChatMessage] +getMessagesSinceState cur old = takeWhile notOld (roomStateMessages cur) + where + notOld msg = cmsgData msg `notElem` roomStateMessageData old + -- TODO: parallel message threads + data ChatroomSetChange = AddedChatroom ChatroomState | RemovedChatroom ChatroomState | UpdatedChatroom ChatroomState ChatroomState @@ -231,22 +342,26 @@ data ChatroomService = ChatroomService { chatRoomQuery :: Bool , chatRoomInfo :: [Stored (Signed ChatroomData)] + , chatRoomMessage :: [Stored (Signed ChatMessageData)] } emptyPacket :: ChatroomService emptyPacket = ChatroomService { chatRoomQuery = False , chatRoomInfo = [] + , chatRoomMessage = [] } instance Storable ChatroomService where store' ChatroomService {..} = storeRec $ do when chatRoomQuery $ storeEmpty "room-query" forM_ chatRoomInfo $ storeRef "room-info" + forM_ chatRoomMessage $ storeRef "room-message" load' = loadRec $ do chatRoomQuery <- isJust <$> loadMbEmpty "room-query" chatRoomInfo <- loadRefs "room-info" + chatRoomMessage <- loadRefs "room-message" return ChatroomService {..} data PeerState = PeerState @@ -288,7 +403,11 @@ -- update local state only if we got roomInfo not present there if roomInfo `notElem` prevRoom && roomInfo `elem` room then do - sdata <- mstore ChatroomStateData { rsdPrev = prev, rsdRoom = room } + sdata <- mstore ChatroomStateData + { rsdPrev = prev + , rsdRoom = room + , rsdMessages = [] + } storeSetAddComponent sdata set else return set foldM upd roomSet chatRoomInfo diff --git a/src/Erebos/Conversation.hs b/src/Erebos/Conversation.hs new file mode 100644 --- /dev/null +++ b/src/Erebos/Conversation.hs @@ -0,0 +1,75 @@ +module Erebos.Conversation ( + Message, + messageFrom, + messageText, + messageUnread, + formatMessage, + + Conversation, + directMessageConversation, + reloadConversation, + lookupConversations, + + conversationName, + conversationPeer, + conversationHistory, + + sendMessage, +) where + +import Control.Monad.Except + +import Data.List +import Data.Maybe +import Data.Text (Text) +import Data.Text qualified as T +import Data.Time.LocalTime + +import Erebos.Identity +import Erebos.Message hiding (formatMessage) +import Erebos.State +import Erebos.Storage + + +data Message = DirectMessageMessage DirectMessage Bool + +messageFrom :: Message -> ComposedIdentity +messageFrom (DirectMessageMessage msg _) = msgFrom msg + +messageText :: Message -> Maybe Text +messageText (DirectMessageMessage msg _) = Just $ msgText msg + +messageUnread :: Message -> Bool +messageUnread (DirectMessageMessage _ unread) = unread + +formatMessage :: TimeZone -> Message -> String +formatMessage tzone (DirectMessageMessage msg _) = formatDirectMessage tzone msg + + +data Conversation = DirectMessageConversation DirectMessageThread + +directMessageConversation :: MonadHead LocalState m => ComposedIdentity -> m Conversation +directMessageConversation peer = do + (find (sameIdentity peer . msgPeer) . toThreadList . lookupSharedValue . lsShared . fromStored <$> getLocalHead) >>= \case + Just thread -> return $ DirectMessageConversation thread + Nothing -> return $ DirectMessageConversation $ DirectMessageThread peer [] [] [] + +reloadConversation :: MonadHead LocalState m => Conversation -> m Conversation +reloadConversation (DirectMessageConversation thread) = directMessageConversation (msgPeer thread) + +lookupConversations :: MonadHead LocalState m => m [Conversation] +lookupConversations = map DirectMessageConversation . toThreadList . lookupSharedValue . lsShared . fromStored <$> getLocalHead + + +conversationName :: Conversation -> Text +conversationName (DirectMessageConversation thread) = fromMaybe (T.pack "") $ idName $ msgPeer thread + +conversationPeer :: Conversation -> Maybe ComposedIdentity +conversationPeer (DirectMessageConversation thread) = Just $ msgPeer thread + +conversationHistory :: Conversation -> [Message] +conversationHistory (DirectMessageConversation thread) = map (\msg -> DirectMessageMessage msg False) $ threadToList thread + + +sendMessage :: (MonadHead LocalState m, MonadError String m) => Conversation -> Text -> m Message +sendMessage (DirectMessageConversation thread) text = DirectMessageMessage <$> (fromStored <$> sendDirectMessage (msgPeer thread) text) <*> pure False diff --git a/src/Erebos/Discovery.hs b/src/Erebos/Discovery.hs --- a/src/Erebos/Discovery.hs +++ b/src/Erebos/Discovery.hs @@ -4,6 +4,7 @@ ) where import Control.Concurrent +import Control.Monad import Control.Monad.Except import Control.Monad.Reader diff --git a/src/Erebos/Message.hs b/src/Erebos/Message.hs --- a/src/Erebos/Message.hs +++ b/src/Erebos/Message.hs @@ -14,6 +14,7 @@ watchReceivedMessages, formatMessage, + formatDirectMessage, ) where import Control.Monad @@ -126,7 +127,7 @@ instance Storable MessageState where store' MessageState {..} = storeRec $ do mapM_ (storeRef "PREV") msPrev - mapM_ (storeRef "peer") $ idDataF msPeer + mapM_ (storeRef "peer") $ idExtDataF msPeer mapM_ (storeRef "ready") msReady mapM_ (storeRef "sent") msSent mapM_ (storeRef "received") msReceived @@ -258,8 +259,12 @@ forM_ (map fromStored sms) $ \ms -> do mapM_ f $ filter (not . sameIdentity self . msgFrom . fromStored) $ msReceived ms +{-# DEPRECATED formatMessage "use formatDirectMessage instead" #-} formatMessage :: TimeZone -> DirectMessage -> String -formatMessage tzone msg = concat +formatMessage = formatDirectMessage + +formatDirectMessage :: TimeZone -> DirectMessage -> String +formatDirectMessage tzone msg = concat [ formatTime defaultTimeLocale "[%H:%M] " $ utcToLocalTime tzone $ zonedTimeToUTC $ msgTime msg , maybe "" T.unpack $ idName $ msgFrom msg , ": " diff --git a/src/Erebos/Network.hs b/src/Erebos/Network.hs --- a/src/Erebos/Network.hs +++ b/src/Erebos/Network.hs @@ -4,6 +4,7 @@ Server, startServer, stopServer, + getCurrentPeerList, getNextPeerChange, ServerOptions(..), serverIdentity, defaultServerOptions, @@ -16,6 +17,8 @@ #ifdef ENABLE_ICE_SUPPORT serverPeerIce, #endif + dropPeer, + isPeerDropped, sendToPeer, sendToPeerStored, sendToPeerWith, runPeerService, @@ -90,6 +93,9 @@ serverIdentity :: Server -> IO UnifiedIdentity serverIdentity = readMVar . serverIdentity_ +getCurrentPeerList :: Server -> IO [Peer] +getCurrentPeerList = fmap M.elems . readMVar . serverPeers + getNextPeerChange :: Server -> IO Peer getNextPeerChange = atomically . readTChan . serverChanPeer @@ -108,7 +114,7 @@ data Peer = Peer { peerAddress :: PeerAddress , peerServer_ :: Server - , peerConnection :: TVar (Either [(Bool, TransportPacket Ref, [TransportHeaderItem])] (Connection PeerAddress)) + , peerState :: TVar PeerState , peerIdentityVar :: TVar PeerIdentity , peerStorage_ :: Storage , peerInStorage :: PartialStorage @@ -123,13 +129,18 @@ peerStorage = peerStorage_ getPeerChannel :: Peer -> STM ChannelState -getPeerChannel Peer {..} = either (const $ return ChannelNone) connGetChannel =<< readTVar peerConnection +getPeerChannel Peer {..} = + readTVar peerState >>= \case + PeerInit _ -> return ChannelNone + PeerConnected conn -> connGetChannel conn + PeerDropped -> return ChannelClosed setPeerChannel :: Peer -> ChannelState -> STM () setPeerChannel Peer {..} ch = do - readTVar peerConnection >>= \case - Left _ -> retry - Right conn -> connSetChannel conn ch + readTVar peerState >>= \case + PeerInit _ -> retry + PeerConnected conn -> connSetChannel conn ch + PeerDropped -> return () instance Eq Peer where (==) = (==) `on` peerIdentityVar @@ -175,6 +186,11 @@ peerIdentity = liftIO . atomically . readTVar . peerIdentityVar +data PeerState = PeerInit [(SecurityRequirement, TransportPacket Ref, [TransportHeaderItem])] + | PeerConnected (Connection PeerAddress) + | PeerDropped + + lookupServiceType :: [TransportHeaderItem] -> Maybe ServiceID lookupServiceType (ServiceType stype : _) = Just stype lookupServiceType (_ : hs) = lookupServiceType hs @@ -196,8 +212,13 @@ forkServerThread :: Server -> IO () -> IO () -forkServerThread server act = modifyMVar_ (serverThreads server) $ \ts -> do - (:ts) <$> forkIO act +forkServerThread server act = do + modifyMVar_ (serverThreads server) $ \ts -> do + t <- forkIO $ do + t <- myThreadId + act + modifyMVar_ (serverThreads server) $ return . filter (/=t) + return (t:ts) startServer :: ServerOptions -> Head LocalState -> (String -> IO ()) -> [SomeService] -> IO Server startServer opt serverOrigHead logd' serverServices = do @@ -299,10 +320,14 @@ forkServerThread server $ do atomically $ do - readTVar (peerConnection peer) >>= \case - Left packets -> writeFlowBulk (connData conn) $ reverse packets - Right _ -> return () - writeTVar (peerConnection peer) (Right conn) + readTVar (peerState peer) >>= \case + PeerInit packets -> do + writeFlowBulk (connData conn) $ reverse packets + writeTVar (peerState peer) (PeerConnected conn) + PeerConnected _ -> do + writeTVar (peerState peer) (PeerConnected conn) + PeerDropped -> do + connClose conn case mbpid of Just dgst -> do @@ -317,12 +342,17 @@ _ -> return () Nothing -> return () - forever $ do - (secure, TransportPacket header objs) <- readFlowIO $ connData conn - prefs <- forM objs $ storeObject $ peerInStorage peer - identity <- readMVar serverIdentity_ - let svcs = map someServiceID serverServices - handlePacket identity secure peer chanSvc svcs header prefs + let peerLoop = readFlowIO (connData conn) >>= \case + Just (secure, TransportPacket header objs) -> do + prefs <- forM objs $ storeObject $ peerInStorage peer + identity <- readMVar serverIdentity_ + let svcs = map someServiceID serverServices + handlePacket identity secure peer chanSvc svcs header prefs + peerLoop + Nothing -> do + dropPeer peer + atomically $ writeTChan serverChanPeer peer + peerLoop ReceivedAnnounce addr _ -> do void $ serverPeer' server addr @@ -391,13 +421,17 @@ runPacketHandler :: Bool -> Peer -> PacketHandler () -> STM () runPacketHandler secure peer@Peer {..} act = do let logd = writeTQueue $ serverErrorLog peerServer_ - runExceptT (flip execStateT (PacketHandlerState peer [] [] []) $ unPacketHandler act) >>= \case + runExceptT (flip execStateT (PacketHandlerState peer [] [] [] False) $ unPacketHandler act) >>= \case Left err -> do logd $ "Error in handling packet from " ++ show peerAddress ++ ": " ++ err Right ph -> do when (not $ null $ phHead ph) $ do let packet = TransportPacket (TransportHeader $ phHead ph) (phBody ph) - sendToPeerS' secure peer (phAckedBy ph) packet + secreq = case (secure, phPlaintextReply ph) of + (True, _) -> EncryptedOnly + (False, False) -> PlaintextAllowed + (False, True) -> PlaintextOnly + sendToPeerS' secreq peer (phAckedBy ph) packet liftSTM :: STM a -> PacketHandler a liftSTM = PacketHandler . lift . lift @@ -416,6 +450,7 @@ , phHead :: [TransportHeaderItem] , phAckedBy :: [TransportHeaderItem] , phBody :: [Ref] + , phPlaintextReply :: Bool } addHeader :: TransportHeaderItem -> PacketHandler () @@ -427,12 +462,15 @@ addBody :: Ref -> PacketHandler () addBody r = modify $ \ph -> ph { phBody = r `appendDistinct` phBody ph } +keepPlaintextReply :: PacketHandler () +keepPlaintextReply = modify $ \ph -> ph { phPlaintextReply = True } + openStream :: PacketHandler RawStreamWriter openStream = do Peer {..} <- gets phPeer - conn <- readTVarP peerConnection >>= \case - Right conn -> return conn - _ -> throwError "can't open stream without established connection" + conn <- readTVarP peerState >>= \case + PeerConnected conn -> return conn + _ -> throwError "can't open stream without established connection" (hdr, writer, handler) <- liftSTM (connAddWriteStream conn) >>= \case Right res -> return res Left err -> throwError err @@ -444,9 +482,9 @@ acceptStream :: Word8 -> PacketHandler RawStreamReader acceptStream streamNumber = do Peer {..} <- gets phPeer - conn <- readTVarP peerConnection >>= \case - Right conn -> return conn - _ -> throwError "can't accept stream without established connection" + conn <- readTVarP peerState >>= \case + PeerConnected conn -> return conn + _ -> throwError "can't accept stream without established connection" liftSTM $ connAddReadStream conn streamNumber appendDistinct :: Eq a => a -> [a] -> [a] @@ -465,9 +503,9 @@ [ [ storedRef sidentity ] , map storedRef $ idUpdates identity , case ochannel of - ChannelOurRequest _ req -> [ storedRef req ] - ChannelOurAccept _ acc _ -> [ storedRef acc ] - _ -> [] + ChannelOurRequest req -> [ storedRef req ] + ChannelOurAccept acc _ -> [ storedRef acc ] + _ -> [] ] runPacketHandler secure peer $ do @@ -475,7 +513,7 @@ forM_ headers $ \case Acknowledged dgst -> do liftSTM (getPeerChannel peer) >>= \case - ChannelOurAccept _ acc ch | refDigest (storedRef acc) == dgst -> do + ChannelOurAccept acc ch | refDigest (storedRef acc) == dgst -> do liftSTM $ finalizedChannel peer ch identity _ -> return () @@ -489,6 +527,11 @@ partialRefFromDigest (peerInStorage peer) dgst addHeader $ DataResponse dgst addAckedBy [ Acknowledged dgst, Rejected dgst ] + + -- Plaintext request may indicate the peer has restarted/changed or + -- otherwise lost the channel, so keep the reply plaintext as well. + when (not secure) keepPlaintextReply + let bytes = lazyLoadBytes mref -- TODO: MTU if (secure && BL.length bytes > 500) @@ -539,22 +582,23 @@ _ -> return () TrChannelRequest dgst -> do - let process cookie = do + let process = do addHeader $ Acknowledged dgst wref <- newWaitingRef dgst $ handleChannelRequest peer identity - liftSTM $ setPeerChannel peer $ ChannelPeerRequest cookie wref + liftSTM $ setPeerChannel peer $ ChannelPeerRequest wref reject = addHeader $ Rejected dgst liftSTM (getPeerChannel peer) >>= \case ChannelNone {} -> return () ChannelCookieWait {} -> return () - ChannelCookieReceived cookie -> process $ Just cookie - ChannelCookieConfirmed cookie -> process $ Just cookie - ChannelOurRequest mbcookie our | dgst < refDigest (storedRef our) -> process mbcookie - | otherwise -> reject - ChannelPeerRequest mbcookie _ -> process mbcookie + ChannelCookieReceived {} -> process + ChannelCookieConfirmed {} -> process + ChannelOurRequest our | dgst < refDigest (storedRef our) -> process + | otherwise -> reject + ChannelPeerRequest {} -> process ChannelOurAccept {} -> reject - ChannelEstablished {} -> process Nothing + ChannelEstablished {} -> process + ChannelClosed {} -> return () TrChannelAccept dgst -> do let process = do @@ -567,9 +611,10 @@ ChannelCookieConfirmed {} -> reject ChannelOurRequest {} -> process ChannelPeerRequest {} -> process - ChannelOurAccept _ our _ | dgst < refDigest (storedRef our) -> process - | otherwise -> addHeader $ Rejected dgst + ChannelOurAccept our _ | dgst < refDigest (storedRef our) -> process + | otherwise -> addHeader $ Rejected dgst ChannelEstablished {} -> process + ChannelClosed {} -> return () ServiceType _ -> return () ServiceRef dgst @@ -604,10 +649,10 @@ ] liftIO $ atomically $ do getPeerChannel peer >>= \case - ChannelCookieConfirmed cookie -> do + ChannelCookieConfirmed -> do sendToPeerPlain peer [ Acknowledged reqref, Rejected reqref ] $ TransportPacket (TransportHeader hitems) [storedRef req] - setPeerChannel peer $ ChannelOurRequest (Just cookie) req + setPeerChannel peer $ ChannelOurRequest req _ -> return () handleChannelRequest :: Peer -> UnifiedIdentity -> Ref -> WaitingRefCallback @@ -616,8 +661,8 @@ (acc, ch) <- flip runReaderT (peerStorage peer) $ acceptChannelRequest identity upid (wrappedLoad req) liftIO $ atomically $ do getPeerChannel peer >>= \case - ChannelPeerRequest mbcookie wr | wrDigest wr == refDigest req -> do - setPeerChannel peer $ ChannelOurAccept mbcookie acc ch + ChannelPeerRequest wr | wrDigest wr == refDigest req -> do + setPeerChannel peer $ ChannelOurAccept acc ch let accref = refDigest $ storedRef acc header = TrChannelAccept accref ackedBy = [ Acknowledged accref, Rejected accref ] @@ -708,7 +753,7 @@ mkPeer :: Server -> PeerAddress -> IO Peer mkPeer peerServer_ peerAddress = do - peerConnection <- newTVarIO (Left []) + peerState <- newTVarIO (PeerInit []) peerIdentityVar <- newTVarIO . PeerIdentityUnknown =<< newTVarIO [] peerStorage_ <- deriveEphemeralStorage $ serverStorage peerServer_ peerInStorage <- derivePartialStorage peerStorage_ @@ -718,7 +763,11 @@ serverPeer :: Server -> SockAddr -> IO Peer serverPeer server paddr = do - serverPeer' server (DatagramAddress paddr) + let paddr' = case IP.fromSockAddr paddr of + Just (IP.IPv4 ipv4, port) + -> IP.toSockAddr (IP.IPv6 $ IP.toIPv6w (0, 0, 0xffff, IP.fromIPv4w ipv4), port) + _ -> paddr + serverPeer' server (DatagramAddress paddr') #ifdef ENABLE_ICE_SUPPORT serverPeerIce :: Server -> IceSession -> IO Peer @@ -741,7 +790,21 @@ writeFlow (serverControlFlow server) (RequestConnection paddr) return peer +dropPeer :: MonadIO m => Peer -> m () +dropPeer peer = liftIO $ do + modifyMVar_ (serverPeers $ peerServer peer) $ \pvalue -> do + atomically $ do + readTVar (peerState peer) >>= \case + PeerConnected conn -> connClose conn + _ -> return() + writeTVar (peerState peer) PeerDropped + return $ M.delete (peerAddress peer) pvalue +isPeerDropped :: MonadIO m => Peer -> m Bool +isPeerDropped peer = liftIO $ atomically $ readTVar (peerState peer) >>= \case + PeerDropped -> return True + _ -> return False + sendToPeer :: (Service s, MonadIO m) => Peer -> s -> m () sendToPeer peer packet = sendToPeerList peer [ServiceReply (Left packet) True] @@ -762,17 +825,18 @@ ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- dgsts ] liftIO $ atomically $ sendToPeerS peer ackedBy packet -sendToPeerS' :: Bool -> Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () +sendToPeerS' :: SecurityRequirement -> Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () sendToPeerS' secure Peer {..} ackedBy packet = do - readTVar peerConnection >>= \case - Left xs -> writeTVar peerConnection $ Left $ (secure, packet, ackedBy) : xs - Right conn -> writeFlow (connData conn) (secure, packet, ackedBy) + readTVar peerState >>= \case + PeerInit xs -> writeTVar peerState $ PeerInit $ (secure, packet, ackedBy) : xs + PeerConnected conn -> writeFlow (connData conn) (secure, packet, ackedBy) + PeerDropped -> return () sendToPeerS :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () -sendToPeerS = sendToPeerS' True +sendToPeerS = sendToPeerS' EncryptedOnly sendToPeerPlain :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () -sendToPeerPlain = sendToPeerS' False +sendToPeerPlain = sendToPeerS' PlaintextAllowed sendToPeerWith :: forall s m. (Service s, MonadIO m, MonadError String m) => Peer -> (ServiceState s -> ExceptT String IO (Maybe s, ServiceState s)) -> m () sendToPeerWith peer fobj = do diff --git a/src/Erebos/Network/Protocol.hs b/src/Erebos/Network/Protocol.hs --- a/src/Erebos/Network/Protocol.hs +++ b/src/Erebos/Network/Protocol.hs @@ -3,6 +3,7 @@ transportToObject, TransportHeader(..), TransportHeaderItem(..), + SecurityRequirement(..), WaitingRef(..), WaitingRefCallback, @@ -19,6 +20,7 @@ connData, connGetChannel, connSetChannel, + connClose, RawStreamReader, RawStreamWriter, connAddWriteStream, @@ -43,6 +45,7 @@ import Data.ByteString qualified as B import Data.ByteString.Char8 qualified as BC import Data.ByteString.Lazy qualified as BL +import Data.Function import Data.List import Data.Maybe import Data.Text (Text) @@ -93,6 +96,11 @@ newtype Cookie = Cookie ByteString deriving (Eq, Show) +data SecurityRequirement = PlaintextOnly + | PlaintextAllowed + | EncryptedOnly + deriving (Eq, Ord) + isHeaderItemAcknowledged :: TransportHeaderItem -> Bool isHeaderItemAcknowledged = \case Acknowledged {} -> False @@ -165,10 +173,15 @@ data Connection addr = Connection { cGlobalState :: GlobalState addr , cAddress :: addr - , cDataUp :: Flow (Bool, TransportPacket PartialObject) (Bool, TransportPacket Ref, [TransportHeaderItem]) - , cDataInternal :: Flow (Bool, TransportPacket Ref, [TransportHeaderItem]) (Bool, TransportPacket PartialObject) + , cDataUp :: Flow + (Maybe (Bool, TransportPacket PartialObject)) + (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem]) + , cDataInternal :: Flow + (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem]) + (Maybe (Bool, TransportPacket PartialObject)) , cChannel :: TVar ChannelState - , cSecureOutQueue :: TQueue (Bool, TransportPacket Ref, [TransportHeaderItem]) + , cCookie :: TVar (Maybe Cookie) + , cSecureOutQueue :: TQueue (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem]) , cMaxInFlightPackets :: TVar Int , cReservedPackets :: TVar Int , cSentPackets :: TVar [SentPacket] @@ -177,10 +190,15 @@ , cOutStreams :: TVar [(Word8, Stream)] } +instance Eq (Connection addr) where + (==) = (==) `on` cChannel + connAddress :: Connection addr -> addr connAddress = cAddress -connData :: Connection addr -> Flow (Bool, TransportPacket PartialObject) (Bool, TransportPacket Ref, [TransportHeaderItem]) +connData :: Connection addr -> Flow + (Maybe (Bool, TransportPacket PartialObject)) + (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem]) connData = cDataUp connGetChannel :: Connection addr -> STM ChannelState @@ -190,6 +208,16 @@ connSetChannel Connection {..} ch = do writeTVar cChannel ch +connClose :: Connection addr -> STM () +connClose conn@Connection {..} = do + let GlobalState {..} = cGlobalState + readTVar cChannel >>= \case + ChannelClosed -> return () + _ -> do + writeTVar cChannel ChannelClosed + writeTVar gConnections . filter (/=conn) =<< readTVar gConnections + writeFlow cDataInternal Nothing + connAddWriteStream :: Connection addr -> STM (Either String (TransportHeaderItem, RawStreamWriter, IO ())) connAddWriteStream conn@Connection {..} = do outStreams <- readTVar cOutStreams @@ -234,9 +262,9 @@ mbReserved = Just reserved mbch <- atomically (readTVar cChannel) >>= return . \case - ChannelEstablished ch -> Just ch - ChannelOurAccept _ _ ch -> Just ch - _ -> Nothing + ChannelEstablished ch -> Just ch + ChannelOurAccept _ ch -> Just ch + _ -> Nothing mbs <- case mbch of Just ch -> do @@ -367,16 +395,18 @@ data ChannelState = ChannelNone | ChannelCookieWait -- sent initiation, waiting for response - | ChannelCookieReceived Cookie -- received cookie, but no cookie echo yet - | ChannelCookieConfirmed Cookie -- received cookie echo, no need to send from our side - | ChannelOurRequest (Maybe Cookie) (Stored ChannelRequest) - | ChannelPeerRequest (Maybe Cookie) WaitingRef - | ChannelOurAccept (Maybe Cookie) (Stored ChannelAccept) Channel + | ChannelCookieReceived -- received cookie, but no cookie echo yet + | ChannelCookieConfirmed -- received cookie echo, no need to send from our side + | ChannelOurRequest (Stored ChannelRequest) + | ChannelPeerRequest WaitingRef + | ChannelOurAccept (Stored ChannelAccept) Channel | ChannelEstablished Channel + | ChannelClosed data ReservedToSend = ReservedToSend { rsAckedBy :: Maybe (TransportHeaderItem -> Bool) , rsOnAck :: IO () + , rsOnFail :: IO () } data SentPacket = SentPacket @@ -384,6 +414,7 @@ , spRetryCount :: Int , spAckedBy :: Maybe (TransportHeaderItem -> Bool) , spOnAck :: IO () + , spOnFail :: IO () , spData :: BC.ByteString } @@ -450,6 +481,7 @@ let cAddress = addr (cDataUp, cDataInternal) <- newFlow cChannel <- newTVar ChannelNone + cCookie <- newTVar Nothing cSecureOutQueue <- newTQueue cMaxInFlightPackets <- newTVar 4 cReservedPackets <- newTVar 0 @@ -465,7 +497,7 @@ passUpIncoming :: GlobalState addr -> STM (IO ()) passUpIncoming GlobalState {..} = do (Connection {..}, up) <- takeTMVar gNextUp - writeFlow cDataInternal up + writeFlow cDataInternal (Just up) return $ return () processIncoming :: GlobalState addr -> STM (IO ()) @@ -479,9 +511,9 @@ mbch <- case mbconn of Nothing -> return Nothing Just conn -> readTVar (cChannel conn) >>= return . \case - ChannelEstablished ch -> Just ch - ChannelOurAccept _ _ ch -> Just ch - _ -> Nothing + ChannelEstablished ch -> Just ch + ChannelOurAccept _ ch -> Just ch + _ -> Nothing return $ do let deserialize = liftEither . runExcept . deserializeObjects gStorage . BL.fromStrict @@ -576,11 +608,12 @@ True -> do atomically $ do conn@Connection {..} <- getConnection gs addr - channel <- readTVar cChannel + oldCookie <- readTVar cCookie let received = listToMaybe $ mapMaybe (\case CookieSet x -> Just x; _ -> Nothing) header - case received `mplus` channelCurrentCookie channel of + case received `mplus` oldCookie of Just current -> do - cookieEchoReceived gs conn mbpid current + writeTVar cCookie (Just current) + cookieEchoReceived gs conn mbpid return $ Just (conn, Just packet) Nothing -> do gLog $ show addr <> ": missing cookie set, dropping " <> show header @@ -597,7 +630,8 @@ -> do atomically $ readTVar cChannel >>= \case ChannelCookieWait -> do - writeTVar cChannel $ ChannelCookieReceived cookie + writeTVar cChannel $ ChannelCookieReceived + writeTVar cCookie $ Just cookie writeFlow gControlFlow (NewConnection conn mbpid) return $ Just (conn, Nothing) _ -> return Nothing @@ -636,17 +670,8 @@ mbpid = listToMaybe $ mapMaybe (\case AnnounceSelf dgst -> Just dgst; _ -> Nothing) header version = listToMaybe $ filter (\v -> ProtocolVersion v `elem` header) protocolVersions -channelCurrentCookie :: ChannelState -> Maybe Cookie -channelCurrentCookie = \case - ChannelCookieReceived cookie -> Just cookie - ChannelCookieConfirmed cookie -> Just cookie - ChannelOurRequest mbcookie _ -> mbcookie - ChannelPeerRequest mbcookie _ -> mbcookie - ChannelOurAccept mbcookie _ _ -> mbcookie - _ -> Nothing - -cookieEchoReceived :: GlobalState addr -> Connection addr -> Maybe RefDigest -> Cookie -> STM () -cookieEchoReceived GlobalState {..} conn@Connection {..} mbpid cookieSet = do +cookieEchoReceived :: GlobalState addr -> Connection addr -> Maybe RefDigest -> STM () +cookieEchoReceived GlobalState {..} conn@Connection {..} mbpid = do readTVar cChannel >>= \case ChannelNone -> newConn ChannelCookieWait -> newConn @@ -654,18 +679,18 @@ _ -> return () where update = do - writeTVar cChannel $ ChannelCookieConfirmed cookieSet + writeTVar cChannel ChannelCookieConfirmed newConn = do update writeFlow gControlFlow (NewConnection conn mbpid) -generateCookieHeaders :: GlobalState addr -> addr -> ChannelState -> IO [TransportHeaderItem] -generateCookieHeaders gs addr ch = catMaybes <$> sequence [ echoHeader, setHeader ] +generateCookieHeaders :: Connection addr -> ChannelState -> IO [TransportHeaderItem] +generateCookieHeaders Connection {..} ch = catMaybes <$> sequence [ echoHeader, setHeader ] where - echoHeader = return $ CookieEcho <$> channelCurrentCookie ch + echoHeader = fmap CookieEcho <$> atomically (readTVar cCookie) setHeader = case ch of - ChannelCookieWait {} -> Just . CookieSet <$> createCookie gs addr - ChannelCookieReceived {} -> Just . CookieSet <$> createCookie gs addr + ChannelCookieWait {} -> Just . CookieSet <$> createCookie cGlobalState cAddress + ChannelCookieReceived {} -> Just . CookieSet <$> createCookie cGlobalState cAddress _ -> return Nothing createCookie :: GlobalState addr -> addr -> IO Cookie @@ -676,7 +701,7 @@ reservePacket :: Connection addr -> STM ReservedToSend -reservePacket Connection {..} = do +reservePacket conn@Connection {..} = do maxPackets <- readTVar cMaxInFlightPackets reserved <- readTVar cReservedPackets sent <- length <$> readTVar cSentPackets @@ -685,7 +710,7 @@ retry writeTVar cReservedPackets $ reserved + 1 - return $ ReservedToSend Nothing (return ()) + return $ ReservedToSend Nothing (return ()) (atomically $ connClose conn) resendBytes :: Connection addr -> Maybe ReservedToSend -> SentPacket -> IO () resendBytes Connection {..} reserved sp = do @@ -709,6 +734,7 @@ , spRetryCount = -1 , spAckedBy = rsAckedBy =<< reserved , spOnAck = maybe (return ()) rsOnAck reserved + , spOnFail = maybe (return ()) rsOnFail reserved , spData = bs } @@ -734,13 +760,13 @@ | isJust mbch = do acks <- readTVar cToAcknowledge if null acks then retry - else return ((True, TransportPacket (TransportHeader []) [], []), Nothing) + else return ((EncryptedOnly, TransportPacket (TransportHeader []) [], []), Nothing) | otherwise = retry ((secure, packet@(TransportPacket (TransportHeader hitems) content), plainAckedBy), mbReserved) <- checkOutstanding <|> checkDataInternal <|> checkAcknowledgements - when (isNothing mbch && secure) $ do + when (isNothing mbch && secure >= EncryptedOnly) $ do writeTQueue cSecureOutQueue (secure, packet, plainAckedBy) acknowledge <- case mbch of @@ -748,28 +774,34 @@ Just _ -> swapTVar cToAcknowledge [] return $ do - cookieHeaders <- generateCookieHeaders gs cAddress channel - let header = TransportHeader $ map AcknowledgedSingle acknowledge ++ cookieHeaders ++ hitems + let onAck = sequence_ $ map (streamAccepted conn) $ + catMaybes (map (\case StreamOpen n -> Just n; _ -> Nothing) hitems) - let plain = BL.concat $ - (serializeObject $ transportToObject gStorage header) - : map lazyLoadBytes content + let mkPlain extraHeaders = + let header = TransportHeader $ map AcknowledgedSingle acknowledge ++ extraHeaders ++ hitems + in BL.concat $ + (serializeObject $ transportToObject gStorage header) + : map lazyLoadBytes content - let onAck = case catMaybes (map (\case StreamOpen n -> Just n; _ -> Nothing) hitems) of - [] -> return () - xs -> sequence_ $ map (streamAccepted conn) xs + let usePlaintext = do + plain <- mkPlain <$> generateCookieHeaders conn channel + return $ Just (BL.toStrict plain, plainAckedBy) - mbs <- case mbch of - Just ch -> do + let useEncryption ch = do + plain <- mkPlain <$> return [] runExceptT (channelEncrypt ch $ BL.toStrict $ 0x00 `BL.cons` plain) >>= \case Right (ctext, counter) -> do let isAcked = any isHeaderItemAcknowledged hitems return $ Just (0x80 `B.cons` ctext, if isAcked then [ AcknowledgedSingle $ fromIntegral counter ] else []) Left err -> do atomically $ gLog $ "Failed to encrypt data: " ++ err return Nothing - Nothing | secure -> return Nothing - | otherwise -> return $ Just (BL.toStrict plain, plainAckedBy) + mbs <- case (secure, mbch) of + (PlaintextOnly, _) -> usePlaintext + (PlaintextAllowed, Nothing) -> usePlaintext + (_, Just ch) -> useEncryption ch + (EncryptedOnly, Nothing) -> return Nothing + case mbs of Just (bs, ackedBy) -> do let mbReserved' = (\rs -> rs @@ -786,17 +818,18 @@ sps@(_:_) -> return (last sps, init sps) _ -> retry let nextTry = spTime sp + fromNanoSecs 1000000000 - if now < nextTry - then do - nextTimeout <- readTVar gNextTimeout - if nextTimeout <= now || nextTry < nextTimeout - then do writeTVar gNextTimeout nextTry - return $ return () - else retry - else do - reserved <- reservePacket conn - writeTVar cSentPackets rest - return $ resendBytes conn (Just reserved) sp + if | now < nextTry -> do + nextTimeout <- readTVar gNextTimeout + if nextTimeout <= now || nextTry < nextTimeout + then do writeTVar gNextTimeout nextTry + return $ return () + else retry + | spRetryCount sp < 2 -> do + reserved <- reservePacket conn + writeTVar cSentPackets rest + return $ resendBytes conn (Just reserved) sp + | otherwise -> do + return $ spOnFail sp let handleControlRequests = readFlow gControlFlow >>= \case RequestConnection addr -> do diff --git a/src/Erebos/Storage.hs b/src/Erebos/Storage.hs --- a/src/Erebos/Storage.hs +++ b/src/Erebos/Storage.hs @@ -112,10 +112,29 @@ type Storage = Storage' Complete type PartialStorage = Storage' Partial +storageVersion :: String +storageVersion = "0.1" + openStorage :: FilePath -> IO Storage -openStorage path = do - createDirectoryIfMissing True $ path ++ "/objects" - createDirectoryIfMissing True $ path ++ "/heads" +openStorage path = modifyIOError annotate $ do + let versionPath = path "erebos-storage" + doesFileExist versionPath >>= \case + True -> readFile versionPath >>= \case + content | (ver:_) <- lines content, ver == storageVersion -> return () + | otherwise -> fail "unsupported storage version" + False -> do + doesDirectoryExist path >>= \case + True -> do + listDirectory path >>= \case + contents@(_:_) | "objects" `notElem` contents || "heads" `notElem` contents + -> fail "directory is neither empty, nor an existing erebos storage" + _ -> return () + False -> do + createDirectoryIfMissing True $ path + writeFile versionPath $ storageVersion <> "\n" + + createDirectoryIfMissing True $ path "objects" + createDirectoryIfMissing True $ path "heads" watchers <- newMVar ([], WatchList 1 []) refgen <- newMVar =<< HT.new refroots <- newMVar =<< HT.new @@ -125,6 +144,8 @@ , stRefGeneration = refgen , stRefRoots = refroots } + where + annotate e = annotateIOError e "failed to open storage" Nothing (Just path) memoryStorage' :: IO (Storage' c') memoryStorage' = do