diff --git a/protocol/agent-protocol.md b/protocol/agent-protocol.md index 00b099e40..f30b33b92 100644 --- a/protocol/agent-protocol.md +++ b/protocol/agent-protocol.md @@ -208,7 +208,7 @@ This syntax of decrypted SMP client message body is defined by `decryptedAgentMe Decrypted SMP message client body can be one of 4 types: - `agentConnInfo` - used by the initiating party when confirming reply queue - sent in `agentConfirmation` envelope. - `agentConnInfoReply` - used by accepting party, includes reply queue(s) in the initial confirmation - sent in `agentConfirmation` envelope. -- `agentRatchetInfo` - used to pass additional information when renegotiating double ratchet encryption - sent in `agentRatchetKey` envelope. +- `agentRatchetInfo` - used to pass additional information when renegotiating double ratchet encryption - sent in `agentRatchetKey` envelope. A key sent in reply to another key includes the hash of that key; agents do not reply to such keys. - `agentMessage` - all other agent messages. `agentMessage` contains these parts: @@ -233,7 +233,8 @@ connInfo = *OCTET agentConnInfoReply = %s"D" smpQueues connInfo smpQueues = length 1*newQueueInfo ; NonEmpty list of reply queues agentRatchetInfo = %s"R" ratchetInfo -ratchetInfo = *OCTET +ratchetInfo = [answeredKeyHash *OCTET] ; bytes after answeredKeyHash are ignored +answeredKeyHash = %s"0" / (%s"1" shortString) ; "0" in a key that starts renegotiation, otherwise SHA-256 of the two raw public keys of the answered key agentMessage = %s"M" agentMsgHeader aMessage agentMsgHeader = agentMsgId prevMsgHash diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index d7cd4d453..067031899 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -2435,62 +2435,66 @@ enqueueMessageB c reqs = do cfg <- asks config (_, reqMids) <- unsafeWithStore c $ \db -> do mapAccumLM (\ids r -> storeSentMsg db cfg ids r `E.catchAny` \e -> (ids,) <$> handleInternal e) IM.empty reqs - forME reqMids $ \((csqs_, _, _, _), InternalId msgId, pqSecr) -> forM csqs_ $ \(cData, sq :| sqs) -> do - submitPendingMsg c sq - let sqs' = filter (isActiveSndQ cData) sqs - pure ((msgId, pqSecr), if null sqs' then Nothing else Just (sqs', msgId)) + forME reqMids $ submitSentMsg c where - storeSentMsg :: - DB.Connection -> - AgentConfig -> - IntMap (Maybe Int64, AMessage) -> - Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage) -> - IO (IntMap (Maybe Int64, AMessage), Either AgentErrorType ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption)) - storeSentMsg db cfg aMessageIds = \case - Left e -> pure (aMessageIds, Left e) - Right req@(csqs_, pqEnc_, msgFlags, mbr) -> case mbr of - VRValue i_ aMessage -> case i_ >>= (`IM.lookup` aMessageIds) of - Just _ -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg duplicate saved message body") - Nothing -> do - (mbId_, r) <- case csqs_ of - Left e -> pure (Nothing, Left e) - Right (cData, sq :| _) -> do - mbId <- createSndMsgBody db aMessage - (Just mbId,) <$> storeSentMsg_ cData sq mbId aMessage - let aMessageIds' = maybe id (`IM.insert` (mbId_, aMessage)) i_ aMessageIds - pure (aMessageIds', r) - VRRef i -> case csqs_ of - Left e -> pure $ (aMessageIds, Left e) - Right (cData, sq :| _) -> case IM.lookup i aMessageIds of - Just (Just mbId, aMessage) -> (aMessageIds,) <$> storeSentMsg_ cData sq mbId aMessage - Just (Nothing, aMessage) -> do - mbId <- createSndMsgBody db aMessage - let aMessageIds' = IM.insert i (Just mbId, aMessage) aMessageIds - (aMessageIds',) <$> storeSentMsg_ cData sq mbId aMessage - Nothing -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg missing saved message body id") - where - storeSentMsg_ cData@ConnData {connId} sq sndMsgBodyId aMessage = fmap (first storeError) $ runExceptT $ do - let AgentConfig {e2eEncryptVRange} = cfg - internalTs <- liftIO getCurrentTime - (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId - -- We need to do pre-flight encoding that is not stored in database - -- to calculate its hash and remember it on connection (createSndMsg -> updateSndMsgHash) - -- to enable next enqueue. - -- (As encoding is different per connection, we can't store shared body, so it's repeated on delivery) - let agentMsgStr = encodeAgentMsgStr aMessage internalSndId prevMsgHash - internalHash = C.sha256Hash agentMsgStr - currentE2EVersion = maxVersion e2eEncryptVRange - (mek, paddedLen, pqEnc) <- agentRatchetEncryptHeader db cData e2eEncAgentMsgLength pqEnc_ currentE2EVersion - withExceptT (SEAgentError . cryptoError) $ CR.rcCheckCanPad paddedLen agentMsgStr - let msgType = aMessageType aMessage - -- msgBody is empty, because snd_messages record is linked to snd_message_bodies - msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody = "", pqEncryption = pqEnc, internalHash, prevMsgHash, sndMsgPrepData_ = Just SndMsgPrepData {encryptKey = mek, paddedLen, sndMsgBodyId}} - liftIO $ createSndMsg db connId msgData - liftIO $ createSndMsgDelivery db sq internalId - pure (req, internalId, pqEnc) handleInternal :: E.SomeException -> IO (Either AgentErrorType b) handleInternal = pure . Left . INTERNAL . show +submitSentMsg :: AgentClient -> ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption) -> AM' (Either AgentErrorType ((AgentMsgId, PQEncryption), Maybe ([SndQueue], AgentMsgId))) +submitSentMsg c ((csqs_, _, _, _), InternalId msgId, pqSecr) = forM csqs_ $ \(cData, sq :| sqs) -> do + submitPendingMsg c sq + let sqs' = filter (isActiveSndQ cData) sqs + pure ((msgId, pqSecr), if null sqs' then Nothing else Just (sqs', msgId)) + +storeSentMsg :: + DB.Connection -> + AgentConfig -> + IntMap (Maybe Int64, AMessage) -> + Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage) -> + IO (IntMap (Maybe Int64, AMessage), Either AgentErrorType ((Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage), InternalId, PQEncryption)) +storeSentMsg db cfg aMessageIds = \case + Left e -> pure (aMessageIds, Left e) + Right req@(csqs_, pqEnc_, msgFlags, mbr) -> case mbr of + VRValue i_ aMessage -> case i_ >>= (`IM.lookup` aMessageIds) of + Just _ -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg duplicate saved message body") + Nothing -> do + (mbId_, r) <- case csqs_ of + Left e -> pure (Nothing, Left e) + Right (cData, sq :| _) -> do + mbId <- createSndMsgBody db aMessage + (Just mbId,) <$> storeSentMsg_ cData sq mbId aMessage + let aMessageIds' = maybe id (`IM.insert` (mbId_, aMessage)) i_ aMessageIds + pure (aMessageIds', r) + VRRef i -> case csqs_ of + Left e -> pure $ (aMessageIds, Left e) + Right (cData, sq :| _) -> case IM.lookup i aMessageIds of + Just (Just mbId, aMessage) -> (aMessageIds,) <$> storeSentMsg_ cData sq mbId aMessage + Just (Nothing, aMessage) -> do + mbId <- createSndMsgBody db aMessage + let aMessageIds' = IM.insert i (Just mbId, aMessage) aMessageIds + (aMessageIds',) <$> storeSentMsg_ cData sq mbId aMessage + Nothing -> pure (aMessageIds, Left $ INTERNAL "enqueueMessageB: storeSentMsg missing saved message body id") + where + storeSentMsg_ cData@ConnData {connId} sq sndMsgBodyId aMessage = fmap (first storeError) $ runExceptT $ do + let AgentConfig {e2eEncryptVRange} = cfg + internalTs <- liftIO getCurrentTime + (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId + -- We need to do pre-flight encoding that is not stored in database + -- to calculate its hash and remember it on connection (createSndMsg -> updateSndMsgHash) + -- to enable next enqueue. + -- (As encoding is different per connection, we can't store shared body, so it's repeated on delivery) + let agentMsgStr = encodeAgentMsgStr aMessage internalSndId prevMsgHash + internalHash = C.sha256Hash agentMsgStr + currentE2EVersion = maxVersion e2eEncryptVRange + (mek, paddedLen, pqEnc) <- agentRatchetEncryptHeader db cData e2eEncAgentMsgLength pqEnc_ currentE2EVersion + withExceptT (SEAgentError . cryptoError) $ CR.rcCheckCanPad paddedLen agentMsgStr + let msgType = aMessageType aMessage + -- msgBody is empty, because snd_messages record is linked to snd_message_bodies + msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody = "", pqEncryption = pqEnc, internalHash, prevMsgHash, sndMsgPrepData_ = Just SndMsgPrepData {encryptKey = mek, paddedLen, sndMsgBodyId}} + liftIO $ createSndMsg db connId msgData + liftIO $ createSndMsgDelivery db sq internalId + pure (req, internalId, pqEnc) + encodeAgentMsgStr :: AMessage -> InternalSndId -> PrevSndMsgHash -> ByteString encodeAgentMsgStr aMessage internalSndId prevMsgHash = do let privHeader = APrivHeader (unSndId internalSndId) prevMsgHash @@ -2865,15 +2869,18 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni SomeConn _ (DuplexConnection cData@ConnData {pqSupport} rqs sqs) | ratchetSyncAllowed cData || force -> do -- check queues are not switching? - when (pqSupport' /= pqSupport) $ withStore' c $ \db -> setConnPQSupport db connId pqSupport' let cData' = cData {pqSupport = pqSupport'} :: ConnData - AgentConfig {e2eEncryptVRange} <- asks config + AgentConfig {e2eEncryptVRange, smpAgentVRange} <- asks config g <- asks random (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eEncryptVRange) pqSupport' - enqueueRatchetKeyMsgs c cData' sqs e2eParams - withStore' c $ \db -> do - setConnRatchetSync db connId RSStarted - setRatchetX3dhKeys db connId pks + msgId <- withStore c $ \db -> runExceptT $ do + msgId <- storeRatchetKey db (L.head sqs) (maxVersion smpAgentVRange) e2eParams Nothing + liftIO $ do + when (pqSupport' /= pqSupport) $ setConnPQSupport db connId pqSupport' + setConnRatchetSync db connId RSStarted + setRatchetX3dhKeys db connId pks + pure msgId + lift $ enqueueRatchetKeyMsgs c cData' sqs msgId let cData'' = cData' {ratchetSyncState = RSStarted} :: ConnData conn' = DuplexConnection cData'' rqs sqs connectionStats c conn' @@ -3613,9 +3620,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar _ -> prohibited "handshake: incorrect state" >> ack (Just e2eDh, Nothing) -> do decryptClientMessage e2eDh clientMsg >>= \case - (SMP.PHEmpty, AgentRatchetKey {agentVersion, e2eEncryption}) -> do + (SMP.PHEmpty, AgentRatchetKey {agentVersion, e2eEncryption, info}) -> do conn' <- updateConnVersion conn cData agentVersion - qDuplex conn' "AgentRatchetKey" $ \a -> newRatchetKey e2eEncryption a >> ack + qDuplex conn' "AgentRatchetKey" $ \a -> newRatchetKey e2eEncryption info a >> ack (SMP.PHEmpty, AgentMsgEnvelope {agentVersion, encAgentMessage}) -> do conn' <- updateConnVersion conn cData agentVersion -- primary queue is set as Active in helloMsg, below is to set additional queues Active @@ -3658,7 +3665,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar qDuplexAckDel conn'' name a = qDuplex conn'' name a >> ackDel msgId resetRatchetSync :: AM (Connection c) resetRatchetSync - | rss `notElem` ([RSOk, RSStarted] :: [RatchetSyncState]) = do + | rss `notElem` ([RSOk, RSStarted] :: [RatchetSyncState]) = ifM ((RSStarted ==) <$> getRatchetSyncState) (pure conn') $ do let cData'' = (toConnData conn') {ratchetSyncState = RSOk} :: ConnData conn'' = updateConnection cData'' conn' cStats <- connectionStats c conn'' @@ -3699,7 +3706,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar notifySync :: AM () notifySync = qDuplex conn' "AGENT A_CRYPTO error" $ \connDuplex -> do let rss' = cryptoErrToSyncState e - when (rss `elem` ([RSOk, RSAllowed, RSRequired] :: [RatchetSyncState])) $ do + whenM ((`elem` ([RSOk, RSAllowed, RSRequired] :: [RatchetSyncState])) <$> getRatchetSyncState) $ do let cData'' = (toConnData conn') {ratchetSyncState = rss'} :: ConnData conn'' = updateConnection cData'' connDuplex cStats <- connectionStats c conn'' @@ -3795,6 +3802,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar enqueueCmd :: InternalCommand -> AM () enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand + getRatchetSyncState :: AM RatchetSyncState + getRatchetSyncState = withStore c (`getConnRatchetSync` connId) + unexpected :: BrokerMsg -> AM () unexpected r = do logServer "<--" c srv rId $ "unexpected: " <> bshow r @@ -4163,42 +4173,47 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar DuplexConnection {} -> action conn' _ -> qError $ name <> ": message must be sent to duplex connection" - newRatchetKey :: CR.RcvE2ERatchetParams 'C.X448 -> Connection 'CDuplex -> AM () - newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) = + newRatchetKey :: CR.RcvE2ERatchetParams 'C.X448 -> ByteString -> Connection 'CDuplex -> AM () + newRatchetKey e2eOtherPartyParams@(CR.E2ERatchetParams e2eVersion k1Rcv k2Rcv _) info conn'@(DuplexConnection cData'@ConnData {lastExternalSndId, pqSupport} _ sqs) = unlessM ratchetExists $ do AgentConfig {e2eEncryptVRange} <- asks config unless (e2eVersion `isCompatible` e2eEncryptVRange) (throwE $ AGENT A_VERSION) - keys <- getSendRatchetKeys + keys_ <- getSendRatchetKeys let rcVs = CR.RatchetVersions {current = e2eVersion, maxSupported = maxVersion e2eEncryptVRange} - initRatchet rcVs keys - notifyAgreed + case keys_ of + Just (keys, replyKey_) -> initRatchet rcVs keys replyKey_ >> notifyAgreed + Nothing -> withStore' c $ \db -> addProcessedRatchetKeyHash db connId rkHashRcv where rkHashRcv = rkHash k1Rcv k2Rcv rkHash k1 k2 = C.sha256Hash $ C.pubKeyBytes k1 <> C.pubKeyBytes k2 + answeredKeyHash_ = case smpDecode info of + Right (AgentRatchetInfo RatchetInfo {answeredKeyHash}) -> answeredKeyHash + _ -> Nothing ratchetExists :: AM Bool ratchetExists = withStore' c $ \db -> do exists <- checkRatchetKeyHashExists db connId rkHashRcv - unless exists $ addProcessedRatchetKeyHash db connId rkHashRcv pure exists - getSendRatchetKeys :: AM (CR.RcvE2EPrivRatchetParams 'C.X448) - getSendRatchetKeys = case rss of - RSOk -> sendReplyKey -- receiving client - RSAllowed -> sendReplyKey - RSRequired -> sendReplyKey - RSStarted -> withStore c (`getRatchetX3dhKeys` connId) -- initiating client - RSAgreed -> do - withStore' c $ \db -> setConnRatchetSync db connId RSRequired + getSendRatchetKeys :: AM (Maybe (CR.RcvE2EPrivRatchetParams 'C.X448, Maybe (CR.RcvE2ERatchetParams 'C.X448))) + getSendRatchetKeys = getRatchetSyncState >>= \rss' -> case (rss', answeredKeyHash_) of + (RSStarted, Nothing) -> Just . (,Nothing) <$> withStore c (`getRatchetX3dhKeys` connId) -- initiating client + (RSStarted, Just h) -> fmap (,Nothing) . answeredKeys h <$> withStore c (`getRatchetX3dhKeys` connId) + (_, Nothing) -> Just <$> sendReplyKey -- receiving client + (RSAgreed, Just _) -> do + withStore' c $ \db -> addProcessedRatchetKeyHash db connId rkHashRcv >> setConnRatchetSync db connId RSRequired notifyRatchetSyncError -- can communicate for other client to reset to RSRequired -- - need to add new AgentMsgEnvelope, AgentMessage, AgentMessageType -- - need to deduplicate on receiving side throwE $ AGENT (A_CRYPTO RATCHET_SYNC) + (_, Just _) -> pure Nothing where + answeredKeys h keys@(pk1, pk2, _) + | rkHash (C.publicKey pk1) (C.publicKey pk2) == h = Just keys + | otherwise = Nothing sendReplyKey = do g <- asks random (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g e2eVersion pqSupport - enqueueRatchetKeyMsgs c cData' sqs e2eParams - pure pks + pure (pks, Just e2eParams) notifyRatchetSyncError = do let cData'' = cData' {ratchetSyncState = RSRequired} :: ConnData conn'' = updateConnection cData'' conn' @@ -4210,23 +4225,33 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar conn'' = updateConnection cData'' conn' cStats <- connectionStats c conn'' notify $ RSYNC RSAgreed Nothing cStats - recreateRatchet :: CR.Ratchet 'C.X448 -> AM () - recreateRatchet rc = withStore' c $ \db -> do - setConnRatchetSync db connId RSAgreed - deleteRatchet db connId - createRatchet db connId rc + recreateRatchet :: Maybe (CR.RcvE2ERatchetParams 'C.X448) -> Maybe AMessage -> CR.Ratchet 'C.X448 -> AM () + recreateRatchet replyKey_ eready_ rc = do + cfg@AgentConfig {smpAgentVRange} <- asks config + (msgId_, r_) <- withStore c $ \db -> runExceptT $ do + msgId_ <- forM replyKey_ $ \e2eParams -> storeRatchetKey db (L.head sqs) (maxVersion smpAgentVRange) e2eParams (Just rkHashRcv) + liftIO $ do + addProcessedRatchetKeyHash db connId rkHashRcv + setConnRatchetSync db connId RSAgreed + deleteRatchet db connId + createRatchet db connId rc + r_ <- forM eready_ $ \msg -> liftIO $ storeSentMsg db cfg IM.empty (Right (Right (cData', sqs), Nothing, SMP.MsgFlags {notification = True}, vrValue msg)) >>= either E.throwIO pure . snd + pure (msgId_, r_) + forM_ msgId_ $ lift . enqueueRatchetKeyMsgs c cData' sqs + forM_ r_ $ \r -> lift $ do + r' <- submitSentMsg c r + enqueueSavedMessageB c $ mapMaybe snd $ rights [r'] -- compare public keys `k1` in AgentRatchetKey messages sent by self and other party -- to determine ratchet initilization ordering - initRatchet :: CR.RatchetVersions -> CR.RcvE2EPrivRatchetParams 'C.X448 -> AM () - initRatchet rcVs (pk1, pk2, pKem) + initRatchet :: CR.RatchetVersions -> CR.RcvE2EPrivRatchetParams 'C.X448 -> Maybe (CR.RcvE2ERatchetParams 'C.X448) -> AM () + initRatchet rcVs (pk1, pk2, pKem) replyKey_ | rkHash (C.publicKey pk1) (C.publicKey pk2) <= rkHashRcv = do rcParams <- liftError cryptoError $ CR.pqX3dhRcv (pk1, pk2, pKem) e2eOtherPartyParams - recreateRatchet $ CR.initRcvRatchet rcVs pk2 rcParams pqSupport + recreateRatchet replyKey_ Nothing $ CR.initRcvRatchet rcVs pk2 rcParams pqSupport | otherwise = do (_, rcDHRs) <- atomically . C.generateKeyPair =<< asks random rcParams <- liftEitherWith cryptoError $ CR.pqX3dhSnd (pk1, pk2, CR.APRKP CR.SRKSProposed <$> pKem) e2eOtherPartyParams - recreateRatchet $ CR.initSndRatchet rcVs k2Rcv rcDHRs rcParams - void . enqueueMessages' c cData' sqs SMP.MsgFlags {notification = True} $ EREADY lastExternalSndId + recreateRatchet replyKey_ (Just $ EREADY lastExternalSndId) $ CR.initSndRatchet rcVs k2Rcv rcDHRs rcParams checkMsgIntegrity :: PrevExternalSndId -> ExternalSndId -> PrevRcvMsgHash -> ByteString -> MsgIntegrity checkMsgIntegrity prevExtSndId extSndId internalPrevMsgHash receivedPrevMsgHash @@ -4339,32 +4364,25 @@ storeConfirmation c cData@ConnData {connId, pqSupport, connAgentVersion = v} sq liftIO $ createSndMsg db connId msgData liftIO $ createSndMsgDelivery db sq internalId -enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM () -enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do - msgId <- enqueueRatchetKey c sq e2eEncryption - mapM_ (lift . enqueueSavedMessage c msgId) $ filter (isActiveSndQ cData) sqs +enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> InternalId -> AM' () +enqueueRatchetKeyMsgs c cData (sq :| sqs) (InternalId msgId) = do + submitPendingMsg c sq + mapM_ (enqueueSavedMessage c msgId) $ filter (isActiveSndQ cData) sqs -enqueueRatchetKey :: AgentClient -> SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM AgentMsgId -enqueueRatchetKey c sq@SndQueue {connId} e2eEncryption = do - aVRange <- asks $ smpAgentVRange . config - msgId <- storeRatchetKey $ maxVersion aVRange - lift $ submitPendingMsg c sq - pure $ unId msgId - where - storeRatchetKey :: VersionSMPA -> AM InternalId - storeRatchetKey agentVersion = withStore c $ \db -> runExceptT $ do - internalTs <- liftIO getCurrentTime - (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId - let agentMsg = AgentRatchetInfo "" - agentMsgStr = smpEncode agentMsg - internalHash = C.sha256Hash agentMsgStr - let msgBody = smpEncode $ AgentRatchetKey {agentVersion, e2eEncryption, info = agentMsgStr} - msgType = agentMessageType agentMsg - -- this message is e2e encrypted with queue key, not with double ratchet - msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, pqEncryption = PQEncOff, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, sndMsgPrepData_ = Nothing} - liftIO $ createSndMsg db connId msgData - liftIO $ createSndMsgDelivery db sq internalId - pure internalId +storeRatchetKey :: DB.Connection -> SndQueue -> VersionSMPA -> CR.RcvE2ERatchetParams 'C.X448 -> Maybe ByteString -> ExceptT StoreError IO InternalId +storeRatchetKey db sq@SndQueue {connId} agentVersion e2eEncryption answeredKeyHash_ = do + internalTs <- liftIO getCurrentTime + (internalId, internalSndId, prevMsgHash) <- ExceptT $ updateSndIds db connId + let agentMsg = AgentRatchetInfo RatchetInfo {answeredKeyHash = answeredKeyHash_} + agentMsgStr = smpEncode agentMsg + internalHash = C.sha256Hash agentMsgStr + let msgBody = smpEncode $ AgentRatchetKey {agentVersion, e2eEncryption, info = agentMsgStr} + msgType = agentMessageType agentMsg + -- this message is e2e encrypted with queue key, not with double ratchet + msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, pqEncryption = PQEncOff, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, sndMsgPrepData_ = Nothing} + liftIO $ createSndMsg db connId msgData + liftIO $ createSndMsgDelivery db sq internalId + pure internalId -- encoded AgentMessage -> encoded EncAgentMessage agentRatchetEncrypt :: DB.Connection -> ConnData -> ByteString -> (PQSupport -> Int) -> Maybe PQEncryption -> CR.VersionE2E -> ExceptT StoreError IO (ByteString, PQEncryption) diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index afe698769..5f7977e45 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -79,6 +79,7 @@ module Simplex.Messaging.Agent.Protocol SMPConfirmation (..), AgentMsgEnvelope (..), AgentMessage (..), + RatchetInfo (..), RequestSignature (..), AgentMessageType (..), APrivHeader (..), @@ -928,7 +929,7 @@ data AgentMessage | -- AgentConnInfoReply is used by accepting party in duplexHandshake mode (v2), allowing to include reply queue(s) in the initial confirmation. -- It made removed REPLY message unnecessary. AgentConnInfoReply (NonEmpty SMPQueueInfo) ConnInfo - | AgentRatchetInfo ByteString + | AgentRatchetInfo RatchetInfo | AgentMessage APrivHeader AMessage | AgentServiceRequest (NonEmpty SMPQueueInfo) (Maybe RequestSignature) MsgBody | AgentServiceResponse MsgBody @@ -939,7 +940,7 @@ instance Encoding AgentMessage where smpEncode = \case AgentConnInfo cInfo -> smpEncode ('I', Tail cInfo) AgentConnInfoReply smpQueues cInfo -> smpEncode ('D', smpQueues, Tail cInfo) -- 'D' stands for "duplex" - AgentRatchetInfo info -> smpEncode ('R', Tail info) + AgentRatchetInfo info -> smpEncode ('R', info) AgentMessage hdr aMsg -> smpEncode ('M', hdr, aMsg) AgentServiceRequest qs sig_ body -> smpEncode ('A', qs, sig_, Tail body) AgentServiceResponse body -> smpEncode ('P', Tail body) @@ -948,13 +949,23 @@ instance Encoding AgentMessage where smpP >>= \case 'I' -> AgentConnInfo . unTail <$> smpP 'D' -> AgentConnInfoReply <$> smpP <*> (unTail <$> smpP) - 'R' -> AgentRatchetInfo . unTail <$> smpP + 'R' -> AgentRatchetInfo <$> smpP 'M' -> AgentMessage <$> smpP <*> smpP 'A' -> AgentServiceRequest <$> smpP <*> smpP <*> (unTail <$> smpP) 'P' -> AgentServiceResponse . unTail <$> smpP 'J' -> AgentRejection . unTail <$> smpP _ -> fail "bad AgentMessage" +data RatchetInfo = RatchetInfo {answeredKeyHash :: Maybe ByteString} + deriving (Eq, Show) + +instance Encoding RatchetInfo where + smpEncode RatchetInfo {answeredKeyHash} = smpEncode answeredKeyHash + smpP = do + answeredKeyHash <- smpP <|> pure Nothing + _ <- A.takeByteString + pure RatchetInfo {answeredKeyHash} + -- internal type for storing message type in the database data AgentMessageType = AM_CONN_INFO diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 25ea50495..21c087d46 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/src/Simplex/Messaging/Agent/Store/AgentStore.hs @@ -76,6 +76,7 @@ module Simplex.Messaging.Agent.Store.AgentStore getExpiredServiceConns, deleteExpiredServiceRequests, getDeletedWaitingDeliveryConnIds, + getConnRatchetSync, setConnRatchetSync, addProcessedRatchetKeyHash, checkRatchetKeyHashExists, @@ -2797,6 +2798,11 @@ getDeletedWaitingDeliveryConnIds :: DB.Connection -> IO [ConnId] getDeletedWaitingDeliveryConnIds db = map fromOnly <$> DB.query_ db "SELECT conn_id FROM connections WHERE deleted_at_wait_delivery IS NOT NULL" +getConnRatchetSync :: DB.Connection -> ConnId -> IO (Either StoreError RatchetSyncState) +getConnRatchetSync db connId = + firstRow fromOnly SEConnNotFound $ + DB.query db "SELECT ratchet_sync_state FROM connections WHERE conn_id = ?" (Only connId) + setConnRatchetSync :: DB.Connection -> ConnId -> RatchetSyncState -> IO () setConnRatchetSync db connId ratchetSyncState = DB.execute db "UPDATE connections SET ratchet_sync_state = ? WHERE conn_id = ?" (ratchetSyncState, connId) diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs index 37b81f3eb..511064cf5 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20260929_ratchet_indexes.hs @@ -11,6 +11,15 @@ m20260929_ratchet_indexes = [r| DROP INDEX idx_skipped_messages_conn_id; CREATE INDEX idx_skipped_messages_conn_id ON skipped_messages(conn_id, skipped_message_id); + +DELETE FROM processed_ratchet_key_hashes AS h +WHERE EXISTS ( + SELECT 1 FROM processed_ratchet_key_hashes d + WHERE d.conn_id = h.conn_id AND d.hash = h.hash AND d.processed_ratchet_key_hash_id < h.processed_ratchet_key_hash_id +); + +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes(conn_id, processed_ratchet_key_hash_id); |] @@ -18,6 +27,8 @@ down_m20260929_ratchet_indexes :: Text down_m20260929_ratchet_indexes = [r| DROP INDEX idx_processed_ratchet_key_hashes_conn_id; +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); DROP INDEX idx_skipped_messages_conn_id; CREATE INDEX idx_skipped_messages_conn_id ON skipped_messages(conn_id); |] diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql index e7d90ae27..55a9ab78d 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql @@ -1177,7 +1177,7 @@ CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON smp_agent_test_proto -CREATE INDEX idx_processed_ratchet_key_hashes_hash ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, hash); +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, hash); diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs index d2ca60a35..a1e70a2e2 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20260929_ratchet_indexes.hs @@ -9,6 +9,14 @@ import Database.SQLite.Simple.QQ (sql) m20260929_ratchet_indexes :: Query m20260929_ratchet_indexes = [sql| +DELETE FROM processed_ratchet_key_hashes AS h +WHERE EXISTS ( + SELECT 1 FROM processed_ratchet_key_hashes d + WHERE d.conn_id = h.conn_id AND d.hash = h.hash AND d.processed_ratchet_key_hash_id < h.processed_ratchet_key_hash_id +); + +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes(conn_id, processed_ratchet_key_hash_id); |] @@ -16,4 +24,6 @@ down_m20260929_ratchet_indexes :: Query down_m20260929_ratchet_indexes = [sql| DROP INDEX idx_processed_ratchet_key_hashes_conn_id; +DROP INDEX idx_processed_ratchet_key_hashes_hash; +CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash); |] diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index 47fd154a0..e118a17eb 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -573,10 +573,6 @@ CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hash conn_id, hash ); -CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes( - conn_id, - hash -); CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages( conn_id, rcpt_internal_id @@ -641,6 +637,10 @@ CREATE INDEX idx_connections_deleted ON connections(deleted); CREATE INDEX idx_connections_service_request_expires_at ON connections( service_request_expires_at ); +CREATE UNIQUE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes( + conn_id, + hash +); CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_hashes( conn_id, processed_ratchet_key_hash_id diff --git a/tests/AgentTests/ConnectionRequestTests.hs b/tests/AgentTests/ConnectionRequestTests.hs index d76d8d305..40ca12cf9 100644 --- a/tests/AgentTests/ConnectionRequestTests.hs +++ b/tests/AgentTests/ConnectionRequestTests.hs @@ -362,6 +362,10 @@ connectionRequestTests = smpEncodingTest $ AgentServiceRequest [qInfo] Nothing "service request payload" smpEncodingTest $ AgentServiceResponse "service response payload" smpEncodingTest $ AgentRejection "rejected: not allowed" + it "should serialize and parse ratchet key info" $ do + smpDecode "R" `shouldBe` Right (AgentRatchetInfo RatchetInfo {answeredKeyHash = Nothing}) + smpDecode "R1\3abcdef" `shouldBe` Right (AgentRatchetInfo RatchetInfo {answeredKeyHash = Just "abc"}) + smpEncodingTest $ AgentRatchetInfo RatchetInfo {answeredKeyHash = Just "0123456789abcdef0123456789abcdef"} where smpEncodingTest :: (Encoding a, Eq a, Show a, HasCallStack) => a -> Expectation smpEncodingTest a = smpDecode (smpEncode a) `shouldBe` Right a diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index ac9ae7a2c..727c47d5a 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -70,7 +70,7 @@ import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import Data.Either (isRight) import Data.Int (Int64) -import Data.List (find, isPrefixOf, isSuffixOf) +import Data.List (find, isInfixOf, isPrefixOf, isSuffixOf) import Data.List.NonEmpty (NonEmpty (..)) import qualified Data.Map as M import Data.Maybe (isJust, isNothing) @@ -465,6 +465,14 @@ functionalAPITests ps = do testRatchetSyncSimultaneous ps it "should ignore replayed ratchet key after expired hashes are deleted" $ testRatchetSyncReplayedKey ps + it "should synchronize ratchets when synchronization is forced again" $ + testRatchetSyncRepeated ps + it "should not mark ratchet key as processed when ratchet recreation fails" $ + testRatchetSyncFailedKeyNotProcessed ps + it "should not store reply ratchet key when ratchet recreation fails" $ + testRatchetSyncFailedRecreationNoReply ps + it "should not store ratchet key when starting synchronization fails" $ + testRatchetSyncStartFailedNoKey ps #endif describe "Subscription mode OnlyCreate" $ do it "messages delivered only when polled" $ @@ -2771,6 +2779,96 @@ testRatchetSyncReplayedKey ps = withAgentClients2 $ \alice bob -> do runRight_ $ exchangeGreetingsMsgIds alice bobId 10 bob2 aliceId 7 disposeAgentClient bob2 +testRatchetSyncRepeated :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncRepeated ps = withAgentClients2 $ \alice bob -> do + (aliceId, bobId, bob2) <- startRatchetSyncOffline ps alice bob + ConnectionStats {ratchetSyncState = rss2} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn True + rss2 `shouldBe` RSStarted + + withSmpServerStoreMsgLogOn ps testPort $ \_ -> do + concurrently_ + (getInAnyOrder alice [ratchetSyncP' bobId RSAgreed, serverUpP]) + (getInAnyOrder bob2 [ratchetSyncP' aliceId RSAgreed, serverUpP]) + runRight_ $ do + get alice =##> ratchetSyncP bobId RSAgreed + get alice =##> ratchetSyncP bobId RSOk + get bob2 =##> ratchetSyncP aliceId RSOk + msgId <- sendMessage alice bobId SMP.noMsgFlags "hello" + get alice ##> ("", bobId, SENT msgId) + get bob2 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False + ackMessage bob2 aliceId 8 Nothing + map fst <$> processedRatchetKeyHashes bob2 `shouldReturn` [aliceId, aliceId] + disposeAgentClient bob2 + +testRatchetSyncFailedKeyNotProcessed :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncFailedKeyNotProcessed ps = withAgentClients2 $ \alice bob -> do + (aliceId, bobId, bob2) <- startRatchetSyncOffline ps alice bob + withTransaction (store $ agentEnv bob2) $ \db -> + DB.execute_ db "UPDATE ratchets SET x3dh_priv_key_1 = NULL" + + withSmpServerStoreMsgLogOn ps testPort $ \_ -> + concurrently_ + (getInAnyOrder alice [ratchetSyncP' bobId RSAgreed, serverUpP]) + (getInAnyOrder bob2 [x3dhKeysNotFoundP aliceId, serverUpP]) + map fst <$> processedRatchetKeyHashes alice `shouldReturn` [bobId] + processedRatchetKeyHashes bob2 `shouldReturn` [] + disposeAgentClient bob2 + where + x3dhKeysNotFoundP :: ConnId -> ATransmission -> Bool + x3dhKeysNotFoundP cId = \case + (_, cId', AEvt SAEConn (ERR (A.INTERNAL e))) -> cId' == cId && "SEX3dhKeysNotFound" `isPrefixOf` e + _ -> False + +testRatchetSyncFailedRecreationNoReply :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncFailedRecreationNoReply ps = withAgentClients2 $ \alice bob -> do + (_, bobId, bob2) <- startRatchetSyncOffline ps alice bob + aliceSndMsgs <- sndMessages alice + withTransaction (store $ agentEnv alice) $ \db -> + DB.execute_ db "CREATE TRIGGER fail_ratchet_insert BEFORE INSERT ON ratchets BEGIN SELECT RAISE(ABORT, 'ratchet insert failed'); END" + withSmpServerStoreMsgLogOn ps testPort $ \_ -> + concurrently_ + (getInAnyOrder alice [ratchetInsertFailedP bobId, serverUpP]) + (getInAnyOrder bob2 [serverUpP]) + processedRatchetKeyHashes alice `shouldReturn` [] + sndMessages alice `shouldReturn` aliceSndMsgs + disposeAgentClient bob2 + where + ratchetInsertFailedP :: ConnId -> ATransmission -> Bool + ratchetInsertFailedP cId = \case + (_, cId', AEvt SAEConn (ERR (A.INTERNAL e))) -> cId' == cId && "ratchet insert failed" `isInfixOf` e + _ -> False + +testRatchetSyncStartFailedNoKey :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testRatchetSyncStartFailedNoKey ps = withAgentClients2 $ \alice bob -> do + (aliceId, _, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ -> + setupDesynchronizedRatchet alice bob + bobSndMsgs <- sndMessages bob2 + withTransaction (store $ agentEnv bob2) $ \db -> + DB.execute_ db "CREATE TRIGGER fail_ratchet_update BEFORE UPDATE ON ratchets BEGIN SELECT RAISE(ABORT, 'ratchet update failed'); END" + Left (A.INTERNAL e) <- runExceptT $ synchronizeRatchet bob2 aliceId PQSupportOff False + e `shouldContain` "ratchet update failed" + ConnectionStats {ratchetSyncState} <- runRight $ getConnectionServers bob2 aliceId + ratchetSyncState `shouldBe` RSRequired + withTransaction (store $ agentEnv bob2) (`DB.query_` "SELECT conn_id, pq_support FROM connections") `shouldReturn` [(aliceId, PQSupportOn)] + sndMessages bob2 `shouldReturn` bobSndMsgs + disposeAgentClient bob2 + +startRatchetSyncOffline :: HasCallStack => (ASrvTransport, AStoreType) -> AgentClient -> AgentClient -> IO (ConnId, ConnId, AgentClient) +startRatchetSyncOffline ps alice bob = do + (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn ps testPort $ \_ -> + setupDesynchronizedRatchet alice bob + ("", "", DOWN _ _) <- nGet alice + ("", "", DOWN _ _) <- nGet bob2 + ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn False + ratchetSyncState `shouldBe` RSStarted + pure (aliceId, bobId, bob2) + +processedRatchetKeyHashes :: AgentClient -> IO [(ConnId, ByteString)] +processedRatchetKeyHashes c = withTransaction (store $ agentEnv c) (`DB.query_` "SELECT conn_id, hash FROM processed_ratchet_key_hashes") + +sndMessages :: AgentClient -> IO [(ConnId, Int64)] +sndMessages c = withTransaction (store $ agentEnv c) (`DB.query_` "SELECT conn_id, internal_id FROM snd_messages ORDER BY conn_id, internal_id") + getMsg :: AgentClient -> ConnId -> ExceptT AgentErrorType IO a -> ExceptT AgentErrorType IO a getMsg c cId action = do liftIO $ noMessages c "nothing should be delivered before GET" diff --git a/tests/AgentTests/SchemaDump.hs b/tests/AgentTests/SchemaDump.hs index d9aa79513..ee401bf19 100644 --- a/tests/AgentTests/SchemaDump.hs +++ b/tests/AgentTests/SchemaDump.hs @@ -48,6 +48,7 @@ schemaDumpTest = do it "verify strict tables" testVerifyStrict it "should NOT create user record for new database" testUsersMigrationNew it "should create user record for old database" testUsersMigrationOld + it "should remove duplicate ratchet key hashes before adding unique index" testRatchetKeyHashesUniqueMigration testVerifySchemaDump :: IO () testVerifySchemaDump = do @@ -114,12 +115,28 @@ testUsersMigrationOld = do `shouldReturn` ([Only (1 :: Int)]) closeDBStore st' +testRatchetKeyHashesUniqueMigration :: IO () +testRatchetKeyHashesUniqueMigration = do + let beforeUnique = takeWhile (("m20260929_ratchet_indexes" /=) . name) appMigrations + Right st <- createDBStore (DBOpts testDB [] "" False True TQOff) beforeUnique (MigrationConfig MCError Nothing) + withTransaction' st $ \db -> do + SQL.execute_ db "INSERT INTO users (user_id) VALUES (1)" + SQL.execute_ db "INSERT INTO connections (conn_id, conn_mode, user_id) VALUES (x'01', 'INV', 1), (x'02', 'INV', 1)" + SQL.execute_ db "INSERT INTO processed_ratchet_key_hashes (conn_id, hash) VALUES (x'01', x'aa'), (x'01', x'aa'), (x'01', x'bb'), (x'02', x'aa'), (x'01', x'aa')" + closeDBStore st + Right st' <- createDBStore (DBOpts testDB [] "" False True TQOff) appMigrations (MigrationConfig MCYesUp Nothing) + withTransaction' st' (`SQL.query_` "SELECT processed_ratchet_key_hash_id FROM processed_ratchet_key_hashes ORDER BY processed_ratchet_key_hash_id") + `shouldReturn` [Only (1 :: Int), Only 3, Only 4] + closeDBStore st' + skipComparisonForDownMigrations :: [String] skipComparisonForDownMigrations = [ -- on down migration idx_messages_internal_snd_id_ts index moves down to the end of the file "m20230814_indexes", -- snd_secure and last_broker_ts columns swap order on down migration - "m20250322_short_links" + "m20250322_short_links", + -- on down migration idx_processed_ratchet_key_hashes_hash index moves down to the end of the file + "m20260929_ratchet_indexes" ] getSchema :: FilePath -> FilePath -> IO String