[WPB-6144] Prevent MLS one-to-one messaging for a blocking user (q1-2024)

cabal.project

@@ -31,6 +31,7 @@ packages:
   , libs/wire-api/
   , libs/wire-api-federation/
   , libs/wire-message-proto-lens/
+  , libs/wire-subsystems/
   , libs/zauth/
   , services/background-worker/
   , services/brig/
@@ -162,6 +163,8 @@ package wire-api-federation
     ghc-options: -Werror
 package wire-message-proto-lens
     ghc-options: -Werror
+package wire-subsystems
+    ghc-options: -Werror
 package zauth
     ghc-options: -Werror
 package fedcalls

changelog.d/3-bug-fixes/wpb-6144-messaging-blocked-user

@@ -0,0 +1 @@
+Do not deliver MLS one-to-one conversation messages to a user that blocked the sender

changelog.d/5-internal/notification-subsystem

@@ -0,0 +1 @@
+Start refactoring code into subsystems, first subsystem being the NotificationSubsystem.

integration/test/Test/MLS/One2One.hs

@@ -17,6 +17,7 @@
 
 module Test.MLS.One2One where
 
+import API.Brig
 import API.Galley
 import qualified Data.ByteString.Base64 as Base64
 import qualified Data.ByteString.Char8 as B8
@@ -54,6 +55,52 @@ testGetMLSOne2OneUnconnected otherDomain = do
   bindResponse (getMLSOne2OneConversation alice bob) $ \resp ->
     resp.status `shouldMatchInt` 403
 
+testMLSOne2OneBlocked :: HasCallStack => Domain -> App ()
+testMLSOne2OneBlocked otherDomain = do
+  [alice, bob] <- for [OwnDomain, otherDomain] $ flip randomUser def
+  void $ postConnection bob alice >>= getBody 201
+  void $ putConnection alice bob "blocked" >>= getBody 200
+  void $ getMLSOne2OneConversation alice bob >>= getJSON 403
+  void $ getMLSOne2OneConversation bob alice >>= getJSON 403
+
+-- | Alice and Bob are initially connected, but then Alice blocks Bob.
+testMLSOne2OneBlockedAfterConnected :: HasCallStack => One2OneScenario -> App ()
+testMLSOne2OneBlockedAfterConnected scenario = do
+  alice <- randomUser OwnDomain def
+  let otherDomain = one2OneScenarioDomain scenario
+      convDomain = one2OneScenarioConvDomain scenario
+  bob <- createMLSOne2OnePartner otherDomain alice convDomain
+  conv <- getMLSOne2OneConversation alice bob >>= getJSON 200
+  convId <- conv %. "qualified_id"
+  do
+    bobConv <- getMLSOne2OneConversation bob alice >>= getJSON 200
+    convId `shouldMatch` (bobConv %. "qualified_id")
+
+  [alice1, bob1] <- traverse (createMLSClient def) [alice, bob]
+  traverse_ uploadNewKeyPackage [bob1]
+  resetGroup alice1 conv
+  commit <- createAddCommit alice1 [bob]
+  withWebSocket bob1 $ \ws -> do
+    void $ sendAndConsumeCommitBundle commit
+    let isMessage n = nPayload n %. "type" `isEqual` "conversation.mls-welcome"
+    n <- awaitMatch isMessage ws
+    nPayload n %. "data" `shouldMatch` B8.unpack (Base64.encode (fold commit.welcome))
+
+  withWebSocket bob1 $ \ws -> do
+    -- Alice blocks Bob
+    void $ putConnection alice bob "blocked" >>= getBody 200
+    -- There is also a proteus 1-to-1 conversation. Neither it nor the MLS
+    -- 1-to-1 conversation should get any events.
+    awaitAnyEvent 2 ws `shouldMatch` (Nothing :: Maybe Value)
+    -- Alice is not in the MLS 1-to-1 conversation given that she has blocked
+    -- Bob.
+    void $ getMLSOne2OneConversation alice bob >>= getJSON 403
+
+  mp <- createApplicationMessage bob1 "hello, world, again"
+  withWebSocket alice1 $ \ws -> do
+    void $ postMLSMessage mp.sender mp.message >>= getJSON 201
+    awaitAnyEvent 2 ws `shouldMatch` (Nothing :: Maybe Value)
+
 testGetMLSOne2OneSameTeam :: App ()
 testGetMLSOne2OneSameTeam = do
   (alice, _, _) <- createTeam OwnDomain 1

