Merge branch 'master' into postgres

2026-10-12 02:22:45 +00:00 · 2024-12-25 12:15:41 +04:00
parent e557dc86e7 bf28902327
commit b7a8df40c4
12 changed files with 113 additions and 43 deletions
@@ -1,3 +1,11 @@
+# 6.2.2
+
+SMP server:
+- add optional Prometheus metrics (#1411).
+
+Build:
+- remove three modules from client library.
+
 # 6.2.0

 Version 6.2.0.7
@@ -1,7 +1,7 @@
 cabal-version: 1.12

 name:           simplexmq
-version:        6.2.1.0
+version:        6.2.2.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -196,6 +196,7 @@ library
          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240702_servers_stats
          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240930_ntf_tokens_to_delete
          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241224_ratchet_e2e_snd_params
  if !flag(client_library)
      exposed-modules:
          Simplex.FileTransfer.Client.Main
@@ -836,26 +836,39 @@ joinConn c userId connId enableNtfs cReq cInfo pqSupport subMode = do
 startJoinInvitation :: AgentClient -> UserId -> ConnId -> Maybe SndQueue -> Bool -> ConnectionRequestUri 'CMInvitation -> PQSupport -> AM (ConnData, SndQueue, CR.SndE2ERatchetParams 'C.X448)
 startJoinInvitation c userId connId sq_ enableNtfs cReqUri pqSup =
  lift (compatibleInvitationUri cReqUri) >>= \case
-    Just (qInfo, Compatible e2eRcvParams@(CR.E2ERatchetParams v _ rcDHRr kem_), Compatible connAgentVersion) -> do
-      g <- asks random
+    Just (qInfo, Compatible e2eRcvParams@(CR.E2ERatchetParams v _ _ _), Compatible connAgentVersion) -> do
+      -- this case avoids re-generating queue keys and subsequent failure of SKEY that timed out
+      -- e2ePubKey is always present, it's Maybe historically
      let pqSupport = pqSup `CR.pqSupportAnd` versionPQSupport_ connAgentVersion (Just v)
