notifications: protocol, server (#335)

* notifications: protocol

* update protocol to include subscription ID and DH public key

* update protocol, started server

* add notification server subscription DH key

* use the same command type in notifications protocol, protocol parsing, server frame

* remove empty files

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -0,0 +1,161 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeFamilies #-}
+
+module Simplex.Messaging.Notifications.Protocol where
+
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.Maybe (isNothing)
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Encoding
+import Simplex.Messaging.Protocol
+
+data NtfCommandTag
+  = NCCreate_
+  | NCCheck_
+  | NCToken_
+  | NCDelete_
+  deriving (Show)
+
+instance Encoding NtfCommandTag where
+  smpEncode = \case
+    NCCreate_ -> "CREATE"
+    NCCheck_ -> "CHECK"
+    NCToken_ -> "TOKEN"
+    NCDelete_ -> "DELETE"
+  smpP = messageTagP
+
+instance ProtocolMsgTag NtfCommandTag where
+  decodeTag = \case
+    "CREATE" -> Just NCCreate_
+    "CHECK" -> Just NCCheck_
+    "TOKEN" -> Just NCToken_
+    "DELETE" -> Just NCDelete_
+    _ -> Nothing
+
+data NtfCommand
+  = NCCreate DeviceToken SMPQueueNtfUri C.APublicVerifyKey C.PublicKeyX25519
+  | NCCheck
+  | NCToken DeviceToken
+  | NCDelete
+
+instance Protocol NtfCommand where
+  type Tag NtfCommand = NtfCommandTag
+  encodeProtocol = \case
+    NCCreate token smpQueue verifyKey dhKey -> e (NCCreate_, ' ', token, smpQueue, verifyKey, dhKey)
+    NCCheck -> e NCCheck_
+    NCToken token -> e (NCToken_, ' ', token)
+    NCDelete -> e NCDelete_
+    where
+      e :: Encoding a => a -> ByteString
+      e = smpEncode
+
+  protocolP = \case
+    NCCreate_ -> NCCreate <$> _smpP <*> smpP <*> smpP <*> smpP
+    NCCheck_ -> pure NCCheck
+    NCToken_ -> NCToken <$> _smpP
+    NCDelete_ -> pure NCDelete
+
+  checkCredentials (sig, _, subId, _) cmd = case cmd of
+    -- CREATE must have signature but NOT subscription ID
+    NCCreate {}
+      | isNothing sig -> Left $ CMD NO_AUTH
+      | not (B.null subId) -> Left $ CMD HAS_AUTH
+      | otherwise -> Right cmd
+    -- other client commands must have both signature and subscription ID
+    _
+      | isNothing sig || B.null subId -> Left $ CMD NO_AUTH
+      | otherwise -> Right cmd
+
+data NtfResponseTag
+  = NRSubId_
+  | NROk_
+  | NRErr_
+  | NRStat_
+  deriving (Show)
+
+instance Encoding NtfResponseTag where
+  smpEncode = \case
+    NRSubId_ -> "ID"
+    NROk_ -> "OK"
+    NRErr_ -> "ERR"
+    NRStat_ -> "STAT"
+  smpP = messageTagP
+
+instance ProtocolMsgTag NtfResponseTag where
+  decodeTag = \case
+    "ID" -> Just NRSubId_
+    "OK" -> Just NROk_
+    "ERR" -> Just NRErr_
+    "STAT" -> Just NRStat_
+    _ -> Nothing
+
+data NtfResponse
+  = NRSubId C.PublicKeyX25519
+  | NROk
+  | NRErr ErrorType
+  | NRStat NtfStatus
+
+instance Protocol NtfResponse where
+  type Tag NtfResponse = NtfResponseTag
+  encodeProtocol = \case
+    NRSubId dhKey -> e (NRSubId_, ' ', dhKey)
+    NROk -> e NROk_
+    NRErr err -> e (NRErr_, ' ', err)
+    NRStat stat -> e (NRStat_, ' ', stat)
+    where
+      e :: Encoding a => a -> ByteString
+      e = smpEncode
+
+  protocolP = \case
+    NRSubId_ -> NRSubId <$> _smpP
+    NROk_ -> pure NROk
+    NRErr_ -> NRErr <$> _smpP
+    NRStat_ -> NRStat <$> _smpP
+
+  checkCredentials (_, _, subId, _) cmd = case cmd of
+    -- ERR response does not always have subscription ID
+    NRErr _ -> Right cmd
+    -- other server responses must have subscription ID
+    _
+      | B.null subId -> Left $ CMD NO_ENTITY
+      | otherwise -> Right cmd
+
+data SMPQueueNtfUri = SMPQueueNtfUri
+  { smpServer :: SMPServer,
+    notifierId :: NotifierId,
+    notifierKey :: NtfPrivateSignKey
+  }
+
+instance Encoding SMPQueueNtfUri where
+  smpEncode SMPQueueNtfUri {smpServer, notifierId, notifierKey} = smpEncode (smpServer, notifierId, notifierKey)
+  smpP = do
+    (smpServer, notifierId, notifierKey) <- smpP
+    pure $ SMPQueueNtfUri smpServer notifierId notifierKey
+
+newtype DeviceToken = DeviceToken ByteString
+
+instance Encoding DeviceToken where
+  smpEncode (DeviceToken t) = smpEncode t
+  smpP = DeviceToken <$> smpP
+
+type NtfSubsciptionId = ByteString
+
+data NtfStatus = NSPending | NSActive | NSEnd | NSSMPAuth
+
+instance Encoding NtfStatus where
+  smpEncode = \case
+    NSPending -> "PENDING"
+    NSActive -> "ACTIVE"
+    NSEnd -> "END"
+    NSSMPAuth -> "SMP_AUTH"
+  smpP =
+    A.takeTill (== ' ') >>= \case
+      "PENDING" -> pure NSPending
+      "ACTIVE" -> pure NSActive
+      "END" -> pure NSEnd
+      "SMP_AUTH" -> pure NSSMPAuth
+      _ -> fail "bad NtfError"

src/Simplex/Messaging/Notifications/Server.hs

@@ -0,0 +1,111 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Notifications.Server where
+
+import Control.Monad.Except
+import Control.Monad.IO.Unlift (MonadUnliftIO)
+import Control.Monad.Reader
+import Crypto.Random (MonadRandom)
+import Data.ByteString.Char8 (ByteString)
+import Network.Socket (ServiceName)
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Notifications.Server.Env
+import Simplex.Messaging.Notifications.Server.Subscriptions
+import Simplex.Messaging.Notifications.Transport
+import Simplex.Messaging.Protocol (ErrorType (..), Transmission, encodeTransmission, tGet, tPut)
+import Simplex.Messaging.Server
+import Simplex.Messaging.Transport (ATransport (..), THandle (..), TProxy, Transport)
+import Simplex.Messaging.Transport.Server (runTransportServer)
+import Simplex.Messaging.Util
+import UnliftIO.Exception
+import UnliftIO.STM
+
+runNtfServer :: (MonadRandom m, MonadUnliftIO m) => NtfServerConfig -> m ()
+runNtfServer cfg = do
+  started <- newEmptyTMVarIO
+  runNtfServerBlocking started cfg
+
+runNtfServerBlocking :: (MonadRandom m, MonadUnliftIO m) => TMVar Bool -> NtfServerConfig -> m ()
+runNtfServerBlocking started cfg@NtfServerConfig {transports} = do
+  env <- newNtfServerEnv cfg
+  runReaderT ntfServer env
+  where
+    ntfServer :: (MonadUnliftIO m', MonadReader NtfEnv m') => m' ()
+    ntfServer = raceAny_ (map runServer transports)
+
+    runServer :: (MonadUnliftIO m', MonadReader NtfEnv m') => (ServiceName, ATransport) -> m' ()
+    runServer (tcpPort, ATransport t) = do
+      serverParams <- asks tlsServerParams
+      runTransportServer started tcpPort serverParams (runClient t)
+
+    runClient :: (Transport c, MonadUnliftIO m, MonadReader NtfEnv m) => TProxy c -> c -> m ()
+    runClient _ h = do
+      kh <- asks serverIdentity
+      liftIO (runExceptT $ ntfServerHandshake h kh) >>= \case
+        Right th -> runNtfClientTransport th
+        Left _ -> pure ()
+
+runNtfClientTransport :: (Transport c, MonadUnliftIO m, MonadReader NtfEnv m) => THandle c -> m ()
+runNtfClientTransport th@THandle {sessionId} = do
+  q <- asks $ tbqSize . config
+  c <- atomically $ newNtfServerClient q sessionId
+  raceAny_ [send th c, client c, receive th c]
+    `finally` clientDisconnected c
+
+clientDisconnected :: MonadUnliftIO m => NtfServerClient -> m ()
+clientDisconnected NtfServerClient {connected} = atomically $ writeTVar connected False
+
+receive :: (Transport c, MonadUnliftIO m, MonadReader NtfEnv m) => THandle c -> NtfServerClient -> m ()
+receive th NtfServerClient {rcvQ, sndQ} = forever $ do
+  (sig, signed, (corrId, queueId, cmdOrError)) <- tGet th
+  case cmdOrError of
+    Left e -> write sndQ (corrId, queueId, NRErr e)
+    Right cmd -> do
+      verified <- verifyTransmission sig signed queueId cmd
+      if verified
+        then write rcvQ (corrId, queueId, cmd)
+        else write sndQ (corrId, queueId, NRErr AUTH)
+  where
+    write q t = atomically $ writeTBQueue q t
+
+send :: (Transport c, MonadUnliftIO m) => THandle c -> NtfServerClient -> m ()
+send h NtfServerClient {sndQ, sessionId} = forever $ do
+  t <- atomically $ readTBQueue sndQ
+  liftIO $ tPut h (Nothing, encodeTransmission sessionId t)
+
+verifyTransmission ::
+  forall m. (MonadUnliftIO m, MonadReader NtfEnv m) => Maybe C.ASignature -> ByteString -> NtfSubsciptionId -> NtfCommand -> m Bool
+verifyTransmission sig_ signed subId cmd = do
+  case cmd of
+    NCCreate _ _ k _ -> pure $ verifyCmdSignature sig_ signed k
+    _ -> do
+      st <- asks store
+      verifySubCmd <$> atomically (getNtfSubscription st subId)
+  where
+    verifySubCmd = \case
+      Right sub -> verifyCmdSignature sig_ signed $ subVerifyKey sub
+      Left _ -> maybe False (dummyVerifyCmd signed) sig_ `seq` False
+
+client :: forall m. (MonadUnliftIO m, MonadReader NtfEnv m) => NtfServerClient -> m ()
+client NtfServerClient {rcvQ, sndQ} =
+  forever $
+    atomically (readTBQueue rcvQ)
+      >>= processCommand
+      >>= atomically . writeTBQueue sndQ
+  where
+    processCommand :: Transmission NtfCommand -> m (Transmission NtfResponse)
+    processCommand (corrId, subId, cmd) = case cmd of
+      NCCreate _token _smpQueue _verifyKey _dhKey -> do
+        pure (corrId, subId, NROk)
+      NCCheck -> do
+        pure (corrId, subId, NROk)
+      NCToken _token -> do
+        pure (corrId, subId, NROk)
+      NCDelete -> do
+        pure (corrId, subId, NROk)

src/Simplex/Messaging/Notifications/Server/Env.hs

@@ -0,0 +1,66 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE NamedFieldPuns #-}
+
+module Simplex.Messaging.Notifications.Server.Env where
+
+import Control.Monad.IO.Unlift
+import Crypto.Random
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.Map.Strict as M
+import Data.X509.Validation (Fingerprint (..))
+import Network.Socket
+import qualified Network.TLS as T
+import Numeric.Natural
+import Simplex.Messaging.Agent.RetryInterval
+import Simplex.Messaging.Client
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Notifications.Server.Subscriptions
+import Simplex.Messaging.Protocol (Transmission)
+import Simplex.Messaging.Transport (ATransport)
+import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams)
+import UnliftIO.STM
+
+data NtfServerConfig = NtfServerConfig
+  { transports :: [(ServiceName, ATransport)],
+    subscriptionIdBytes :: Int,
+    tbqSize :: Natural,
+    smpCfg :: SMPClientConfig,
+    reconnectInterval :: RetryInterval,
+    -- CA certificate private key is not needed for initialization
+    caCertificateFile :: FilePath,
+    privateKeyFile :: FilePath,
+    certificateFile :: FilePath
+  }
+
+data NtfEnv = NtfEnv
+  { config :: NtfServerConfig,
+    serverIdentity :: C.KeyHash,
+    store :: NtfSubscriptions,
+    idsDrg :: TVar ChaChaDRG,
+    tlsServerParams :: T.ServerParams,
+    serverIdentity :: C.KeyHash
+  }
+
+newNtfServerEnv :: (MonadUnliftIO m, MonadRandom m) => NtfServerConfig -> m NtfEnv
+newNtfServerEnv config@NtfServerConfig {caCertificateFile, certificateFile, privateKeyFile} = do
+  idsDrg <- newTVarIO =<< drgNew
+  store <- newTVarIO M.empty
+  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+  Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
+  let serverIdentity = C.KeyHash fp
+  pure NtfEnv {config, store, idsDrg, tlsServerParams, serverIdentity}
+
+data NtfServerClient = NtfServerClient
+  { rcvQ :: TBQueue (Transmission NtfCommand),
+    sndQ :: TBQueue (Transmission NtfResponse),
+    sessionId :: ByteString,
+    connected :: TVar Bool
+  }
+
+newNtfServerClient :: Natural -> ByteString -> STM NtfServerClient
+newNtfServerClient qSize sessionId = do
+  rcvQ <- newTBQueue qSize
+  sndQ <- newTBQueue qSize
+  connected <- newTVar True
+  return NtfServerClient {rcvQ, sndQ, sessionId, connected}