integration/test/Testlib/Cannon.hs

@@ -28,6 +28,7 @@ module Testlib.Cannon
     awaitNMatchesResult,
     awaitNMatches,
     awaitMatch,
+    awaitAnyEvent,
     awaitAtLeastNMatchesResult,
     awaitAtLeastNMatches,
     awaitNToMMatchesResult,
@@ -282,7 +283,7 @@ printAwaitResult = prettyAwaitResult >=> liftIO . putStrLn
 printAwaitAtLeastResult :: AwaitAtLeastResult -> App ()
 printAwaitAtLeastResult = prettyAwaitAtLeastResult >=> liftIO . putStrLn
 
-awaitAnyEvent :: MonadIO m => Int -> WebSocket -> m (Maybe Value)
+awaitAnyEvent :: Int -> WebSocket -> App (Maybe Value)
 awaitAnyEvent tSecs = liftIO . timeout (tSecs * 1000 * 1000) . atomically . readTChan . wsChan
 
 -- | 'await' an expected number of notification events on the websocket that

libs/extended/default.nix

@@ -31,6 +31,7 @@
 , servant-server
 , temporary
 , text
+, time
 , tinylog
 , unliftio
 , wai
@@ -63,6 +64,7 @@ mkDerivation {
     servant-openapi3
     servant-server
     text
+    time
     tinylog
     unliftio
     wai

libs/extended/extended.cabal

@@ -19,6 +19,7 @@ build-type:    Simple
 library
   -- cabal-fmt: expand src
   exposed-modules:
+    Data.Time.Clock.DiffTime
     Network.AMQP.Extended
     Network.RabbitMqAdmin
     Options.Applicative.Extended
@@ -101,6 +102,7 @@ library
     , servant-openapi3
     , servant-server
     , text
+    , time
     , tinylog
     , unliftio
     , wai

libs/extended/src/Data/Time/Clock/DiffTime.hs

@@ -0,0 +1,43 @@
+module Data.Time.Clock.DiffTime
+  ( DiffTime,
+    weeksToDiffTime,
+    daysToDiffTime,
+    hoursToDiffTime,
+    minutesToDiffTime,
+    secondsToDiffTime,
+    millisecondsToDiffTime,
+    microsecondsToDiffTime,
+    nanosecondsToDiffTime,
+    picosecondsToDiffTime,
+    diffTimeToFullMicroseconds,
+    diffTimeToPicoseconds,
+  )
+where
+
+import Data.Time
+import Imports
+
+weeksToDiffTime,
+  daysToDiffTime,
+  hoursToDiffTime,
+  minutesToDiffTime,
+  millisecondsToDiffTime,
+  microsecondsToDiffTime,
+  nanosecondsToDiffTime ::
+    Integer -> DiffTime
+weeksToDiffTime = daysToDiffTime . (7 *)
+daysToDiffTime = hoursToDiffTime . (24 *)
+hoursToDiffTime = minutesToDiffTime . (60 *)
+minutesToDiffTime = secondsToDiffTime . (60 *)
+millisecondsToDiffTime = picosecondsToDiffTime . (e9 *)
+microsecondsToDiffTime = picosecondsToDiffTime . (e6 *)
+nanosecondsToDiffTime = picosecondsToDiffTime . (e3 *)
+
+-- | Rounds down. Useful for 'threadDelay', 'timeout', etc.
+diffTimeToFullMicroseconds :: DiffTime -> Int
+diffTimeToFullMicroseconds = fromInteger . (`div` e6) . diffTimeToPicoseconds
+
+e3, e6, e9 :: Integer
+e3 = 1_000
+e6 = 1_000_000
+e9 = 1_000_000_000

libs/gundeck-types/src/Gundeck/Types/Push/V2.hs

@@ -171,14 +171,15 @@ instance ToJSON RecipientClients where
 -- ApsData
 
 newtype ApsSound = ApsSound {fromSound :: Text}
-  deriving (Eq, Show, ToJSON, FromJSON)
+  deriving (Eq, Show, ToJSON, FromJSON, Arbitrary)
 
 newtype ApsLocKey = ApsLocKey {fromLocKey :: Text}
-  deriving (Eq, Show, ToJSON, FromJSON)
+  deriving (Eq, Show, ToJSON, FromJSON, Arbitrary)
 
 data ApsPreference
   = ApsStdPreference
-  deriving (Eq, Show)
+  deriving (Eq, Show, Generic)
+  deriving (Arbitrary) via GenericUniform ApsPreference
 
 instance ToJSON ApsPreference where
   toJSON ApsStdPreference = "std"
@@ -195,7 +196,8 @@ data ApsData = ApsData
     _apsPreference :: !(Maybe ApsPreference),
     _apsBadge :: !Bool
   }