+      (sq', e2eSndParams) <- case sq_ of
+        Just sq@SndQueue {e2ePubKey = Just _k} -> do
+          e2eSndParams <-
+            withStore' c (\db -> getSndRatchet db connId v) >>= \case
+              Right r -> pure $ snd r
+              Left e -> do
+                atomically $ writeTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no snd ratchet " <> show e))
+                createRatchet_ pqSupport e2eRcvParams
+          pure (sq, e2eSndParams)
+        _ -> do
+          q <- lift $ fst <$> newSndQueue userId "" qInfo
+          e2eSndParams <- createRatchet_ pqSupport e2eRcvParams
+          withStore c $ \db -> runExceptT $ do
+            sq' <- maybe (ExceptT $ updateNewConnSnd db connId q) pure sq_
+            pure (sq', e2eSndParams)
+      let cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
+      pure (cData, sq', e2eSndParams)
+    Nothing -> throwE $ AGENT A_VERSION
+  where
+    createRatchet_ pqSupport e2eRcvParams@(CR.E2ERatchetParams v _ rcDHRr kem_) = do
+      g <- asks random
      (pk1, pk2, pKem, e2eSndParams) <- liftIO $ CR.generateSndE2EParams g v (CR.replyKEM_ v kem_ pqSupport)
      (_, rcDHRs) <- atomically $ C.generateKeyPair g
      rcParams <- liftEitherWith cryptoError $ CR.pqX3dhSnd pk1 pk2 pKem e2eRcvParams
      maxSupported <- asks $ maxVersion . e2eEncryptVRange . config
      let rcVs = CR.RatchetVersions {current = v, maxSupported}
          rc = CR.initSndRatchet rcVs rcDHRr rcDHRs rcParams
-      -- this case avoids re-generating queue keys and subsequent failure of SKEY that timed out
-      -- e2ePubKey is always present, it's Maybe historically
-      q <- case sq_ of
-        Just sq@SndQueue {e2ePubKey = Just _k} -> pure (sq :: SndQueue) {dbQueueId = DBNewQueue}
-        _ -> lift $ fst <$> newSndQueue userId "" qInfo
-      let cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
-      sq' <- withStore c $ \db -> runExceptT $ do
-        liftIO $ createRatchet db connId rc
-        maybe (ExceptT $ updateNewConnSnd db connId q) pure sq_
-      pure (cData, sq', e2eSndParams)
-    Nothing -> throwE $ AGENT A_VERSION
+      withStore' c $ \db -> createSndRatchet db connId rc e2eSndParams
+      pure e2eSndParams

 connRequestPQSupport :: AgentClient -> PQSupport -> ConnectionRequestUri c -> IO (Maybe (VersionSMPA, PQSupport))
 connRequestPQSupport c pqSup cReq = withAgentEnv' c $ case cReq of
@@ -894,6 +907,7 @@ joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSup subMod
    case conn of
      NewConnection _ -> doJoin Nothing
      SndConnection _ sq -> doJoin $ Just sq
+      DuplexConnection _ (RcvQueue {status = New} :| _) (sq@SndQueue {status = New} :| _) -> doJoin $ Just sq
      _ -> throwE $ CMD PROHIBITED $ "joinConnSrv: bad connection " <> show cType
  where
    doJoin :: Maybe SndQueue -> AM SndQueueSecured
@@ -117,6 +117,8 @@ module Simplex.Messaging.Agent.Store.AgentStore
    createRatchetX3dhKeys,
    getRatchetX3dhKeys,
    setRatchetX3dhKeys,
+    createSndRatchet,
+    getSndRatchet,
    createRatchet,
    deleteRatchet,
    getRatchet,
@@ -230,7 +232,6 @@ where

 import Control.Logger.Simple
 import Control.Monad
-import Control.Monad.Except
 import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
@@ -387,7 +388,6 @@ updateNewConnSnd :: DB.Connection -> ConnId -> NewSndQueue -> IO (Either StoreEr
 updateNewConnSnd db connId sq =
  getConn db connId $>>= \case
    (SomeConn _ NewConnection {}) -> updateConn
-    (SomeConn _ SndConnection {}) -> updateConn -- to allow retries
    (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
  where
    updateConn :: IO (Either StoreError SndQueue)
@@ -1046,18 +1046,47 @@ getRatchetX3dhKeys db connId =
      _ -> Left SEX3dhKeysNotFound

 -- used to remember new keys when starting ratchet re-synchronization
-- TODO remove the columns for public keys in v5.7.
-- Currently, the keys are not used but still stored to support app downgrade to the previous version.
 setRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> Maybe CR.RcvPrivRKEMParams -> IO ()
 setRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 pqPrivKem =
  DB.execute
    db
    [sql|
      UPDATE ratchets
-      SET x3dh_priv_key_1 = ?, x3dh_priv_key_2 = ?, x3dh_pub_key_1 = ?, x3dh_pub_key_2 = ?, pq_priv_kem = ?
+      SET x3dh_priv_key_1 = ?, x3dh_priv_key_2 = ?, pq_priv_kem = ?
      WHERE conn_id = ?
    |]
-    (x3dhPrivKey1, x3dhPrivKey2, C.publicKey x3dhPrivKey1, C.publicKey x3dhPrivKey2, pqPrivKem, connId)
+    (x3dhPrivKey1, x3dhPrivKey2, pqPrivKem, connId)
+
+createSndRatchet :: DB.Connection -> ConnId -> RatchetX448 -> CR.AE2ERatchetParams 'C.X448 -> IO ()
+createSndRatchet db connId ratchetState (CR.AE2ERatchetParams s (CR.E2ERatchetParams _ x3dhPubKey1 x3dhPubKey2 pqPubKem)) =
+  DB.execute
+    db
+    [sql|
+      INSERT INTO ratchets
+        (conn_id, ratchet_state, x3dh_pub_key_1, x3dh_pub_key_2, pq_pub_kem) VALUES (?, ?, ?, ?, ?)
+      ON CONFLICT (conn_id) DO UPDATE SET
+        ratchet_state = EXCLUDED.ratchet_state,
+        x3dh_priv_key_1 = NULL,
+        x3dh_priv_key_2 = NULL,
+        x3dh_pub_key_1 = EXCLUDED.x3dh_pub_key_1,
+        x3dh_pub_key_2 = EXCLUDED.x3dh_pub_key_2,
+        pq_priv_kem = NULL,
+        pq_pub_kem = EXCLUDED.pq_pub_kem
+    |]
+    (connId, ratchetState, x3dhPubKey1, x3dhPubKey2, CR.ARKP s <$> pqPubKem)
+
+getSndRatchet :: DB.Connection -> ConnId -> CR.VersionE2E -> IO (Either StoreError (RatchetX448, CR.AE2ERatchetParams 'C.X448))
+getSndRatchet db connId v =
+  firstRow' result SEX3dhKeysNotFound $
+    DB.query db "SELECT ratchet_state, x3dh_pub_key_1, x3dh_pub_key_2, pq_pub_kem FROM ratchets WHERE conn_id = ?" (Only connId)
+  where
+    result = \case
+      (Just ratchetState, Just k1, Just k2, pKem_) -> 
+        let params = case pKem_ of
+              Nothing -> CR.AE2ERatchetParams CR.SRKSProposed (CR.E2ERatchetParams v k1 k2 Nothing)
+              Just (CR.ARKP s pKem) -> CR.AE2ERatchetParams s (CR.E2ERatchetParams v k1 k2 (Just pKem))
+         in Right (ratchetState, params)
+      _ -> Left SEX3dhKeysNotFound

 -- TODO remove the columns for public keys in v5.7.
 createRatchet :: DB.Connection -> ConnId -> RatchetX448 -> IO ()
@@ -1073,7 +1102,8 @@ createRatchet db connId rc =
        x3dh_priv_key_2 = NULL,
        x3dh_pub_key_1 = NULL,
        x3dh_pub_key_2 = NULL,
-        pq_priv_kem = NULL
+        pq_priv_kem = NULL,
+        pq_pub_kem = NULL
    |]
    (connId, rc, rc)

@@ -174,7 +174,8 @@ CREATE TABLE ratchets(
  e2e_version INTEGER NOT NULL DEFAULT 1,
  x3dh_pub_key_1 BYTEA,
  x3dh_pub_key_2 BYTEA,
-  pq_priv_kem BYTEA
+  pq_priv_kem BYTEA,
+  pq_pub_kem BYTEA
 );
 CREATE TABLE skipped_messages(
  skipped_message_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
@@ -12,22 +12,12 @@ module Simplex.Messaging.Agent.Store.Postgres.Util
  )
 where

-import Control.Exception (bracket, throwIO)
+import Control.Exception (bracket)
 import Control.Monad (forM_, unless, void, when)
-import Data.Functor (($>))
 import Data.String (fromString)
-import Data.Text (Text)
 import Database.PostgreSQL.Simple (ConnectInfo (..), Only (..), defaultConnectInfo)
 import qualified Database.PostgreSQL.Simple as PSQL
 import Database.PostgreSQL.Simple.SqlQQ (sql)
-import Simplex.Messaging.Agent.Store.Migrations (migrateSchema)
-import Simplex.Messaging.Agent.Store.Postgres.Common
-import qualified Simplex.Messaging.Agent.Store.Postgres.DB as DB
-import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
-import Simplex.Messaging.Util (ifM)
-import UnliftIO.Exception (onException)
-import UnliftIO.MVar
-import UnliftIO.STM

 createDBAndUserIfNotExists :: ConnectInfo -> IO ()
 createDBAndUserIfNotExists ConnectInfo {connectUser = user, connectDatabase = dbName} = do
@@ -65,6 +65,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240624_snd_secure
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240702_servers_stats
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240930_ntf_tokens_to_delete
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241224_ratchet_e2e_snd_params
 import Simplex.Messaging.Agent.Store.Shared
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -106,7 +107,8 @@ schemaMigrations =
    ("m20240624_snd_secure", m20240624_snd_secure, Just down_m20240624_snd_secure),
    ("m20240702_servers_stats", m20240702_servers_stats, Just down_m20240702_servers_stats),
    ("m20240930_ntf_tokens_to_delete", m20240930_ntf_tokens_to_delete, Just down_m20240930_ntf_tokens_to_delete),
-    ("m20241007_rcv_queues_last_broker_ts", m20241007_rcv_queues_last_broker_ts, Just down_m20241007_rcv_queues_last_broker_ts)
+    ("m20241007_rcv_queues_last_broker_ts", m20241007_rcv_queues_last_broker_ts, Just down_m20241007_rcv_queues_last_broker_ts),
+    ("m20241224_ratchet_e2e_snd_params", m20241224_ratchet_e2e_snd_params, Just down_m20241224_ratchet_e2e_snd_params)
  ]

 -- | The list of migrations in ascending order by date
@@ -0,0 +1,18 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241224_ratchet_e2e_snd_params where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20241224_ratchet_e2e_snd_params :: Query
+m20241224_ratchet_e2e_snd_params =
+    [sql|
+ALTER TABLE ratchets ADD COLUMN pq_pub_kem BLOB;
+|]
+
+down_m20241224_ratchet_e2e_snd_params :: Query
+down_m20241224_ratchet_e2e_snd_params =
+    [sql|
+ALTER TABLE ratchets DROP COLUMN pq_pub_kem;
+|]
@@ -166,7 +166,8 @@ CREATE TABLE ratchets(
  ,
  x3dh_pub_key_1 BLOB,
  x3dh_pub_key_2 BLOB,
-  pq_priv_kem BLOB
+  pq_priv_kem BLOB,
+  pq_pub_kem BLOB
 ) WITHOUT ROWID;
 CREATE TABLE skipped_messages(
  skipped_message_id INTEGER PRIMARY KEY,
@@ -213,6 +213,10 @@ instance Encoding ARKEMParams where
      'A' -> ARKP SRKSAccepted .: RKParamsAccepted <$> smpP <*> smpP
      _ -> fail "bad ratchet KEM params"

+instance ToField ARKEMParams where toField = toField . Binary . smpEncode
+
+instance FromField ARKEMParams where fromField = blobFieldDecoder smpDecode
+
 data E2ERatchetParams (s :: RatchetKEMState) (a :: Algorithm)
  = E2ERatchetParams VersionE2E (PublicKey a) (PublicKey a) (Maybe (RKEMParams s))
  deriving (Show)
@@ -13,6 +13,7 @@ import Data.Time.Clock.System (systemEpochDay)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Network.Socket (ServiceName)
 import Simplex.Messaging.Server.Stats
+import Simplex.Messaging.Transport.Server (SocketStats (..))

 data ServerMetrics = ServerMetrics
  { statsData :: ServerStatsData,
@@ -32,13 +33,6 @@ data RealTimeMetrics = RealTimeMetrics
    ntfSubClientsCount :: Int
  }

-data SocketStats = SocketStats
-  { socketsAccepted :: Int,
-    socketsClosed :: Int,
-    socketsActive :: Int,
-    socketsLeaked :: Int
-  }
-
 {-# FOURMOLU_DISABLE\n#-}
 prometheusMetrics :: ServerMetrics -> RealTimeMetrics -> UTCTime -> Text
 prometheusMetrics sm rtm ts =
@@ -12,6 +12,7 @@ module Simplex.Messaging.Transport.Server
    runTransportServerState,
    runTransportServerState_,
    SocketState,
+    SocketStats (..),
    newSocketState,
    getSocketStats,
    runTransportServer,
@@ -44,7 +45,6 @@ import Foreign.C.Error
 import GHC.IO.Exception (ioe_errno)
 import Network.Socket
 import qualified Network.TLS as T
-import Simplex.Messaging.Server.Prometheus
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
 import System.Exit (exitFailure)
@@ -165,6 +165,13 @@ safeAccept sock =

 type SocketState = (TVar Int, TVar Int, TVar (IntMap (Weak ThreadId)))

+data SocketStats = SocketStats
+  { socketsAccepted :: Int,
+    socketsClosed :: Int,
+    socketsActive :: Int,
+    socketsLeaked :: Int
+  }
+
 newSocketState :: IO SocketState
 newSocketState = (,,) <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO mempty