src/Simplex/Messaging/Notifications/Server/Subscriptions.hs

@@ -0,0 +1,25 @@
+module Simplex.Messaging.Notifications.Server.Subscriptions where
+
+import Control.Concurrent.STM
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Protocol (ErrorType (..), NotifierId, NtfPrivateSignKey, SMPServer)
+
+type NtfSubscriptionsData = Map NtfSubsciptionId NtfSubsciptionRec
+
+type NtfSubscriptions = TVar NtfSubscriptionsData
+
+data NtfSubsciptionRec = NtfSubsciptionRec
+  { smpServer :: SMPServer,
+    notifierId :: NotifierId,
+    notifierKey :: NtfPrivateSignKey,
+    token :: DeviceToken,
+    status :: TVar NtfStatus,
+    subVerifyKey :: C.APublicVerifyKey,
+    subDHSecret :: C.DhSecretX25519
+  }
+
+getNtfSubscription :: NtfSubscriptions -> NtfSubsciptionId -> STM (Either ErrorType NtfSubsciptionRec)
+getNtfSubscription st subId = maybe (Left AUTH) Right . M.lookup subId <$> readTVar st

src/Simplex/Messaging/Notifications/Transport.hs

@@ -0,0 +1,19 @@
+module Simplex.Messaging.Notifications.Transport where
+
+import Control.Monad.Except
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Transport
+
+ntfBlockSize :: Int
+ntfBlockSize = 512
+
+-- | Notifcations server transport handshake.
+ntfServerHandshake :: Transport c => c -> C.KeyHash -> ExceptT TransportError IO (THandle c)
+ntfServerHandshake c _ = pure $ ntfTHandle c
+
+-- | Notifcations server client transport handshake.
+ntfClientHandshake :: Transport c => c -> C.KeyHash -> ExceptT TransportError IO (THandle c)
+ntfClientHandshake c _ = pure $ ntfTHandle c
+
+ntfTHandle :: Transport c => c -> THandle c
+ntfTHandle c = THandle {connection = c, sessionId = tlsUnique c, blockSize = ntfBlockSize, thVersion = 0}