-  deriving (Eq, Show)
+  deriving (Eq, Show, Generic)
+  deriving (Arbitrary) via GenericUniform ApsData
 
 makeLenses ''ApsData
 

libs/polysemy-wire-zoo/polysemy-wire-zoo.cabal

@@ -17,6 +17,7 @@ library
     Wire.Sem.Concurrency
     Wire.Sem.Concurrency.IO
     Wire.Sem.Concurrency.Sequential
+    Wire.Sem.Delay
     Wire.Sem.FromUTC
     Wire.Sem.Jwk
     Wire.Sem.Logger

libs/polysemy-wire-zoo/src/Wire/Sem/Concurrency.hs

@@ -80,6 +80,32 @@ unsafePooledMapConcurrentlyN_ n f as =
     (UnsafePooledMapConcurrentlyN_ n f as :: Concurrency 'Unsafe (Sem r) ())
 {-# INLINEABLE unsafePooledMapConcurrentlyN_ #-}
 
+unsafePooledForConcurrentlyN ::
+  forall r t a b.
+  (Member (Concurrency 'Unsafe) r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r b) ->
+  Sem r [b]
+unsafePooledForConcurrentlyN n as f =
+  send
+    (UnsafePooledMapConcurrentlyN n f as :: Concurrency 'Unsafe (Sem r) [b])
+{-# INLINEABLE unsafePooledForConcurrentlyN #-}
+
+unsafePooledForConcurrentlyN_ ::
+  forall r t a b.
+  (Member (Concurrency 'Unsafe) r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r b) ->
+  Sem r ()
+unsafePooledForConcurrentlyN_ n as f =
+  send
+    (UnsafePooledMapConcurrentlyN_ n f as :: Concurrency 'Unsafe (Sem r) ())
+{-# INLINEABLE unsafePooledForConcurrentlyN_ #-}
+
 pooledMapConcurrentlyN ::
   forall r' r t a b.
   r' ~ '[Final IO] =>
@@ -111,3 +137,35 @@ pooledMapConcurrentlyN_ n f as =
         Concurrency 'Safe (Sem r) ()
     )
 {-# INLINEABLE pooledMapConcurrentlyN_ #-}
+
+pooledForConcurrentlyN ::
+  forall r' r t a b.
+  r' ~ '[Final IO] =>
+  (Member (Concurrency 'Safe) r, Subsume r' r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r' b) ->
+  Sem r [b]
+pooledForConcurrentlyN n as f =
+  send
+    ( UnsafePooledMapConcurrentlyN n (subsume_ @r' @r . f) as ::
+        Concurrency 'Safe (Sem r) [b]
+    )
+{-# INLINEABLE pooledForConcurrentlyN #-}
+
+pooledForConcurrentlyN_ ::
+  forall r' r t a b.
+  r' ~ '[Final IO] =>
+  (Member (Concurrency 'Safe) r, Subsume r' r, Foldable t) =>
+  -- | Max. number of threads. Should not be less than 1.
+  Int ->
+  t a ->
+  (a -> Sem r' b) ->
+  Sem r ()
+pooledForConcurrentlyN_ n as f =
+  send
+    ( UnsafePooledMapConcurrentlyN_ n (subsume_ @r' @r . f) as ::
+        Concurrency 'Safe (Sem r) ()
+    )
+{-# INLINEABLE pooledForConcurrentlyN_ #-}

libs/polysemy-wire-zoo/src/Wire/Sem/Delay.hs

@@ -0,0 +1,32 @@
+{-# LANGUAGE TemplateHaskell #-}
+
+module Wire.Sem.Delay where
+
+import Imports
+import Polysemy
+
+data Delay m a where
+  Delay :: Int -> Delay m ()
+
+makeSem ''Delay
+
+runDelay :: Member (Embed IO) r => Sem (Delay ': r) a -> Sem r a
+runDelay = interpret $ \case
+  Delay i -> threadDelay i
+
+runControlledDelay :: forall r a. (Member (Embed IO) r) => MVar Int -> Sem (Delay : r) a -> Sem r a
+runControlledDelay tickSource = interpret $ \case
+  Delay n -> waitForTicks n
+  where
+    waitForTicks :: Int -> Sem r ()
+    waitForTicks 0 = pure ()
+    waitForTicks remaining0 = do
+      passedTicks <- takeMVar tickSource
+      let remaining = remaining0 - passedTicks
+      if remaining <= 0
+        then pure ()
+        else waitForTicks remaining
+
+runDelayInstantly :: Sem (Delay : r) a -> Sem r a
+runDelayInstantly = interpret $ \case
+  Delay _ -> pure ()

libs/polysemy-wire-zoo/src/Wire/Sem/Logger/TinyLog.hs

@@ -1,3 +1,5 @@
+{-# LANGUAGE RecordWildCards #-}
+
 -- This file is part of the Wire Server implementation.
 --
 -- Copyright (C) 2022 Wire Swiss GmbH <opensource@wire.com>
@@ -21,12 +23,16 @@ module Wire.Sem.Logger.TinyLog
     stringLoggerToTinyLog,
     discardTinyLogs,
     module Wire.Sem.Logger.Level,
+    LogRecorder (..),
+    newLogRecorder,
+    recordLogs,
   )
 where
 
 import Data.Id
 import Imports
 import Polysemy
+import Polysemy.TinyLog (TinyLog)
 import qualified System.Logger as Log
 import Wire.Sem.Logger
 import Wire.Sem.Logger.Level
@@ -58,3 +64,12 @@ stringLoggerToTinyLog = mapLogger @String Log.msg
 
 discardTinyLogs :: Sem (Logger (Log.Msg -> Log.Msg) ': r) a -> Sem r a
 discardTinyLogs = discardLogs
+
+newtype LogRecorder = LogRecorder {recordedLogs :: IORef [(Level, LByteString)]}
+
+newLogRecorder :: IO LogRecorder
+newLogRecorder = LogRecorder <$> newIORef []
+
+recordLogs :: Member (Embed IO) r => LogRecorder -> Sem (TinyLog ': r) a -> Sem r a
+recordLogs LogRecorder {..} = interpret $ \(Log lvl msg) ->
+  modifyIORef' recordedLogs (++ [(lvl, Log.render (Log.renderDefault ", ") msg)])

libs/types-common/src/Data/Range.hs

@@ -508,6 +508,9 @@ genRange pack_ gc =
 instance (KnownNat n, KnownNat m, n <= m) => Arbitrary (Range n m Integer) where
   arbitrary = genIntegral
 
+instance (KnownNat n, KnownNat m, n <= m) => Arbitrary (Range n m Int32) where
+  arbitrary = genIntegral
+
 instance (KnownNat n, KnownNat m, n <= m) => Arbitrary (Range n m Word) where
   arbitrary = genIntegral