diff --git a/apps/simplex-directory-service/src/Directory/Options.hs b/apps/simplex-directory-service/src/Directory/Options.hs index 89709e66d9..03f4d7bb0c 100644 --- a/apps/simplex-directory-service/src/Directory/Options.hs +++ b/apps/simplex-directory-service/src/Directory/Options.hs @@ -41,6 +41,7 @@ data DirectoryOpts = DirectoryOpts linkCheckInterval :: Int, prohibitedToObserver :: Bool, alwaysCaptcha :: Bool, + alwaysObserver :: Bool, knocking :: Bool, testing :: Bool } @@ -170,6 +171,11 @@ directoryOpts appDir defaultDbName = do ( long "always-captcha" <> help "Require a captcha from joining members in all groups, regardless of per-group filter settings" ) + alwaysObserver <- + switch + ( long "always-observer" + <> help "Make joining members observers in all groups, regardless of per-group setting in directory" + ) knocking <- switch ( long "knocking" @@ -197,6 +203,7 @@ directoryOpts appDir defaultDbName = do linkCheckInterval, prohibitedToObserver, alwaysCaptcha, + alwaysObserver, knocking, testing = False } diff --git a/apps/simplex-directory-service/src/Directory/Service.hs b/apps/simplex-directory-service/src/Directory/Service.hs index 61ed7e66af..b2a42c1eb2 100644 --- a/apps/simplex-directory-service/src/Directory/Service.hs +++ b/apps/simplex-directory-service/src/Directory/Service.hs @@ -268,7 +268,7 @@ directoryService opts cfg = do acceptMemberHook :: DirectoryOpts -> ServiceState -> GroupInfo -> GroupLinkInfo -> Profile -> IO (Either GroupRejectionReason (GroupAcceptance, GroupMemberRole)) acceptMemberHook - DirectoryOpts {profileNameLimit, alwaysCaptcha, knocking} + DirectoryOpts {profileNameLimit, alwaysCaptcha, alwaysObserver, knocking} ServiceState {blockedWordsCfg} g GroupLinkInfo {memberRole} @@ -279,7 +279,7 @@ acceptMemberHook if | knocking -> (GAPendingReview, memberRole) | alwaysCaptcha || useMemberFilter img (passCaptcha a) -> (GAPendingApproval, GRMember) - | useMemberFilter img (makeObserver a) -> (GAAccepted, GRObserver) + | alwaysObserver || useMemberFilter img (makeObserver a) -> (GAAccepted, GRObserver) | otherwise -> (GAAccepted, memberRole) where checkName :: ExceptT GroupRejectionReason IO () @@ -314,7 +314,7 @@ readBlockedWordsConfig DirectoryOpts {blockedFragmentsFile, blockedWordsFile, na pure BlockedWordsConfig {blockedFragments, blockedWords, extensionRules, spelling} directoryServiceEvent :: DirectoryOpts -> ServiceState -> User -> ChatController -> DirectoryEvent -> IO () -directoryServiceEvent opts@DirectoryOpts {adminUsers, superUsers, serviceName, ownersGroup, searchResults, prohibitedToObserver, alwaysCaptcha} env@ServiceState {searchRequests} user@User {userId} cc = \case +directoryServiceEvent opts@DirectoryOpts {adminUsers, superUsers, serviceName, ownersGroup, searchResults, prohibitedToObserver, alwaysCaptcha, alwaysObserver} env@ServiceState {searchRequests} user@User {userId} cc = \case DEContactConnected ct -> deContactConnected ct DEGroupInvitation {contact = ct, groupInfo = g, fromMemberRole, memberRole} -> deGroupInvitation ct g fromMemberRole memberRole DEServiceJoinedGroup ctId g owner -> deServiceJoinedGroup ctId g owner @@ -678,7 +678,7 @@ directoryServiceEvent opts@DirectoryOpts {adminUsers, superUsers, serviceName, o approvePendingMember :: DirectoryMemberAcceptance -> GroupInfo -> GroupMember -> IO () approvePendingMember a g@GroupInfo {groupId} m@GroupMember {memberProfile = LocalProfile {displayName, image}} = do gli_ <- join . eitherToMaybe <$> withDB' "getGroupLinkInfo" cc (\db -> getGroupLinkInfo db userId groupId) - let role = if useMemberFilter image (makeObserver a) then GRObserver else maybe GRMember (\GroupLinkInfo {memberRole} -> memberRole) gli_ + let role = if alwaysObserver || useMemberFilter image (makeObserver a) then GRObserver else maybe GRMember (\GroupLinkInfo {memberRole} -> memberRole) gli_ gmId = groupMemberId' m sendChatCmd cc (APIAcceptMember groupId gmId role) >>= \case Right CRMemberAccepted {member} -> do diff --git a/src/Simplex/Chat/Controller.hs b/src/Simplex/Chat/Controller.hs index 4be017bf9e..e69bb7e6ac 100644 --- a/src/Simplex/Chat/Controller.hs +++ b/src/Simplex/Chat/Controller.hs @@ -95,7 +95,7 @@ import Simplex.Messaging.Protocol (AProtoServerWithAuth, AProtocolType (..), Msg import Simplex.Messaging.TMap (TMap) import Simplex.Messaging.Transport (TLS, TransportPeer (..), simplexMQVersion) import Simplex.Messaging.Transport.Client (SocksProxyWithAuth, TransportHost) -import Simplex.Messaging.Util (AnyError (..), catchAllErrors, (<$$>)) +import Simplex.Messaging.Util (AnyError (..), catchAllErrors, catchOwn', (<$$>)) import Simplex.RemoteControl.Client import Simplex.RemoteControl.Invitation (RCSignedInvitation, RCVerifiedInvitation) import Simplex.RemoteControl.Types @@ -1755,12 +1755,12 @@ withFastStore = withStorePriority True withStorePriority :: Bool -> (DB.Connection -> ExceptT StoreError IO a) -> CM a withStorePriority priority action = do ChatController {chatStore} <- ask - liftIOEither $ withTransactionPriority chatStore priority (runExceptT . withExceptT ChatErrorStore . action) `E.catch` handleDBErrors + liftIOEither $ withTransactionPriority chatStore priority (runExceptT . withExceptT ChatErrorStore . action) `catchOwn'` handleDBErrors withStoreBatch :: Traversable t => (DB.Connection -> t (IO (Either ChatError a))) -> CM' (t (Either ChatError a)) withStoreBatch actions = do ChatController {chatStore} <- ask - liftIO $ withTransaction chatStore $ mapM (`E.catch` handleDBErrors) . actions + liftIO $ withTransaction chatStore $ mapM (`catchOwn'` handleDBErrors) . actions handleDBErrors :: E.SomeException -> IO (Either ChatError a) handleDBErrors e = pure $ Left $ ChatErrorStore $ case E.fromException e of diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index 01fc4b55ff..4e4cb92afc 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/src/Simplex/Chat/Library/Commands.hs @@ -123,23 +123,24 @@ import Simplex.RemoteControl.Types (RCCtrlAddress (..)) import System.Exit (ExitCode, exitSuccess) import System.FilePath (takeExtension, takeFileName, ()) import System.IO (Handle, IOMode (..)) +import System.Mem.Weak (deRefWeak) import System.Random (randomRIO) import System.Timeout (timeout) import UnliftIO.Async -import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO.Concurrent (forkIO, killThread, threadDelay) import UnliftIO.Directory import qualified UnliftIO.Exception as E import UnliftIO.IO (hClose) import UnliftIO.STM #if defined(dbPostgres) import Data.Bifunctor (bimap, first, second) -import Simplex.Messaging.Agent.Client (SubInfo (..), getAgentQueuesInfo, getAgentWorkersDetails, getAgentWorkersSummary, temporaryOrHostError) +import Simplex.Messaging.Agent.Client (SubInfo (..), cancelWorker, getAgentQueuesInfo, getAgentWorkersDetails, getAgentWorkersSummary, temporaryOrHostError) #else import Data.Bifunctor (bimap, first, second) import qualified Data.ByteArray as BA import qualified Database.SQLite.Simple as SQL import Simplex.Chat.Archive -import Simplex.Messaging.Agent.Client (SubInfo (..), agentClientStore, getAgentQueuesInfo, getAgentWorkersDetails, getAgentWorkersSummary, temporaryOrHostError) +import Simplex.Messaging.Agent.Client (SubInfo (..), agentClientStore, cancelWorker, getAgentQueuesInfo, getAgentWorkersDetails, getAgentWorkersSummary, temporaryOrHostError) import Simplex.Messaging.Agent.Store.Common (withConnection) import Simplex.Messaging.Agent.Store.SQLite.DB (SlowQueryStats (..)) #endif @@ -346,11 +347,19 @@ restoreCalls = do atomically $ writeTVar calls callsMap stopChatController :: ChatController -> IO () -stopChatController ChatController {smpAgent, agentAsync = s, sndFiles, rcvFiles, expireCIFlags, remoteHostSessions, remoteCtrlSession} = do +stopChatController ChatController {smpAgent, agentAsync = s, sndFiles, rcvFiles, expireCIFlags, remoteHostSessions, remoteCtrlSession, cleanupManagerAsync, relayGroupLinkChecksAsync, webPreviewState, expireCIThreads, timedItemThreads, deliveryTaskWorkers, deliveryJobWorkers, relayRequestWorkers} = do readTVarIO remoteHostSessions >>= mapM_ (cancelRemoteHost False . snd) atomically (stateTVar remoteCtrlSession (,Nothing)) >>= mapM_ (cancelRemoteCtrl False . snd) disconnectAgentClient smpAgent - readTVarIO s >>= mapM_ (\(a1, a2) -> forkIO $ uninterruptibleCancel a1 >> mapM_ uninterruptibleCancel a2) + readTVarIO s >>= mapM_ (\(a1, a2) -> uninterruptibleCancel a1 >> mapM_ uninterruptibleCancel a2) + cancelAsync cleanupManagerAsync + cancelAsync relayGroupLinkChecksAsync + forM_ webPreviewState $ \WebPreviewState {webPreviewWorkerAsync} -> cancelAsync webPreviewWorkerAsync + clearMap expireCIThreads >>= mapM_ (mapM_ uninterruptibleCancel) + clearMap timedItemThreads >>= mapM_ (readTVarIO >=> mapM_ (deRefWeak >=> mapM_ killThread)) + clearMap deliveryTaskWorkers >>= mapM_ cancelWorker + clearMap deliveryJobWorkers >>= mapM_ cancelWorker + clearMap relayRequestWorkers >>= mapM_ cancelWorker closeFiles sndFiles closeFiles rcvFiles atomically $ do @@ -358,6 +367,10 @@ stopChatController ChatController {smpAgent, agentAsync = s, sndFiles, rcvFiles, forM_ keys $ \k -> TM.insert k False expireCIFlags writeTVar s Nothing where + cancelAsync :: TVar (Maybe (Async ())) -> IO () + cancelAsync a = atomically (swapTVar a Nothing) >>= mapM_ uninterruptibleCancel + clearMap :: TM.TMap k a -> IO (Map k a) + clearMap m = atomically $ swapTVar m M.empty closeFiles :: TVar (Map Int64 Handle) -> IO () closeFiles files = do fs <- readTVarIO files diff --git a/src/Simplex/Chat/Web.hs b/src/Simplex/Chat/Web.hs index fc3e4b2a26..9720116d91 100644 --- a/src/Simplex/Chat/Web.hs +++ b/src/Simplex/Chat/Web.hs @@ -24,7 +24,7 @@ module Simplex.Chat.Web where import Control.Concurrent.STM (check, flushTQueue) -import Control.Exception (SomeException, catch) +import Control.Exception (SomeException) import Control.Logger.Simple import Control.Monad import Control.Monad.Except (runExceptT) @@ -75,7 +75,7 @@ import Simplex.Chat.Types ) import Simplex.Messaging.Agent.Store.Common (withTransaction) import Simplex.Messaging.Encoding.String (strEncode) -import Simplex.Messaging.Util (catchOwn, eitherToMaybe, safeDecodeUtf8, tshow) +import Simplex.Messaging.Util (catchOwn, catchOwn', eitherToMaybe, safeDecodeUtf8, tshow) import Simplex.Messaging.Parsers (defaultJSON) import System.Directory (createDirectoryIfMissing, listDirectory, removeFile, renameFile) import System.FilePath (dropExtension, takeExtension, ()) @@ -150,7 +150,7 @@ webPreviewWorker cfg@WebPreviewConfig {webJsonDir, webCorsFile, webUpdateInterva drainRemovals = atomically (tryReadTQueue filesToRemove) >>= \case Nothing -> pure () Just f -> do - removeFile (webJsonDir f) `catch` \(_ :: SomeException) -> pure () + removeFile (webJsonDir f) `catchOwn'` \(_ :: SomeException) -> pure () drainRemovals -- flush the whole queue and render each group once: a burst of changes in one @@ -202,7 +202,7 @@ webPreviewWorker cfg@WebPreviewConfig {webJsonDir, webCorsFile, webUpdateInterva renderOneGroup WebPreviewState {publishableGroupIds} gId = do publishable <- atomically $ M.member gId <$> readTVar publishableGroupIds when publishable $ - renderOrRemoveStale `catch` \(e :: SomeException) -> + renderOrRemoveStale `catchOwn'` \(e :: SomeException) -> logError $ "web preview: error rendering group " <> T.pack (show gId) <> ": " <> T.pack (show e) where renderOrRemoveStale = do @@ -217,7 +217,7 @@ webPreviewWorker cfg@WebPreviewConfig {webJsonDir, webCorsFile, webUpdateInterva modifyTVar' publishableGroupIds (M.delete gId) pure $ pgFileName <$> pg forM_ fName $ \f -> - removeFile (webJsonDir f) `catch` \(_ :: SomeException) -> pure () + removeFile (webJsonDir f) `catchOwn'` \(_ :: SomeException) -> pure () logInfo $ "web preview: group " <> T.pack (show gId) <> " no longer publishable" findUser f = go users diff --git a/tests/Bots/DirectoryTests.hs b/tests/Bots/DirectoryTests.hs index 89ee276dfa..65a9e153ba 100644 --- a/tests/Bots/DirectoryTests.hs +++ b/tests/Bots/DirectoryTests.hs @@ -78,6 +78,7 @@ directoryServiceTests = do describe "member admission" $ do it "should require captcha by default for new groups" testCaptchaByDefault it "should require captcha in all groups with --always-captcha" testAlwaysCaptcha + it "should make joining members observers in all groups with --always-observer" testAlwaysObserver it "should require admin review in all groups with --knocking" testKnocking it "should ask member to pass captcha screen" testCapthaScreening it "should send voice captcha on /audio command" testVoiceCaptchaScreening @@ -144,6 +145,7 @@ mkDirectoryOpts ps superUsers ownersGroup webFolder = linkCheckInterval = 0, prohibitedToObserver = False, alwaysCaptcha = False, + alwaysObserver = False, knocking = False, testing = True } @@ -1260,6 +1262,34 @@ testAlwaysCaptcha ps = bob <## "#privacy: 'SimpleX Directory' added cath (Catherine) to the group (connecting...)" bob <## "#privacy: new member cath is connected" +testAlwaysObserver :: HasCallStack => TestParams -> IO () +testAlwaysObserver ps = + withDirectoryServiceOpts ps (\o -> o {alwaysObserver = True}) $ \superUser dsLink -> + withNewTestChat ps "bob" bobProfile $ \bob -> + withNewTestChat ps "cath" cathProfile $ \cath -> do + bob `connectVia` dsLink + submitGroup bob "privacy" "Privacy" + groupAccepted bob "privacy" 1 + welcomeWithLink <- completeRegistration superUser bob "privacy" "Privacy" 1 + let groupLink = dropStrPrefix "Link to join the group privacy: " welcomeWithLink + cath ##> ("/c " <> groupLink) + cath <## "connection request sent!" + cath <## "#privacy: joining the group..." + cath <## "#privacy: you joined the group, pending approval" + cath <# "#privacy (support) 'SimpleX Directory'> Captcha is generated by SimpleX Directory service." + cath <## "" + cath <## "Send captcha text to join the group privacy." + captcha <- dropStrPrefix "#privacy (support) 'SimpleX Directory'> " . dropTime <$> getTermLine cath + cath #> ("#privacy (support) " <> captcha) + cath <# ("#privacy (support) 'SimpleX Directory'!> > cath " <> captcha) + cath <## " Correct, you joined the group privacy" + cath <## "#privacy: you joined the group" + cath <## "#privacy: member bob (Bob) is connected" + bob <## "#privacy: 'SimpleX Directory' added cath (Catherine) to the group (connecting...)" + bob <## "#privacy: new member cath is connected" + cath ##> "#privacy hello" + cath <## "#privacy: you don't have permission to send messages" + testKnocking :: HasCallStack => TestParams -> IO () testKnocking ps = withDirectoryServiceOpts ps (\o -> o {knocking = True}) $ \superUser dsLink -> diff --git a/tests/ChatTests/Direct.hs b/tests/ChatTests/Direct.hs index a929bc3a67..5e5512573c 100644 --- a/tests/ChatTests/Direct.hs +++ b/tests/ChatTests/Direct.hs @@ -15,18 +15,21 @@ import ChatClient import ChatTests.DBUtils import ChatTests.Utils import Control.Concurrent (threadDelay) -import Control.Concurrent.Async (concurrently_) -import Control.Monad (forM_, void) +import Control.Concurrent.Async (concurrently_, poll) +import Control.Monad (forM_, void, (>=>)) import Data.Aeson (ToJSON) import qualified Data.Aeson as J import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy.Char8 as LB import Data.List (intercalate, stripPrefix) +import qualified Data.Map.Strict as M +import Data.Maybe (isJust, isNothing) import qualified Data.Text as T +import GHC.Conc (ThreadStatus (..), threadStatus) import Simplex.Chat.AppSettings (defaultAppSettings) import qualified Simplex.Chat.AppSettings as AS import Simplex.Chat.Call -import Simplex.Chat.Controller (ChatConfig (..), PresetServers (..)) +import Simplex.Chat.Controller (ChatConfig (..), ChatController (..), PresetServers (..)) import Simplex.Chat.Messages (ChatItemId) import Simplex.Chat.Options import Simplex.Chat.Protocol (supportedChatVRange) @@ -35,7 +38,7 @@ import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.RetryInterval import qualified Simplex.Messaging.Agent.Store.DB as DB import Simplex.Messaging.Client (NetworkTimeout (..)) -import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM (atomically, readTVarIO) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding.String (strEncode) import Simplex.Messaging.Server.Env.STM hiding (subscriptions) @@ -43,6 +46,7 @@ import Simplex.Messaging.Transport import Simplex.Messaging.Util (safeDecodeUtf8) import Simplex.Messaging.Version import System.Directory (copyFile, doesDirectoryExist, doesFileExist) +import System.Mem.Weak (deRefWeak) import Test.Hspec hiding (it) #if defined(dbPostgres) import Database.PostgreSQL.Simple (Only (..)) @@ -101,8 +105,9 @@ chatDirectTests = do it "connect, fully asynchronous (when clients are never simultaneously online)" $ testFullAsyncFast describe "webrtc calls api" $ do it "negotiate call" testNegotiateCall -#if !defined(dbPostgres) describe "maintenance mode" $ do + it "stop chat stops all threads, start chat restarts them" testStopStartChat +#if !defined(dbPostgres) it "start/stop/export/import chat" testMaintenanceMode it "export/import chat with files" testMaintenanceModeWithFiles it "encrypt/decrypt database" testDatabaseEncryption @@ -1358,6 +1363,52 @@ testNegotiateCall = alice <## "call with bob ended" alice #$> ("/_get chat @2 count=100", chat, chatFeatures <> [(1, "outgoing call: ended (00:00)")]) +testStopStartChat :: HasCallStack => TestParams -> IO () +testStopStartChat ps = + withNewTestChat ps "bob" bobProfile $ \bob -> + withNewTestChatCfg ps cfg "alice" aliceProfile $ \alice -> do + connectUsers alice bob + alice #> "@bob hi" + bob <# "alice> hi" + alice #$> ("/_ttl 1 4", id, "ok") + alice ##> "/_set prefs @2 {\"timedMessages\": {\"allow\": \"yes\", \"ttl\": 2}}" + alice <## "you updated preferences for bob:" + alice <## "Disappearing messages: enabled (you allow: yes (2 sec), contact allows: yes)" + bob <## "alice updated preferences for you:" + bob <## "Disappearing messages: enabled (you allow: yes (2 sec), contact allows: yes (2 sec))" + alice #> "@bob hi timed" + bob <# "alice> hi timed" + let ChatController {agentAsync, cleanupManagerAsync, expireCIThreads, timedItemThreads} = chatController alice + Just (a1, Just a2) <- readTVarIO agentAsync + Just cleanupA <- readTVarIO cleanupManagerAsync + [Just expireA] <- M.elems <$> readTVarIO expireCIThreads + [Just timedTId] <- mapM (readTVarIO >=> maybe (pure Nothing) deRefWeak) . M.elems =<< readTVarIO timedItemThreads + alice ##> "/_stop" + alice <## "chat stopped" + forM_ [a1, a2, cleanupA, expireA] $ \a -> isJust <$> poll a `shouldReturn` True + threadDelay 100000 + threadStatus timedTId `shouldReturn` ThreadFinished + isNothing <$> readTVarIO agentAsync `shouldReturn` True + isNothing <$> readTVarIO cleanupManagerAsync `shouldReturn` True + M.null <$> readTVarIO expireCIThreads `shouldReturn` True + M.null <$> readTVarIO timedItemThreads `shouldReturn` True + alice ##> "/_start" + alice <## "chat started" + alice <## "subscribed 1 connections on server localhost" + bob #> "@alice hello" + alice <# "bob> hello" + alice <### ["timed message deleted: hi timed", "timed message deleted: hello"] + bob <### ["timed message deleted: hi timed", "timed message deleted: hello"] + threadDelay 3000000 + alice #$> ("/_get chat @2 count=100", chat, [(1, "chat banner")]) + Just (a1', _) <- readTVarIO agentAsync + (a1' == a1) `shouldBe` False + Just cleanupA' <- readTVarIO cleanupManagerAsync + (cleanupA' == cleanupA) `shouldBe` False + M.keys <$> readTVarIO expireCIThreads `shouldReturn` [1] + where + cfg = testCfg {initialCleanupManagerDelay = 0, cleanupManagerStepDelay = 0, ciExpirationInterval = 500000} + testMaintenanceMode :: HasCallStack => TestParams -> IO () testMaintenanceMode ps = do withNewTestChat ps "bob" bobProfile $ \bob -> do