Compare commits

...
Author SHA1 Message Date
Evgeny Poberezkin a2d777bda0 6.4.4.1 2025-08-23 20:05:10 +01:00
Evgeny e48bedeaf2 servers: fix server pages when source code is not specified, include commit SHA in printed version and in web page. (#1608)
* smp server: fix server pages when source code is not specified

* servers: include git commit in version

* flexible alpn

* fix test

* fix ghc 8.10.7 build
2025-08-23 19:59:00 +01:00
Evgeny Poberezkin a2d35281b2 6.4.4.0 2025-08-18 16:22:24 +01:00
spaced4ndy 46035af9a3 agent: add context to SEMsgNotFound error (#1604) 2025-08-18 10:59:57 +01:00
Evgeny Poberezkin 4b7fc34fe3 enable all tests 2025-08-17 10:40:12 +01:00
Evgeny 96e8b4a146 smp server: wrap all queries in transactions (#1603)
* smp server: wrap all queries in transactions

* fix test

* fix schema test
2025-08-17 10:38:20 +01:00
Evgeny 2cedb66667 agent: chat relay address type for short links (#1602) 2025-08-15 07:24:37 +01:00
Evgeny e345671c76 agent: extend xrcp certificate validity 1 hour in the past, to allow out of sync clocks (#1601) 2025-08-14 22:48:11 +01:00
Evgeny 86fb2cddc5 smp server: support short SimpleX addresses in server information / pages (#1600) 2025-08-12 07:31:32 +01:00
Evgeny Poberezkin 931c533a3d 6.4.3.0 2025-08-05 12:30:28 +01:00
Evgeny 79ba60e3ad agent: update contact request server hosts to match server in short link (#1597) 2025-08-05 12:29:50 +01:00
Evgeny fb477b24d7 smp server: support short link URI as queue identifier in control port commands (#1596) 2025-08-05 08:18:39 +01:00
Evgeny Poberezkin 9f263e8f3e docs: changelog 2025-08-02 09:21:44 +01:00
Evgeny Poberezkin db325cb81f 6.4.2.0 2025-07-31 09:21:11 +01:00
Evgeny b167d01f8a rfc: using the same profile from multiple devices (#1588) 2025-07-30 20:54:03 +01:00
Evgenyandsh f4e7469f96 xftp server: prometheus metrics (#1595)
* xftp server: prometheus metrics

* update

Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>

---------

Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>
2025-07-30 16:18:49 +01:00
Evgeny 4647d69d4b smp server: do not include previously blocked queues in stats, prevent leak of client threads (#1593) 2025-07-30 11:12:57 +01:00
sh 9ab071d62c ci: fix reproduce-schedule (#1594) 2025-07-30 09:17:28 +01:00
Evgeny Poberezkin f4c09ac51f Merge branch 'stable' 2025-07-29 09:09:36 +01:00
sh fc581bf729 github/workflows: fix reproduce schedule (#1587) 2025-07-16 11:42:14 +01:00
38 changed files with 527 additions and 124 deletions
+2 -2
View File
@@ -25,7 +25,7 @@ jobs:
- name: Execute reproduce script
run: |
${GITHUB_WORKSPACE}/scripts/reproduce-builds.sh "$TAG"
${GITHUB_WORKSPACE}/scripts/simplexmq-reproduce-builds.sh "$TAG" || :
- name: Check if build has been reproduced
env:
@@ -33,7 +33,7 @@ jobs:
user: ${{ secrets.STATUS_SIMPLEX_WEBHOOK_USER }}
pass: ${{ secrets.STATUS_SIMPLEX_WEBHOOK_PASS }}
run: |
if [ -f "${GITHUB_WORKSPACE}/$TAG/_sha256sums" ]; then
if [ -f "${GITHUB_WORKSPACE}/${TAG}-simplexmq/_sha256sums" ]; then
exit 0
else
curl --proto '=https' --tlsv1.2 -sSf \
+31
View File
@@ -1,3 +1,34 @@
# 6.4.4
Servers:
- fix server pages when source code is not specified.
- include commit SHA in printed version and in web page (#1608).
SMP server:
- support short SimpleX addresses in server information page (#1600).
- wrap all queries in transactions (#1603).
SMP agent:
- chat relay address type for short links (#1602).
- extend xrcp certificate validity 1 hour in the past, to allow out of sync clocks (#1601).
# 6.4.3
SMP agent:
- fix some connection errors by updating contact request server hosts to match server in short link (#1597).
SMP server:
- support short link URI as queue identifier in control port commands (#1596).
# 6.4.2
SMP server:
- fix memory leak when connection interrupts straight after client connects.
- do not include repeated queue blocking into stats/quota.
XFTP server:
- prometheus metrics
# 6.4.1
SMP protocol:
+5 -2
View File
@@ -223,11 +223,14 @@
<table id="public-info">
<tr class="text-grey-black dark:text-white text-base">
<td>Server version:</td>
<td>${version}</td>
<td>${version}<x-commit> / <a href="${commitSourceCode}/commit/${commit}" target="_blank">${shortCommit}</a></x-commit></td>
</tr>
<tr class="text-grey-black dark:text-white text-base">
<td>Source code:</td>
<td><a href="${sourceCode}" target="_blank">${sourceCode}</a></td>
<td>
<x-sourceCode><a href="${sourceCode}" target="_blank">${sourceCode}</a></x-sourceCode>
<x-noSourceCode>add to smp-server.ini (required by <a href="https://github.com/simplex-chat/simplexmq/blob/stable/LICENSE" target="_blank">AGPLv3</a>)</x-noSourceCode>
</td>
</tr>
<x-website>
<tr class="text-grey-black dark:text-white text-base">
+2
View File
@@ -512,6 +512,8 @@
element.innerHTML = 'This is a one-time link of the SimpleX&nbsp;network user'
} else if (url.includes('/c')) {
element.innerHTML = 'This is a public channel address on SimpleX&nbsp;network'
} else if (url.includes('/r')) {
element.innerHTML = 'This is a chat relay address on SimpleX&nbsp;network'
}
}
</script>
+15 -6
View File
@@ -12,6 +12,7 @@ import Data.Char (toUpper)
import Data.IORef (readIORef)
import Data.Maybe (fromMaybe)
import Data.String (fromString)
import qualified Data.Text as T
import Data.Text.Encoding (encodeUtf8)
import Network.Socket (getPeerName)
import Network.Wai (Application, Request (..))
@@ -22,8 +23,9 @@ import qualified Network.Wai.Handler.Warp.Internal as WI
import qualified Network.Wai.Handler.WarpTLS as WT
import Simplex.Messaging.Encoding.String (strEncode)
import Simplex.Messaging.Server (AttachHTTP)
import Simplex.Messaging.Server.CLI (simplexmqCommit)
import Simplex.Messaging.Server.Information
import Simplex.Messaging.Server.Main (EmbeddedWebParams (..), WebHttpsParams (..))
import Simplex.Messaging.Server.Main (EmbeddedWebParams (..), WebHttpsParams (..), simplexmqSource)
import Simplex.Messaging.Transport (simplexMQVersion)
import Simplex.Messaging.Transport.Client (TransportHost (..))
import Simplex.Messaging.Util (tshow)
@@ -103,6 +105,7 @@ generateSite si onionHost sitePath = do
createLinkPage "a"
createLinkPage "c"
createLinkPage "g"
createLinkPage "r"
createLinkPage "i"
logInfo $ "Generated static site contents at " <> tshow sitePath
where
@@ -116,7 +119,7 @@ generateSite si onionHost sitePath = do
serverInformation :: ServerInformation -> Maybe TransportHost -> ByteString
serverInformation ServerInformation {config, information} onionHost = render E.indexHtml substs
where
substs = substConfig <> maybe [] substInfo information <> [("onionHost", strEncode <$> onionHost)]
substs = substConfig <> substInfo <> [("onionHost", strEncode <$> onionHost)]
substConfig =
[ ( "persistence",
Just $ case persistence config of
@@ -131,7 +134,7 @@ serverInformation ServerInformation {config, information} onionHost = render E.i
]
yesNo True = "Yes"
yesNo False = "No"
substInfo spi =
substInfo =
concat
[ basic,
maybe [("usageConditions", Nothing), ("usageAmendments", Nothing)] conds (usageConditions spi),
@@ -143,10 +146,16 @@ serverInformation ServerInformation {config, information} onionHost = render E.i
]
where
basic =
[ ("sourceCode", Just . encodeUtf8 $ sourceCode spi),
[ ("sourceCode", if T.null sc then Nothing else Just (encodeUtf8 sc)),
("noSourceCode", if T.null sc then Just "none" else Nothing),
("version", Just $ B.pack simplexMQVersion),
("commitSourceCode", Just $ encodeUtf8 $ maybe (T.pack simplexmqSource) sourceCode information),
("shortCommit", Just $ B.pack $ take 7 simplexmqCommit),
("commit", Just $ B.pack simplexmqCommit),
("website", encodeUtf8 <$> website spi)
]
spi = fromMaybe (emptyServerInfo "") information
sc = sourceCode spi
conds ServerConditions {conditions, amendments} =
[ ("usageConditions", Just $ encodeUtf8 conditions),
("usageAmendments", encodeUtf8 <$> amendments)
@@ -228,8 +237,8 @@ section_ label content' src =
(inside, next') ->
let next = B.drop (B.length endMarker) next'
in case content' of
Nothing -> before <> next -- collapse section
Just content -> before <> item_ label content inside <> section_ label content' next
Just content | not (B.null content) -> before <> item_ label content inside <> section_ label content' next
_ -> before <> next -- collapse section
where
startMarker = "<x-" <> label <> ">"
endMarker = "</x-" <> label <> ">"
+104
View File
@@ -0,0 +1,104 @@
# Using the same profile from multiple devices
## Problem
Double Ratchet algorithm makes it hard to send/receive messages sent to the user from different devices, as each message changes the state of Double Ratchet keys, and these state changes must be strictly sequential and they cannot be reversed (although skipping is possible).
Traditional approach for multi-device converts each direct conversation into a group, where each device participates as a member. Likewise, for group conversations each device also participates as a member. While these members *look* as if they are the same user to others, a very simple client app modification may show device ID for each message, and the communication peers, both in direct chats and in groups would know how many devices a user has and which device the user sent the message from. In addition to that, with this approach communication peers can send different messages to different devices (it can be prevented by provider who would request that only message key is encrypted with DR, while the encrypted message is the same) or withheld from some devices (it cannot be prevented by provider, as it cannot add key to the communication in case it is missing, and cannot withhold the message completely too). These opens various vectors for targeted attacks, e.g.:
- tracking movements of the user: once each devices is identified as "desk" and "phone" it would allow to know where the user is at a given time.
- manipulating information by sending messages to one device (to have proof it was sent) and withholding from others, or sending different messages if the protocol allows it.
In addition to that, the specific implementation of this approach in Signal compromises break-in recovery property (aka post-compromise security) of Double-Ratchet algorithm, making its design ineffective - the only reason to have the second ratchet in DR algorithm is to provide break-in recovery, without it a much simpler design with a single ratchet is sufficient. See [this paper](https://eprint.iacr.org/2021/626.pdf) for details.
While this limitation can be addressed with notifications when a new device is added and per-device keys, we still find the remaining attack vectors on user security and privacy to be unacceptable, and opening unsuspecting users to various criminal actions - and it is wrong to say that would only affect security conscious users, and most people would not be affected by these risks. Allowing potential criminals in groups to know which device you are currently using is a real risk for all users.
Another approach was offered by Threema that is ["mediator" server](https://threema.com/en/blog/md-architectural-overview) where the state of encryption ratchets is stored server-side. While it protects the user from their communication peers, it increases required level of trust to the servers, and in case of SimpleX network it would expose the knowledge of who communicates to whom. So while the idea of server-side storage of encryption state is promising, it has to be per-connection, to retain "no-accounts" property of SimpleX messaging network.
Also see [FAQ](https://simplex.chat/faq/#why-cant-i-use-the-same-profile-on-different-devices) and [this issue](https://github.com/simplex-chat/simplex-chat/issues/444#issuecomment-3066968358).
## Proposed solution
One of the ideas presented in FAQ - to store the state of Double Ratchet algorithm in the encrypted container on the server seems promising. The RFC develops this idea.
### Considerations for the design
1. The largest ratchet state size with the current implementation is less than 8kb (which is achieved when both sides shared PQ keys and ciphertexts), so while it cannot fit in the same transport blocks together with sent and received messages, it would fit in one transport block.
2. Protocol commands and events may be changed (even if at the cost of slightly reducing message size) can fit the hash of the ratchet state (32 bytes sha256 would be sufficient), so that the client can determine whether it has the most recent ratchet state or if it needs to retrieve the latest copy. Message size reduction won't affect the users because we use compression, and there is a substantial reserve.
3. Client commands that modify ratchet state would include the hash of the previous ratchet state so that the server can reject or ignore the command in case the previous ratchet state is different or in case command is repeated in case of lost response).
4. The client does not need to retrieve message state for each encryption and decryption operation - it can "speculatively" use the ratchet state it has, and receive correct ratchet state in the "error" response after attempting encryption based on incorrect ratchet state.
## Proposed protocol design
Ratchet state will be stored on the same server that stores message queue, as part of message queue record. 8kb is a sufficient size for this blob (the actual max size is 7800 bytes). The server would also store the hashes of the current and, possibly, the previous ratchet states (TBC).
While ratchet is used for duplex connection, the connection still has primary queue, and with redundancy the same ratchet state can be stored on all secondary queues.
Ratchet state will be encrypted using secret_box - a symmetric encryption scheme, so PQ-resistant. If ratchet state is stored on more than one server, it has to be encrypted with a different key for each server.
Questions: how to rotate the key used to store ratchet? Should key used to encrypt ratchet rotate at the same time when queue is rotated? The latter is a logical option, as it prevents additional complexity and solves the problem anyway. A possible option is to have "ratchet version" that will be used to advance the key used to encrypt ratchet via HKDF.
Security considerations: the scheme may reduce break-in recovery to the points queues are rotated, unless there is some randomness mixed-in into the key derivation (the key used to encrypt ratchet state). But including randomness would defeat the purpose, as other devices wouldn't be able to access the ratchets. Another approach would be to have each device use its own key for encryption, and encrypt to all keys of all devices (or to encrypt key, to avoid size increase). Having multiple encryptions would show how many devices use the queue, but servers already can observe it, so it is a better tradeoff. Another idea would be to rotate the key used to authorize queue commands - we already support multiple recipient keys, and it can be used for multi-device scenario. That would partially mitigate break-in attacks as the attacker who obtained the key from ratchet state would be able to decrypt it, but won't be able to decrypt it (the attacker collusion with the server is not mitigated). Yet another idea would be for each party (device) to share its private (or encapsulation) key and to have a symmetric key (used to encrypt the ratchet state) encrypted (encapsulated) separately for each device. This would reduce the size of the stored data to `ratchet size` + `encrypted key size` * N, so even in case of PQ encryption (e.g. sntrup) the size required to store the ratchet would be under transport block size, while limiting it to say 4-8 devices, which is sufficient.
To participate in multi-device scheme the devices would join the usual group that will be used to share public (encapsulation) device keys and to communicate updates to conversations that were received by the currently "active" device. "Active" means the device that received or sent and processed the message, and while only one device can receive messages from a given queue, device "active" state may be determined per queue, allowing concurrent usage.
The scheme must be resilient to state updates being lost, and in case of direct messages it would result in some messages not being shown (or shown as skipped), while conversation preference and profile updates can be re-requested from peers, while the current profile of the user would become the latest. Likewise, for groups state updates ca be requested from super-peers or for decentralized groups - from owners. Maintaining chat state consistency is an important consideration, but is not a focus of this RFC - the focus is managing message delivery and DR encryption for multiple devices. Other multi-device schemes have the same issues with state consistency. Partially, the profile state consistency can be improved by using a single shared queue (or set of queues) to store user's profile and chat preferences to synchronize profile updates asynchronously between the devices.
## The protocol to send the message
`rsi` - ratchet state on device `i`.
`enc(rs)` - current authoritative ratchet state on the server.
`pt` and `ct` - plaintext and ciphertext messages.
Encryption is a state transition function ratchetEnc: `(ct, rs') = ratchetEnc(pt, rs)`.
1. Device encrypts the message using the stored ratchet state: `(ct, rsi') = ratchetEnc(pt, rsi)`
2. Device sends modified encrypted ratchet state and the hash of the previous encrypted state to the server that stores the queue: `RSET (hash(enc(rsi)), enc(rsi'))`.
3. If the hash of the previous state matches state stored on the server (`hash(enc(rsi)) == hash(enc(rs))`), the server updates the state and responds with `ratchet_ok` (that may include the current state or it's hash, for validation). If the hash is different, the server responds with `bad_ratchet(enc(rs))` message that includes the correct ratchet state. These updates must be atomic. In this case device has to update the local ratchet state (provided it can decrypt it), and repeat encryption attempt. If device cannot decrypt the provided ratchet state, it means that the connection is disrupted (possibly, device is removed from device group, but missed the notifications).
4. After successful state update in primary receiving queue, the device would update it in secondary receiving queues.
5. Device sends encrypted message as usual, via proxy that must be different both from the server that stores the ratchet and from the destination server.
6. Device broadcasts sent message and new ratchet state to other devices in the device group.
This protocol is simple, and it minimizes requests when sending the message to one additional request to update ratchet state in most cases, only requiring two requests when device state was not updated via device group prior to message sending attempt.
## The protocol to receive the message
Decryption is also a state transition function: `(pt, rs') = ratchetDec(ct, rs)`
1. Server sends the message to the device (can be in response to SUB or ACK commands, or with active subscription). Pushed message would include the hash of the currently stored ratchet state: `hash(enc(rs))`.
2. If device has the ratchet state with the same hash (`hash(enc(rs)) == hash(enc(rsi))`), it decrypts the message: `(pt, rsi') = ratchetDec(ct, rsi)`.
3. If device has ratchet state with a different hash, it requests ratchet from the server with additional protocol command `RGET` with response `RCHT (enc(rs))` and updates the local state.
4. Device decrypts the message `(pt, rsi') = ratchetDec(ct, rsi)` and processes it as usual.
5. Device sends acknowledgement to the server as usual, but now it includes the new ratchet state and the hash of the previous state: `ACK msgId (hash(enc(rsi)), enc(rsi'))`
6. The server compares ratchet state with stored state hash, and in case it matches it processes `ACK` and responds with `OK` as usual (or `NO_MSG` in case msgId is incorrect, also as usual - it would happen in repeated ACK requests). If ratchet state hash does not match, the server would respond with `bad_ratchet(enc(rs))` - which means that the message was already processed by another device and ratchet was advanced. This is a complex scenario, as the client has to either revert the change from message processing or somehow combine the change with the updates communicated via device group (as a side note, device group can simply re-broadcast messages, not state updates, but it will result in state divergence between devices when different messages are lost).
Unlike sending messages, this flow does not require any additional requests in most cases, only requiring requesting message state reconciliation when the same message was received and processed by more than one client, but it does not require re-acknowledgement.
## Challenges
This is an idea of the design rather than the actual design, as it requires more thinking about:
- how to handle concurrent ratchet state updates,
- "active" status transitions per queue,
- avoiding concurrent subscriptions to queues from multiple devices,
- state updates and synchronization between devices,
- handling skipped messages,
- costs to update ratchets in bulk send scenario - this scheme would substantially increase costs of preparing large broadcasts, and it makes this scheme not acceptable for chat relays. Which means that "profile" on desktop used as chat relay won't be synched to other devices.
- etc.
## Advantages
The communication peers won't know how many devices the user has, and which device was used to send the message. Also, the communication peers won't be able to send different messages to different user's devices, or to withhold messages from some devices.
+5 -1
View File
@@ -1,7 +1,7 @@
cabal-version: 1.12
name: simplexmq
version: 6.4.1.0
version: 6.4.4.1
synopsis: SimpleXMQ message broker
description: This package includes <./docs/Simplex-Messaging-Server.html server>,
<./docs/Simplex-Messaging-Client.html client> and
@@ -224,6 +224,7 @@ library
Simplex.FileTransfer.Server.Control
Simplex.FileTransfer.Server.Env
Simplex.FileTransfer.Server.Main
Simplex.FileTransfer.Server.Prometheus
Simplex.FileTransfer.Server.Stats
Simplex.FileTransfer.Server.Store
Simplex.FileTransfer.Server.StoreLog
@@ -233,6 +234,7 @@ library
Simplex.Messaging.Server.Env.STM
Simplex.Messaging.Server.Information
Simplex.Messaging.Server.Main
Simplex.Messaging.Server.Main.GitCommit
Simplex.Messaging.Server.Main.Init
Simplex.Messaging.Server.MsgStore
Simplex.Messaging.Server.MsgStore.Journal
@@ -353,10 +355,12 @@ library
if impl(ghc >= 9.6.2)
build-depends:
bytestring ==0.11.*
, template-haskell ==2.20.*
, text >=2.0.1 && <2.2
if impl(ghc < 9.6.2)
build-depends:
bytestring ==0.10.*
, template-haskell ==2.16.*
, text >=1.2.3.0 && <1.3
executable ntf-server
+8 -4
View File
@@ -31,6 +31,7 @@ import qualified Data.X509 as X
import qualified Data.X509.Validation as XV
import qualified Network.HTTP.Types as N
import qualified Network.HTTP2.Client as H
import Network.Socket (HostName)
import Simplex.FileTransfer.Chunks
import Simplex.FileTransfer.Protocol
import Simplex.FileTransfer.Transport
@@ -45,6 +46,7 @@ import Simplex.Messaging.Client
transportClientConfig,
clientSocksCredentials,
unexpectedResponse,
useWebPort,
)
import qualified Simplex.Messaging.Crypto as C
import qualified Simplex.Messaging.Crypto.Lazy as LC
@@ -104,12 +106,13 @@ defaultXFTPClientConfig =
clientALPN = Just alpnSupportedXFTPhandshakes
}
getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> UTCTime -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, xftpNetworkConfig, serverVRange} proxySessTs disconnected = runExceptT $ do
getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> [HostName] -> UTCTime -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, xftpNetworkConfig, serverVRange} presetDomains proxySessTs disconnected = runExceptT $ do
let socksCreds = clientSocksCredentials xftpNetworkConfig proxySessTs transportSession
ProtocolServer _ host port keyHash = srv
useALPN = if useWebPort xftpNetworkConfig presetDomains srv then Just [httpALPN11] else clientALPN
useHost <- liftEither $ chooseTransportHost xftpNetworkConfig host
let tcConfig = transportClientConfig xftpNetworkConfig NRMBackground useHost False clientALPN
let tcConfig = transportClientConfig xftpNetworkConfig NRMBackground useHost False useALPN
http2Config = xftpHTTP2Config tcConfig config
clientVar <- newTVarIO Nothing
let usePort = if null port then "443" else port
@@ -121,7 +124,8 @@ getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN,
thParams0 = THandleParams {sessionId, blockSize = xftpBlockSize, thVersion = v, thServerVRange, thAuth = Nothing, implySessId = False, encryptBlock = Nothing, batch = True, serviceAuth = False}
logDebug $ "Client negotiated handshake protocol: " <> tshow sessionALPN
thParams@THandleParams {thVersion} <- case sessionALPN of
Just "xftp/1" -> xftpClientHandshakeV1 serverVRange keyHash http2Client thParams0
Just alpn | alpn == xftpALPNv1 || alpn == httpALPN11 ->
xftpClientHandshakeV1 serverVRange keyHash http2Client thParams0
_ -> pure thParams0
logDebug $ "Client negotiated protocol: " <> tshow thVersion
let c = XFTPClient {http2Client, thParams, transportSession, config}
+1 -1
View File
@@ -71,7 +71,7 @@ getXFTPServerClient XFTPClientAgent {xftpClients, startedAt, config} srv = do
connectClient =
ExceptT $
first (XFTPClientAgentError srv)
<$> getXFTPClient (1, srv, Nothing) (xftpConfig config) startedAt clientDisconnected
<$> getXFTPClient (1, srv, Nothing) (xftpConfig config) [] startedAt clientDisconnected
clientDisconnected :: XFTPClient -> IO ()
clientDisconnected _ = do
+35 -2
View File
@@ -45,6 +45,7 @@ import Network.Socket
import Simplex.FileTransfer.Protocol
import Simplex.FileTransfer.Server.Control
import Simplex.FileTransfer.Server.Env
import Simplex.FileTransfer.Server.Prometheus
import Simplex.FileTransfer.Server.Stats
import Simplex.FileTransfer.Server.Store
import Simplex.FileTransfer.Server.StoreLog
@@ -69,6 +70,7 @@ import Simplex.Messaging.Transport.HTTP2.Server
import Simplex.Messaging.Transport.Server (runLocalTCPServer)
import Simplex.Messaging.Util
import Simplex.Messaging.Version
import System.Environment (lookupEnv)
import System.Exit (exitFailure)
import System.FilePath ((</>))
import System.IO (hPrint, hPutStrLn, universalNewlineMode)
@@ -105,7 +107,14 @@ xftpServer :: XFTPServerConfig -> TMVar Bool -> M ()
xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpiration, fileExpiration, xftpServerVRange} started = do
mapM_ (expireServerFiles Nothing) fileExpiration
restoreServerStats
raceAny_ (runServer : expireFilesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg) `finally` stopServer
raceAny_
( runServer
: expireFilesThread_ cfg
<> serverStatsThread_ cfg
<> prometheusMetricsThread_ cfg
<> controlPortThread_ cfg
)
`finally` stopServer
where
runServer :: M ()
runServer = do
@@ -124,7 +133,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
req0 = XFTPTransportRequest {thParams = thParams0, request = r, reqBody, sendResponse}
flip runReaderT env $ case sessionALPN of
Nothing -> processRequest req0
Just "xftp/1" ->
Just alpn | alpn == xftpALPNv1 || alpn == httpALPN11 ->
xftpServerHandshakeV1 chain signKey sessions req0 >>= \case
Nothing -> pure () -- handshake response sent
Just thParams -> processRequest req0 {thParams} -- proceed with new version (XXX: may as well switch the request handler here)
@@ -240,6 +249,30 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
]
liftIO $ threadDelay' interval
prometheusMetricsThread_ :: XFTPServerConfig -> [M ()]
prometheusMetricsThread_ XFTPServerConfig {prometheusInterval = Just interval, prometheusMetricsFile} =
[savePrometheusMetrics interval prometheusMetricsFile]
prometheusMetricsThread_ _ = []
savePrometheusMetrics :: Int -> FilePath -> M ()
savePrometheusMetrics saveInterval metricsFile = do
labelMyThread "savePrometheusMetrics"
liftIO $ putStrLn $ "Prometheus metrics saved every " <> show saveInterval <> " seconds to " <> metricsFile
ss <- asks serverStats
rtsOpts <- liftIO $ maybe ("set " <> rtsOptionsEnv) T.pack <$> lookupEnv (T.unpack rtsOptionsEnv)
let interval = 1000000 * saveInterval
liftIO $ forever $ do
threadDelay interval
ts <- getCurrentTime
sm <- getFileServerMetrics ss rtsOpts
T.writeFile metricsFile $ xftpPrometheusMetrics sm ts
getFileServerMetrics :: FileServerStats -> T.Text -> IO FileServerMetrics
getFileServerMetrics ss rtsOptions = do
d <- getFileServerStatsData ss
let fd = periodStatDataCounts $ _filesDownloaded d
pure FileServerMetrics {statsData = d, filesDownloadedPeriods = fd, rtsOptions}
controlPortThread_ :: XFTPServerConfig -> [M ()]
controlPortThread_ XFTPServerConfig {controlPort = Just port} = [runCPServer port]
controlPortThread_ _ = []
+2
View File
@@ -64,6 +64,8 @@ data XFTPServerConfig = XFTPServerConfig
logStatsStartTime :: Int64,
serverStatsLogFile :: FilePath,
serverStatsBackupFile :: Maybe FilePath,
prometheusInterval :: Maybe Int,
prometheusMetricsFile :: FilePath,
transportConfig :: TransportServerConfig,
responseDelay :: Int
}
+9 -4
View File
@@ -27,10 +27,10 @@ import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern XFTPServer)
import Simplex.Messaging.Server.CLI
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Transport (simplexMQVersion)
import Simplex.Messaging.Transport.Client (TransportHost (..))
import Simplex.Messaging.Transport.HTTP2 (httpALPN)
import Simplex.Messaging.Transport.Server (ServerCredentials (..), mkTransportServerConfig)
import Simplex.Messaging.Util (safeDecodeUtf8, tshow)
import Simplex.Messaging.Util (eitherToMaybe, safeDecodeUtf8, tshow)
import System.Directory (createDirectoryIfMissing, doesFileExist)
import System.FilePath (combine)
import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
@@ -60,7 +60,7 @@ xftpServerCLI cfgPath logPath = do
putStrLn "Deleted configuration and log files"
where
iniFile = combine cfgPath "file-server.ini"
serverVersion = "SimpleX XFTP server v" <> simplexMQVersion
serverVersion = "SimpleX XFTP server v" <> simplexmqVersionCommit
defaultServerPort = "443"
executableName = "file-server"
storeLogFilePath = combine logPath "file-server-store.log"
@@ -89,6 +89,9 @@ xftpServerCLI cfgPath logPath = do
<> "# Expire files after the specified number of hours.\n"
<> ("expire_files_hours: " <> tshow defFileExpirationHours <> "\n\n")
<> "log_stats: off\n\
\\n\
\# Log interval for real-time Prometheus metrics\n\
\# prometheus_interval: 60\n\
\\n\
\[AUTH]\n\
\# Set new_files option to off to completely prohibit uploading new files.\n\
@@ -188,10 +191,12 @@ xftpServerCLI cfgPath logPath = do
logStatsStartTime = 0, -- seconds from 00:00 UTC
serverStatsLogFile = combine logPath "file-server-stats.daily.log",
serverStatsBackupFile = logStats $> combine logPath "file-server-stats.log",
prometheusInterval = eitherToMaybe $ read . T.unpack <$> lookupValue "STORE_LOG" "prometheus_interval" ini,
prometheusMetricsFile = combine logPath "xftp-server-metrics.txt",
transportConfig =
mkTransportServerConfig
(fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini)
(Just alpnSupportedXFTPhandshakes)
(Just $ alpnSupportedXFTPhandshakes <> httpALPN)
False,
responseDelay = 0
}
@@ -0,0 +1,134 @@
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeApplications #-}
{-# OPTIONS_GHC -fno-warn-unrecognised-pragmas #-}
module Simplex.FileTransfer.Server.Prometheus where
import Data.Int (Int64)
import Data.Text (Text)
import qualified Data.Text as T
import Data.Time.Clock (UTCTime (..), diffUTCTime)
import Data.Time.Clock.System (systemEpochDay)
import Data.Time.Format.ISO8601 (iso8601Show)
import Simplex.FileTransfer.Server.Stats
import Simplex.Messaging.Server.Stats (PeriodStatCounts (..))
import Simplex.Messaging.Transport (simplexMQVersion)
import Simplex.Messaging.Util (tshow)
data FileServerMetrics = FileServerMetrics
{ statsData :: FileServerStatsData,
filesDownloadedPeriods :: PeriodStatCounts,
rtsOptions :: Text
}
rtsOptionsEnv :: Text
rtsOptionsEnv = "XFTP_RTS_OPTIONS"
{-# FOURMOLU_DISABLE\n#-}
xftpPrometheusMetrics :: FileServerMetrics -> UTCTime -> Text
xftpPrometheusMetrics sm ts =
time <> files <> info
where
FileServerMetrics {statsData, filesDownloadedPeriods, rtsOptions} = sm
FileServerStatsData
{ _fromTime,
_filesCreated,
_fileRecipients,
_filesUploaded,
_filesExpired,
_filesDeleted,
_filesBlocked,
_fileDownloads,
_fileDownloadAcks,
_filesCount,
_filesSize
} = statsData
time =
"# Recorded at: " <> T.pack (iso8601Show ts) <> "\n\
\# Stats from: " <> T.pack (iso8601Show _fromTime) <> "\n\
\\n"
files =
"# Files\n\
\# -----\n\
\\n\
\# HELP simplex_xftp_files_created Created files\n\
\# TYPE simplex_xftp_files_created counter\n\
\simplex_xftp_files_created " <> mshow _filesCreated <> "\n\
\# filesCreated\n\
\\n\
\# HELP simplex_xftp_files_recipients Files recipients\n\
\# TYPE simplex_xftp_files_recipients counter\n\
\simplex_xftp_files_recipients " <> mshow _fileRecipients <> "\n\
\# fileRecipients\n\
\\n\
\# HELP simplex_xftp_files_uploaded Uploaded files\n\
\# TYPE simplex_xftp_files_uploaded counter\n\
\simplex_xftp_files_uploaded " <> mshow _filesUploaded <> "\n\
\# filesUploaded\n\
\\n\
\# HELP simplex_xftp_files_expired Expired files\n\
\# TYPE simplex_xftp_files_expired counter\n\
\simplex_xftp_files_expired " <> mshow _filesExpired <> "\n\
\# filesExpired\n\
\\n\
\# HELP simplex_xftp_files_deleted Deleted files\n\
\# TYPE simplex_xftp_files_deleted counter\n\
\simplex_xftp_files_deleted " <> mshow _filesDeleted <> "\n\
\# filesDeleted\n\
\\n\
\# HELP simplex_xftp_files_blocked Blocked files\n\
\# TYPE simplex_xftp_files_blocked counter\n\
\simplex_xftp_files_blocked " <> mshow _filesBlocked <> "\n\
\# filesBlocked\n\
\\n\
\# HELP simplex_xftp_file_downloads File downloads\n\
\# TYPE simplex_xftp_file_downloads counter\n\
\simplex_xftp_file_downloads " <> mshow _fileDownloads <> "\n\
\# fileDownloads\n\
\\n\
\# HELP simplex_xftp_file_download_acks File download ACKs\n\
\# TYPE simplex_xftp_file_download_acks counter\n\
\simplex_xftp_file_download_acks " <> mshow _fileDownloadAcks <> "\n\
\# fileDownloadAcks\n\
\\n\
\# HELP simplex_xftp_files_count_total Total files count \n\
\# TYPE simplex_xftp_files_count_total gauge\n\
\simplex_xftp_files_count_total " <> mshow _filesCount <> "\n\
\# filesCount\n\
\\n\
\# HELP simplex_xftp_files_size Size of files \n\
\# TYPE simplex_xftp_files_size gauge\n\
\simplex_xftp_files_size " <> mshow _filesSize <> "\n\
\# filesSize \n\
\\n\
\# HELP simplex_xftp_files_count_daily Daily files count\n\
\# TYPE simplex_xftp_files_count_daily gauge\n\
\simplex_xftp_files_count_daily " <> mstr (dayCount filesDownloadedPeriods) <> "\n\
\# filesDownloaded.dayCount\n\
\\n\
\# HELP simplex_xftp_files_count_weekly Weekly files count\n\
\# TYPE simplex_xftp_files_count_weekly gauge\n\
\simplex_xftp_files_count_weekly " <> mstr (weekCount filesDownloadedPeriods) <> "\n\
\# filesDownloaded.weekCount\n\
\\n\
\# HELP simplex_xftp_files_count_monthly Monthly files count\n\
\# TYPE simplex_xftp_files_count_monthly gauge\n\
\simplex_xftp_files_count_monthly " <> mstr (monthCount filesDownloadedPeriods) <> "\n\
\# filesDownloaded.monthCount\n\
\\n"
info =
"# Info\n\
\# ----\n\
\\n\
\# HELP simplex_xftp_info Server information. RTS options have to be passed via " <> rtsOptionsEnv <> " env var\n\
\# TYPE simplex_xftp_info gauge\n\
\simplex_xftp_info{version=\"" <> T.pack simplexMQVersion <> "\",rts_options=\"" <> rtsOptions <> "\"} 1\n\
\\n"
mstr a = a <> " " <> tsEpoch
mshow :: Show a => a -> Text
mshow = mstr . tshow
tsEpoch = tshow @Int64 $ floor @Double $ realToFrac (ts `diffUTCTime` epoch) * 1000
epoch = UTCTime systemEpochDay 0
{-# FOURMOLU_ENABLE\n#-}
+5 -1
View File
@@ -14,6 +14,7 @@ module Simplex.FileTransfer.Transport
blockedFilesXFTPVersion,
xftpClientHandshakeStub,
alpnSupportedXFTPhandshakes,
xftpALPNv1,
XFTPClientHandshake (..),
-- xftpClientHandshake,
XFTPServerHandshake (..),
@@ -105,7 +106,10 @@ xftpClientHandshakeStub :: c 'TClient -> Maybe C.KeyPairX25519 -> C.KeyHash -> V
xftpClientHandshakeStub _c _ks _keyHash _xftpVRange _proxyServer _serviceKeys = throwE TEVersion
alpnSupportedXFTPhandshakes :: [ALPN]
alpnSupportedXFTPhandshakes = ["xftp/1"]
alpnSupportedXFTPhandshakes = [xftpALPNv1]
xftpALPNv1 :: ALPN
xftpALPNv1 = "xftp/1"
data XFTPServerHandshake = XFTPServerHandshake
{ xftpVersionRange :: VersionRangeXFTP,
+9 -2
View File
@@ -912,12 +912,19 @@ getConnShortLink' c nm userId = \case
where
decryptData :: ConnectionModeI c => SMPServer -> LinkKey -> C.SbKey -> (SMP.SenderId, QueueLinkData) -> AM (ConnectionRequestUri c, ConnLinkData c)
decryptData srv linkKey k (sndId, d) = do
r@(cReq, _) <- liftEither $ SL.decryptLinkData @c linkKey k d
r@(cReq, clData) <- liftEither $ SL.decryptLinkData @c linkKey k d
let (srv', sndId') = qAddress (connReqQueue cReq)
unless (srv `sameSrvHost` srv' && sndId == sndId') $
throwE $ AGENT $ A_LINK "different address"
pure r
pure $ if srv' == srv then r else (updateConnReqServer srv cReq, clData)
sameSrvHost ProtocolServer {host = h :| _} ProtocolServer {host = hs} = h `elem` hs
updateConnReqServer :: SMPServer -> ConnectionRequestUri c -> ConnectionRequestUri c
updateConnReqServer srv = \case
CRInvitationUri crData e2eParams -> CRInvitationUri (updateQueues crData) e2eParams
CRContactUri crData -> CRContactUri $ updateQueues crData
where
updateQueues crData@(ConnReqUriData {crSmpQueues = SMPQueueUri vr addr :| qs}) =
crData {crSmpQueues = SMPQueueUri vr addr {smpServer = srv} :| qs}
deleteLocalInvShortLink' :: AgentClient -> ConnShortLink 'CMInvitation -> AM ()
deleteLocalInvShortLink' c (CSLInvitation _ srv linkId _) = withStore' c $ \db -> deleteInvShortLink db srv linkId
+14 -14
View File
@@ -330,7 +330,7 @@ data AgentClient = AgentClient
xftpServers :: TMap UserId (UserServers 'PXFTP),
xftpClients :: TMap XFTPTransportSession XFTPClientVar,
useNetworkConfig :: TVar (NetworkConfig, NetworkConfig), -- (slow, fast) networks
presetSMPDomains :: [HostName],
presetDomains :: [HostName],
userNetworkInfo :: TVar UserNetworkInfo,
userNetworkUpdated :: TVar (Maybe UTCTime),
subscrConns :: TVar (Set ConnId),
@@ -537,7 +537,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, presetDomai
xftpServers,
xftpClients,
useNetworkConfig,
presetSMPDomains = presetDomains,
presetDomains,
userNetworkInfo,
userNetworkUpdated,
subscrConns,
@@ -686,7 +686,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq
Nothing -> Left $ BROKER (B.unpack $ strEncode srv) TIMEOUT
smpConnectClient :: AgentClient -> NetworkRequestMode -> SMPTransportSession -> TMap SMPServer ProxiedRelayVar -> SMPClientVar -> AM SMPConnectedClient
smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs} nm tSess@(_, srv, _) prs v =
smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm tSess@(_, srv, _) prs v =
newProtocolClient c tSess smpClients connectClient v
`catchAgentError` \e -> lift (resubscribeSMPSession c tSess) >> throwE e
where
@@ -697,7 +697,7 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs} nm tSess@(_, srv,
env <- ask
liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ do
ts <- readTVarIO proxySessTs
smp <- ExceptT $ getProtocolClient g nm tSess cfg (presetSMPDomains c) (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
smp <- ExceptT $ getProtocolClient g nm tSess cfg presetDomains (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
pure SMPConnectedClient {connectedClient = smp, proxiedRelays = prs}
smpClientDisconnected :: AgentClient -> SMPTransportSession -> Env -> SMPClientVar -> TMap SMPServer ProxiedRelayVar -> SMPClient -> IO ()
@@ -786,7 +786,7 @@ reconnectSMPClient c tSess@(_, srv, _) qs = handleNotify $ do
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, AEvt (sAEntity @e) cmd)
getNtfServerClient :: AgentClient -> NetworkRequestMode -> NtfTransportSession -> AM NtfClient
getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs} nm tSess@(_, srv, _) = do
getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs, presetDomains} nm tSess@(_, srv, _) = do
unlessM (readTVarIO active) $ throwE INACTIVE
ts <- liftIO getCurrentTime
atomically (getSessVar workerSeq tSess ntfClients ts)
@@ -800,7 +800,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs} nm
g <- asks random
ts <- readTVarIO proxySessTs
liftError' (protocolClientError NTF $ B.unpack $ strEncode srv) $
getProtocolClient g nm tSess cfg [] Nothing ts $
getProtocolClient g nm tSess cfg presetDomains Nothing ts $
clientDisconnected v
clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
@@ -810,7 +810,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs} nm
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs} tSess@(_, srv, _) = do
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs, presetDomains} tSess@(_, srv, _) = do
unlessM (readTVarIO active) $ throwE INACTIVE
ts <- liftIO getCurrentTime
atomically (getSessVar workerSeq tSess xftpClients ts)
@@ -824,7 +824,7 @@ getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs}
xftpNetworkConfig <- getNetworkConfig c
ts <- readTVarIO proxySessTs
liftError' (protocolClientError XFTP $ B.unpack $ strEncode srv) $
X.getXFTPClient tSess cfg {xftpNetworkConfig} ts $
X.getXFTPClient tSess cfg {xftpNetworkConfig} presetDomains ts $
clientDisconnected v
clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
@@ -1227,7 +1227,7 @@ data ProtocolTestFailure = ProtocolTestFailure
deriving (Eq, Show)
runSMPServerTest :: AgentClient -> NetworkRequestMode -> UserId -> SMPServerWithAuth -> AM' (Maybe ProtocolTestFailure)
runSMPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
runSMPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth srv auth) = do
cfg <- getClientConfig c smpCfg
C.AuthAlg ra <- asks $ rcvAuthAlg . config
C.AuthAlg sa <- asks $ sndAuthAlg . config
@@ -1235,7 +1235,7 @@ runSMPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
liftIO $ do
let tSess = (userId, srv, Nothing)
ts <- readTVarIO $ proxySessTs c
getProtocolClient g nm tSess cfg (presetSMPDomains c) Nothing ts (\_ -> pure ()) >>= \case
getProtocolClient g nm tSess cfg presetDomains Nothing ts (\_ -> pure ()) >>= \case
Right smp -> do
rKeys@(_, rpKey) <- atomically $ C.generateAuthKeyPair ra g
(sKey, spKey) <- atomically $ C.generateAuthKeyPair sa g
@@ -1256,7 +1256,7 @@ runSMPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
testErr step = ProtocolTestFailure step . protocolClientError SMP addr
runXFTPServerTest :: AgentClient -> NetworkRequestMode -> UserId -> XFTPServerWithAuth -> AM' (Maybe ProtocolTestFailure)
runXFTPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
runXFTPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth srv auth) = do
cfg <- asks $ xftpCfg . config
g <- asks random
xftpNetworkConfig <- getNetworkConfig c
@@ -1266,7 +1266,7 @@ runXFTPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
liftIO $ do
let tSess = (userId, srv, Nothing)
ts <- readTVarIO $ proxySessTs c
X.getXFTPClient tSess cfg {xftpNetworkConfig} ts (\_ -> pure ()) >>= \case
X.getXFTPClient tSess cfg {xftpNetworkConfig} presetDomains ts (\_ -> pure ()) >>= \case
Right xftp -> withTestChunk filePath $ do
(sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
(rcvKey, rpKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -1304,14 +1304,14 @@ runXFTPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
createTestChunk fp = B.writeFile fp =<< atomically . C.randomBytes chSize =<< C.newRandom
runNTFServerTest :: AgentClient -> NetworkRequestMode -> UserId -> NtfServerWithAuth -> AM' (Maybe ProtocolTestFailure)
runNTFServerTest c nm userId (ProtoServerWithAuth srv _) = do
runNTFServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth srv _) = do
cfg <- getClientConfig c ntfCfg
C.AuthAlg a <- asks $ rcvAuthAlg . config
g <- asks random
liftIO $ do
let tSess = (userId, srv, Nothing)
ts <- readTVarIO $ proxySessTs c
getProtocolClient g nm tSess cfg [] Nothing ts (\_ -> pure ()) >>= \case
getProtocolClient g nm tSess cfg presetDomains Nothing ts (\_ -> pure ()) >>= \case
Right ntf -> do
(nKey, npKey) <- atomically $ C.generateAuthKeyPair a g
(dhKey, _) <- atomically $ C.generateKeyPair g
+4 -2
View File
@@ -1129,7 +1129,7 @@ instance StrEncoding AConnectionRequestUri where
connReqUriP :: Maybe ServiceScheme -> Parser AConnectionRequestUri
connReqUriP overrideScheme = do
crScheme <- (`fromMaybe` overrideScheme) <$> strP
crScheme <- (`fromMaybe` overrideScheme) <$> strP -- always parse, but use the passed one if any
crMode <- A.char '/' *> crModeP <* optional (A.char '/') <* "#/?"
query <- strP
aVRange <- queryParam "v" query
@@ -1445,7 +1445,7 @@ instance ConnectionModeI c => ToField (ConnShortLink c) where toField = toField
instance (Typeable c, ConnectionModeI c) => FromField (ConnShortLink c) where fromField = blobFieldDecoder strDecode
data ContactConnType = CCTContact | CCTChannel | CCTGroup deriving (Eq, Show)
data ContactConnType = CCTContact | CCTChannel | CCTGroup | CCTRelay deriving (Eq, Show)
data AConnShortLink = forall m. ConnectionModeI m => ACSL (SConnectionMode m) (ConnShortLink m)
@@ -1593,6 +1593,7 @@ ctTypeP = \case
'A' -> pure CCTContact
'C' -> pure CCTChannel
'G' -> pure CCTGroup
'R' -> pure CCTRelay
_ -> fail "unknown contact address type"
{-# INLINE ctTypeP #-}
@@ -1601,6 +1602,7 @@ ctTypeChar = \case
CCTContact -> 'A'
CCTChannel -> 'C'
CCTGroup -> 'G'
CCTRelay -> 'R'
{-# INLINE ctTypeChar #-}
-- the servers passed to this function should be all preset servers, not servers configured by the user.
+1 -1
View File
@@ -672,7 +672,7 @@ data StoreError
| -- | Invitation not found
SEInvitationNotFound String InvitationId
| -- | Message not found
SEMsgNotFound
SEMsgNotFound String
| -- | Command not found
SECmdNotFound
| -- | Currently not used. The intention was to pass current expected queue status in methods,
@@ -891,7 +891,7 @@ createSndMsgDelivery db connId SndQueue {dbQueueId} msgId =
getSndMsgViaRcpt :: DB.Connection -> ConnId -> InternalSndId -> IO (Either StoreError SndMsg)
getSndMsgViaRcpt db connId sndMsgId =
firstRow toSndMsg SEMsgNotFound $
firstRow toSndMsg (SEMsgNotFound "getSndMsgViaRcpt") $
DB.query
db
[sql|
@@ -1027,7 +1027,7 @@ getExpiredSndMessages db connId SndQueue {dbQueueId} expireTs = do
setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (RcvQueue, SMP.MsgId))
setMsgUserAck db connId agentMsgId = runExceptT $ do
(dbRcvId, srvMsgId) <-
ExceptT . firstRow id SEMsgNotFound $
ExceptT . firstRow id (SEMsgNotFound "setMsgUserAck") $
DB.query db "SELECT rcv_queue_id, broker_id FROM rcv_messages WHERE conn_id = ? AND internal_id = ?" (connId, agentMsgId)
rq <- ExceptT $ getRcvQueueById db connId dbRcvId
liftIO $ DB.execute db "UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id = ?" (BI True, connId, agentMsgId)
@@ -1035,7 +1035,7 @@ setMsgUserAck db connId agentMsgId = runExceptT $ do
getRcvMsg :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError RcvMsg)
getRcvMsg db connId agentMsgId =
firstRow toRcvMsg SEMsgNotFound $
firstRow toRcvMsg (SEMsgNotFound "getRcvMsg") $
DB.query
db
[sql|
@@ -1085,7 +1085,7 @@ checkRcvMsgHashExists db connId hash = do
getRcvMsgBrokerTs :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Either StoreError BrokerTs)
getRcvMsgBrokerTs db connId msgId =
firstRow fromOnly SEMsgNotFound $
firstRow fromOnly (SEMsgNotFound "getRcvMsgBrokerTs") $
DB.query db "SELECT broker_ts FROM rcv_messages WHERE conn_id = ? AND broker_id = ?" (connId, Binary msgId)
deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
+18 -10
View File
@@ -38,6 +38,7 @@ module Simplex.Messaging.Client
protocolClientServer',
transportHost',
transportSession',
useWebPort,
-- * SMP protocol command functions
createSMPQueue,
@@ -160,6 +161,7 @@ import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Client (SocksAuth (..), SocksProxyWithAuth (..), TransportClientConfig (..), TransportHost (..), defaultSMPPort, runTransportClient)
import Simplex.Messaging.Transport.HTTP2 (httpALPN11)
import Simplex.Messaging.Transport.KeepAlive
import Simplex.Messaging.Util
import Simplex.Messaging.Version
@@ -560,7 +562,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
`catch` \(e :: IOException) -> pure . Left $ PCEIOError e
Left e -> pure $ Left e
where
NetworkConfig {smpWebPortServers, tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
NetworkConfig {tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
mkProtocolClient :: TransportHost -> UTCTime -> IO (PClient v err msg)
mkProtocolClient transportHost ts = do
connected <- newTVarIO False
@@ -591,7 +593,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
runClient :: (ServiceName, ATransport 'TClient) -> TransportHost -> PClient v err msg -> IO (Either (ProtocolClientError err) (ProtocolClient v err msg))
runClient (port', ATransport t) useHost c = do
cVar <- newEmptyTMVarIO
let tcConfig = (transportClientConfig networkConfig nm useHost useSNI clientALPN) {clientCredentials = serviceCreds <$> serviceCredentials}
let tcConfig = (transportClientConfig networkConfig nm useHost useSNI useALPN) {clientCredentials = serviceCreds <$> serviceCredentials}
socksCreds = clientSocksCredentials networkConfig proxySessTs transportSession
tId <-
runTransportClient tcConfig socksCreds useHost port' (Just $ keyHash srv) (client t c cVar)
@@ -605,16 +607,14 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
useTransport :: (ServiceName, ATransport 'TClient)
useTransport = case port srv of
"" -> case protocolTypeI @(ProtoType msg) of
SPSMP | smpWebPort -> ("443", transport @TLS)
SPSMP | web -> ("443", transport @TLS)
_ -> defaultTransport cfg
p -> (p, transport @TLS)
where
smpWebPort = case smpWebPortServers of
SWPAll -> True
SWPPreset -> case srv of
ProtocolServer {host = THDomainName h :| _} -> any (`isSuffixOf` h) presetDomains
_ -> False
SWPOff -> False
useALPN :: Maybe [ALPN]
useALPN = if web then Just [httpALPN11] else clientALPN
web = useWebPort networkConfig presetDomains srv
client :: forall c. Transport c => TProxy c 'TClient -> PClient v err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient v err msg)) -> c 'TClient -> IO ()
client _ c cVar h = do
@@ -709,6 +709,14 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
Left e -> logError $ "SMP client error: " <> tshow e
Right _ -> logWarn "SMP client unprocessed event"
useWebPort :: NetworkConfig -> [HostName] -> ProtocolServer p -> Bool
useWebPort cfg presetDomains srv = case smpWebPortServers cfg of
SWPAll -> True
SWPPreset -> case srv of
ProtocolServer {host = THDomainName h :| _} -> any (`isSuffixOf` h) presetDomains
_ -> False
SWPOff -> False
unexpectedResponse :: Show r => r -> ProtocolClientError err
unexpectedResponse = PCEUnexpectedResponse . B.pack . take 32 . show
@@ -46,8 +46,9 @@ import Simplex.Messaging.Server.Main (strParse)
import Simplex.Messaging.Server.Main.Init (iniDbOpts)
import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
import Simplex.Messaging.Server.StoreLog (closeStoreLog)
import Simplex.Messaging.Transport (ASrvTransport, simplexMQVersion)
import Simplex.Messaging.Transport (ASrvTransport)
import Simplex.Messaging.Transport.Client (TransportHost (..))
import Simplex.Messaging.Transport.HTTP2 (httpALPN)
import Simplex.Messaging.Transport.Server (AddHTTP, ServerCredentials (..), mkTransportServerConfig)
import Simplex.Messaging.Util (eitherToMaybe, ifM, tshow)
import System.Directory (createDirectoryIfMissing, doesFileExist, renameFile)
@@ -136,7 +137,7 @@ ntfServerCLI cfgPath logPath =
(putStrLn ("Store log file " <> storeLogFile <> " not found") >> exitFailure)
Nothing -> putStrLn "Store log disabled, see `[STORE_LOG] enable`" >> exitFailure
iniFile = combine cfgPath "ntf-server.ini"
serverVersion = "SMP notifications server v" <> simplexMQVersion
serverVersion = "SMP notifications server v" <> simplexmqVersionCommit
defaultServerPort = "443"
executableName = "ntf-server"
storeLogFilePath = combine logPath "ntf-server-store.log"
@@ -167,6 +168,9 @@ ntfServerCLI cfgPath logPath =
<> "Time to retain deleted entities in the database, days.\n"
<> ("# db_deleted_ttl: " <> tshow defaultDeletedTTL <> "\n\n")
<> "log_stats: off\n\n\
\# Log interval for real-time Prometheus metrics\n\
\# prometheus_interval: 60\n\
\\n\
\[AUTH]\n\
\# control_port_admin_password:\n\
\# control_port_user_password:\n\
@@ -280,7 +284,7 @@ ntfServerCLI cfgPath logPath =
transportConfig =
mkTransportServerConfig
(fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini)
(Just alpnSupportedNTFHandshakes)
(Just $ alpnSupportedNTFHandshakes <> httpALPN)
False,
startOptions
}
+27 -19
View File
@@ -203,15 +203,12 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
env <- ask
liftIO $ case (httpCreds_, attachHTTP_) of
(Just httpCreds, Just attachHTTP) | addHTTP ->
runTransportServerState_ ss started tcpPort defaultSupportedParamsHTTPS combinedCreds tCfg {serverALPN = Just combinedALPNs} $ \s (sniUsed, h) ->
runTransportServerState_ ss started tcpPort defaultSupportedParamsHTTPS combinedCreds tCfg $ \s (sniUsed, h) ->
case cast h of
Just (TLS {tlsContext} :: TLS 'TServer) | sniUsed -> labelMyThread "https client" >> attachHTTP s tlsContext
_ -> runClient srvCert srvSignKey t h `runReaderT` env
where
combinedCreds = TLSServerCredential {credential = smpCreds, sniCredential = Just httpCreds}
combinedALPNs = alpnSupportedSMPHandshakes <> httpALPN
httpALPN :: [ALPN]
httpALPN = ["h2", "http/1.1"]
_ ->
runTransportServerState ss started tcpPort defaultSupportedParams smpCreds tCfg $ \h -> runClient srvCert srvSignKey t h `runReaderT` env
@@ -963,24 +960,24 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
SubPending -> (c1, c2 + 1, c3, c4)
SubThread _ -> (c1, c2, c3 + 1, c4)
ProhibitSub -> pure (c1, c2, c3, c4 + 1)
CPDelete sId -> withAdminRole $ unliftIO u $ do
CPDelete qId -> withAdminRole $ unliftIO u $ do
st <- asks msgStore
r <- liftIO $ runExceptT $ do
q <- ExceptT $ getQueue st SSender sId
(q, _) <- ExceptT $ getSenderQueue st qId
ExceptT $ deleteQueueSize st q
case r of
Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
Right (qr, numDeleted) -> do
updateDeletedStats qr
liftIO $ hPutStrLn h $ "ok, " <> show numDeleted <> " messages deleted"
CPStatus sId -> withUserRole $ unliftIO u $ do
CPStatus qId -> withUserRole $ unliftIO u $ do
st <- asks msgStore
q <- liftIO $ getQueueRec st SSender sId
q <- liftIO $ getSenderQueue st qId
liftIO $ hPutStrLn h $ case q of
Left e -> "error: " <> show e
Right (_, QueueRec {queueMode, status, updatedAt}) ->
"status: " <> show status <> ", updatedAt: " <> show updatedAt <> ", queueMode: " <> show queueMode
CPBlock sId info -> withUserRole $ unliftIO u $ do
CPBlock qId info -> withUserRole $ unliftIO u $ do
st <- asks msgStore
stats <- asks serverStats
blocked <- liftIO $ readIORef $ qBlocked stats
@@ -989,21 +986,27 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
then liftIO $ hPutStrLn h $ "error: reached limit of " <> show quota <> " queues blocked daily"
else do
r <- liftIO $ runExceptT $ do
q <- ExceptT $ getQueue st SSender sId
ExceptT $ blockQueue (queueStore st) q info
(q, QueueRec {status}) <- ExceptT $ getSenderQueue st qId
when (status == EntityActive) $ ExceptT $ blockQueue (queueStore st) q info
pure status
case r of
Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
Right () -> do
Right EntityActive -> do
incStat $ qBlocked stats
liftIO $ hPutStrLn h "ok"
CPUnblock sId -> withUserRole $ unliftIO u $ do
liftIO $ hPutStrLn h "ok, queue blocked"
Right status -> liftIO $ hPutStrLn h $ "ok, already inactive: " <> show status
CPUnblock qId -> withUserRole $ unliftIO u $ do
st <- asks msgStore
r <- liftIO $ runExceptT $ do
q <- ExceptT $ getQueue st SSender sId
ExceptT $ unblockQueue (queueStore st) q
(q, QueueRec {status}) <- ExceptT $ getSenderQueue st qId
case status of
EntityBlocked info -> Right info <$ ExceptT (unblockQueue (queueStore st) q)
EntityActive -> pure $ Left True
EntityOff -> pure $ Left False
liftIO $ hPutStrLn h $ case r of
Left e -> "error: " <> show e
Right () -> "ok"
Right (Right info) -> "ok, queue unblocked, reason to block was: " <> show info
Right (Left unblocked) -> if unblocked then "ok, queue was active" else "error, queue is inactive"
CPSave -> withAdminRole $ withLock' (savingLock srv) "control" $ do
hPutStrLn h "saving server state..."
unliftIO u $ saveServer False
@@ -1012,6 +1015,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
CPQuit -> pure ()
CPSkip -> pure ()
where
getSenderQueue st qId =
getQueueRec st SSender qId >>= \case
Right r -> pure $ Right r
Left AUTH -> getQueueRec st SSenderLink qId
Left e -> pure $ Left e
withUserRole action = readTVarIO role >>= \case
CPRAdmin -> action
CPRUser -> action
@@ -1506,8 +1514,8 @@ client
rcvId <- randId
ntf <- forM ntfKeys_ $ \(notifierKey, rcvNtfDhSecret, rcvPubDhKey) -> do
notifierId <- randId
let ntfCreds = NtfCreds {notifierId, notifierKey, rcvNtfDhSecret, ntfServiceId = Nothing}
pure (ntfCreds, ServerNtfCreds notifierId rcvPubDhKey)
let ntfCreds' = NtfCreds {notifierId, notifierKey, rcvNtfDhSecret, ntfServiceId = Nothing}
pure (ntfCreds', ServerNtfCreds notifierId rcvPubDhKey)
let queueMode = queueReqMode <$> queueReqData
qr =
QueueRec
+9 -1
View File
@@ -7,6 +7,7 @@
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeApplications #-}
@@ -33,8 +34,9 @@ import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), ProtocolServer (..), ProtocolTypeI)
import Simplex.Messaging.Server.Env.STM (ServerStoreCfg (..), StartOptions (..), StorePaths (..))
import Simplex.Messaging.Server.Main.GitCommit
import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), TLS, Transport (..))
import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), TLS, Transport (..), simplexMQVersion)
import Simplex.Messaging.Transport.Server (AddHTTP, loadFileFingerprint)
import Simplex.Messaging.Transport.WebSockets (WS)
import Simplex.Messaging.Util (eitherToMaybe, whenM)
@@ -97,6 +99,12 @@ getCliCommand' cmdP version =
where
versionOption = infoOption version (long "version" <> short 'v' <> help "Show version")
simplexmqVersionCommit :: String
simplexmqVersionCommit = simplexMQVersion <> " / " <> take 7 simplexmqCommit
simplexmqCommit :: String
simplexmqCommit = $(gitCommit)
createServerX509 :: FilePath -> X509Config -> IO ByteString
createServerX509 = createServerX509_ True
+2 -2
View File
@@ -14,7 +14,7 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
import Data.Int (Int64)
import Data.Maybe (isJust)
import Data.Text (Text)
import Simplex.Messaging.Agent.Protocol (ConnectionMode (..), ConnectionRequestUri)
import Simplex.Messaging.Agent.Protocol (ConnectionLink, ConnectionMode (..), ConnectionRequestUri)
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON)
@@ -100,7 +100,7 @@ data Entity = Entity {name :: Text, country :: Maybe Text}
deriving (Show)
data ServerContactAddress = ServerContactAddress
{ simplex :: Maybe (ConnectionRequestUri 'CMContact),
{ simplex :: Maybe (ConnectionLink 'CMContact),
email :: Maybe Text, -- it is recommended that it matches DNS email address, if either is present
pgp :: Maybe PGPKey
}
+8 -6
View File
@@ -37,7 +37,7 @@ import qualified Data.Text as T
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
import qualified Data.Text.IO as T
import Options.Applicative
import Simplex.Messaging.Agent.Protocol (connReqUriP')
import Simplex.Messaging.Agent.Protocol (ConnectionLink (..), connReqUriP')
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import Simplex.Messaging.Client (HostMode (..), NetworkConfig (..), ProtocolClientConfig (..), SMPWebPortServers (..), SocksMode (..), defaultNetworkConfig, textToHostMode)
@@ -56,8 +56,9 @@ import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore (..), QStoreCf
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), SQSType (..), SMSType (..), newMsgStore)
import Simplex.Messaging.Server.QueueStore.Postgres.Config
import Simplex.Messaging.Server.StoreLog.ReadWrite (readQueueStore)
import Simplex.Messaging.Transport (simplexMQVersion, supportedProxyClientSMPRelayVRange, alpnSupportedSMPHandshakes, supportedServerSMPRelayVRange)
import Simplex.Messaging.Transport (supportedProxyClientSMPRelayVRange, alpnSupportedSMPHandshakes, supportedServerSMPRelayVRange)
import Simplex.Messaging.Transport.Client (TransportHost (..), defaultSocksProxy)
import Simplex.Messaging.Transport.HTTP2 (httpALPN)
import Simplex.Messaging.Transport.Server (ServerCredentials (..), mkTransportServerConfig)
import Simplex.Messaging.Util (eitherToMaybe, ifM)
import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist)
@@ -234,7 +235,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
(putStrLn ("Store log file " <> storeLogFile <> " not found") >> exitFailure)
Nothing -> putStrLn "Store log disabled, see `[STORE_LOG] enable`" >> exitFailure
iniFile = combine cfgPath "smp-server.ini"
serverVersion = "SMP server v" <> simplexMQVersion
serverVersion = "SMP server v" <> simplexmqVersionCommit
executableName = "smp-server"
storeLogFilePath = combine logPath "smp-server-store.log"
storeMsgsFilePath = combine logPath "smp-server-messages.log"
@@ -450,7 +451,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
transportConfig =
mkTransportServerConfig
(fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini)
(Just alpnSupportedSMPHandshakes)
(Just $ alpnSupportedSMPHandshakes <> httpALPN)
(fromMaybe True $ iniOnOff "TRANSPORT" "accept_service_credentials" ini), -- TODO [certs] remove this option
controlPort = eitherToMaybe $ T.unpack <$> lookupValue "TRANSPORT" "control_port" ini,
smpAgentCfg =
@@ -638,7 +639,8 @@ serverPublicInfo ini = serverInfo <$!> infoValue "source_code"
<$!> infoValue nameField
countryValue field = (either error id . validCountryValue (T.unpack field) . T.unpack) <$!> infoValue field
iniContacts simplexField emailField pgpKeyUriField pgpKeyFingerprintField =
let simplex = either error id . parseAll (connReqUriP' Nothing) . encodeUtf8 <$!> eitherToMaybe (lookupValue "INFORMATION" simplexField ini)
let simplex = either error id . parseAll linkP . encodeUtf8 <$!> eitherToMaybe (lookupValue "INFORMATION" simplexField ini)
linkP = CLFull <$> connReqUriP' Nothing <|> CLShort <$> strP
email = infoValue emailField
pkURI_ = infoValue pgpKeyUriField
pkFingerprint_ = infoValue pgpKeyFingerprintField
@@ -807,7 +809,7 @@ cliCommandP cfgPath logPath iniFile =
sourceCode = T.pack <$> sourceCode,
serverInfo =
ServerPublicInfo
{ sourceCode = T.pack simplexmqSource,
{ sourceCode = T.pack $ fromMaybe simplexmqSource sourceCode,
usageConditions = Nothing,
operator = fst operator_,
website,
@@ -0,0 +1,17 @@
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE TemplateHaskell #-}
module Simplex.Messaging.Server.Main.GitCommit where
import Language.Haskell.TH
import System.Process
import Control.Exception
import System.Exit
gitCommit :: Q Exp
gitCommit = stringE . commit =<< runIO (try $ readProcessWithExitCode "git" ["rev-parse", "HEAD"] "")
where
commit :: Either SomeException (ExitCode, String, String) -> String
commit = \case
Right (ExitSuccess, out, _) -> take 40 out
_ -> ""
+1 -1
View File
@@ -92,7 +92,7 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
<> "# Log daily server statistics to CSV file\n"
<> ("log_stats: " <> onOff logStats <> "\n\n")
<> "# Log interval for real-time Prometheus metrics\n\
\# prometheus_interval: 300\n\n\
\# prometheus_interval: 60\n\n\
\[AUTH]\n\
\# Set new_queues option to off to completely prohibit creating new messaging queues.\n\
\# This can be useful when you want to decommission the server, but not all connections are switched yet.\n\
@@ -142,7 +142,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
getEntityCounts :: PostgresQueueStore q -> IO EntityCounts
getEntityCounts st =
withConnection (dbStore st) $ \db -> do
withTransaction (dbStore st) $ \db -> do
(queueCount, notifierCount, rcvServiceCount, ntfServiceCount, rcvServiceQueuesCount, ntfServiceQueuesCount) : _ <-
DB.query
db
@@ -496,7 +496,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
batchInsertServices :: [STMService] -> PostgresQueueStore q -> IO Int64
batchInsertServices services' toStore =
withConnection (dbStore toStore) $ \db ->
withTransaction (dbStore toStore) $ \db ->
DB.executeMany db insertServiceQuery $ map (serviceRecToRow . serviceRec) services'
batchInsertQueues :: StoreQueueClass q => Bool -> M.Map RecipientId q -> PostgresQueueStore q' -> IO Int64
@@ -505,7 +505,7 @@ batchInsertQueues tty queues toStore = do
putStrLn $ "Importing " <> show (length qs) <> " queues..."
let st = dbStore toStore
count <-
withConnection st $ \db -> do
withTransaction st $ \db -> do
DB.copy_
db
[sql|
@@ -514,7 +514,7 @@ batchInsertQueues tty queues toStore = do
|]
mapM_ (putQueue db) (zip [1..] qs)
DB.putCopyEnd db
Only qCnt : _ <- withConnection st (`DB.query_` "SELECT count(*) FROM msg_queues")
Only qCnt : _ <- withTransaction st (`DB.query_` "SELECT count(*) FROM msg_queues")
putStrLn $ progress count
pure qCnt
where
@@ -541,13 +541,13 @@ insertServiceQuery =
foldServiceRecs :: forall a q. Monoid a => PostgresQueueStore q -> (ServiceRec -> IO a) -> IO a
foldServiceRecs st f =
withConnection (dbStore st) $ \db ->
withTransaction (dbStore st) $ \db ->
DB.fold_ db "SELECT service_id, service_role, service_cert, service_cert_hash, created_at FROM services" mempty $
\ !acc -> fmap (acc <>) . f . rowToServiceRec
foldQueueRecs :: forall a q. Monoid a => Bool -> Bool -> PostgresQueueStore q -> Maybe Int64 -> ((RecipientId, QueueRec) -> IO a) -> IO a
foldQueueRecs tty withData st skipOld_ f = do
(n, r) <- withConnection (dbStore st) $ \db ->
(n, r) <- withTransaction (dbStore st) $ \db ->
foldRecs db (0 :: Int, mempty) $ \(i, acc) qr -> do
r <- f qr
let !i' = i + 1
@@ -686,7 +686,7 @@ withDB' op st action = withDB op st $ fmap Right . action
withDB :: forall a q. Text -> PostgresQueueStore q -> (DB.Connection -> IO (Either ErrorType a)) -> ExceptT ErrorType IO a
withDB op st action =
ExceptT $ E.try (withConnection (dbStore st) action) >>= either logErr pure
ExceptT $ E.try (withTransaction (dbStore st) action) >>= either logErr pure
where
logErr :: E.SomeException -> IO (Either ErrorType a)
logErr e = logError ("STORE: " <> err) $> Left (STORE err)
@@ -25,14 +25,6 @@ import qualified Simplex.Messaging.Crypto as C
import qualified Time.System as Hourglass
import qualified Time.Types as HT
-- | Generate a certificate chain to be used with TLS fingerprint-pinning
--
-- @
-- genTlsCredentials = do
-- ca <- genCredentials Nothing (-25, 365 * 24) "Root" -- long-lived root cert
-- leaf <- genCredentials (Just ca) (0, 1) "Entity" -- session-signing cert
-- pure $ tlsCredentials (leaf :| [ca])
-- @
tlsCredentials :: NonEmpty Credentials -> (C.KeyHash, TLS.Credential)
tlsCredentials credentials = (C.KeyHash rootFP, (X509.CertificateChain certs, privateToTls $ snd leafKey))
where
+8 -1
View File
@@ -1,4 +1,5 @@
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
module Simplex.Messaging.Transport.HTTP2 where
@@ -15,7 +16,7 @@ import qualified Network.HTTP2.Server as HS
import Network.Socket (SockAddr (..))
import qualified Network.TLS as T
import qualified Network.TLS.Extra as TE
import Simplex.Messaging.Transport (TLS, Transport (cGet, cPut))
import Simplex.Messaging.Transport (ALPN, TLS, Transport (cGet, cPut))
import Simplex.Messaging.Transport.Buffer
import qualified System.TimeManager as TI
@@ -81,3 +82,9 @@ getHTTP2Body r n = do
-- TODO check bodySize once it is set
bodyPart = if B.length bodyHead == n then Just getPart else Nothing
pure HTTP2Body {bodyHead, bodySize, bodyPart, bodyBuffer}
httpALPN :: [ALPN]
httpALPN = ["h2", "http/1.1"]
httpALPN11 :: ALPN
httpALPN11 = "http/1.1"
+4 -3
View File
@@ -49,7 +49,7 @@ import Network.Socket
import qualified Network.TLS as T
import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Shared
import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow, unlessM)
import System.Exit (exitFailure)
import System.IO.Error (tryIOError)
import System.Mem.Weak (Weak, deRefWeak)
@@ -172,12 +172,13 @@ runTCPServerSocket (accepted, gracefullyClosed, clients) started getSocket serve
E.bracket getSocket (closeServer started clients) $ \sock ->
forever . E.bracketOnError (safeAccept sock) (close . fst) $ \(conn, _peer) -> do
cId <- atomically $ stateTVar accepted $ \cId -> let cId' = cId + 1 in cId' `seq` (cId', cId')
closed <- newTVarIO False
let closeConn _ = do
atomically $ modifyTVar' clients $ IM.delete cId
atomically $ writeTVar closed True >> modifyTVar' clients (IM.delete cId)
gracefulClose conn 5000 `catchAll_` pure () -- catchAll_ is needed here in case the connection was closed earlier
atomically $ modifyTVar' gracefullyClosed (+ 1)
tId <- mkWeakThreadId =<< server conn `forkFinally` closeConn
atomically $ modifyTVar' clients $ IM.insert cId tId
atomically $ unlessM (readTVar closed) $ modifyTVar' clients $ IM.insert cId tId
-- | Recover from errors in `accept` whenever it is safe.
-- Some errors are safe to ignore, while blindly restaring `accept` may trigger a busy loop.
+3 -3
View File
@@ -85,7 +85,7 @@ encInvitationSize = 900
newRCHostPairing :: TVar ChaChaDRG -> IO RCHostPairing
newRCHostPairing drg = do
((_, caKey), caCert) <- genCredentials drg Nothing (-25, 24 * 999999) "ca"
((_, caKey), caCert) <- genCredentials drg Nothing (25, 24 * 999999) "ca"
(_, idPrivKey) <- atomically $ C.generateKeyPair drg
pure RCHostPairing {caKey, caCert, idPrivKey, knownHost = Nothing}
@@ -193,7 +193,7 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
genTLSCredentials :: TVar ChaChaDRG -> C.APrivateSignKey -> X.SignedCertificate -> IO TLS.Credential
genTLSCredentials drg caKey caCert = do
let caCreds = (C.signatureKeyPair caKey, caCert)
leaf <- genCredentials drg (Just caCreds) (0, 24 * 999999) "localhost" -- session-signing cert
leaf <- genCredentials drg (Just caCreds) (1, 24 * 999999) "localhost" -- session-signing cert
pure . snd $ tlsCredentials (leaf :| [caCreds])
certFingerprint :: X.SignedCertificate -> C.KeyHash
@@ -259,7 +259,7 @@ connectRCCtrl drg (RCVerifiedInvitation inv@RCInvitation {ca, idkey}) pairing_ h
where
newCtrlPairing :: IO RCCtrlPairing
newCtrlPairing = do
((_, caKey), caCert) <- genCredentials drg Nothing (0, 24 * 999999) "ca"
((_, caKey), caCert) <- genCredentials drg Nothing (1, 24 * 999999) "ca"
(_, dhPrivKey) <- atomically $ C.generateKeyPair drg
pure RCCtrlPairing {caKey, caCert, ctrlFingerprint = ca, idPubKey = idkey, dhPrivKey, prevDhPrivKey = Nothing}
updateCtrlPairing :: RCCtrlPairing -> ExceptT RCErrorType IO RCCtrlPairing
+1
View File
@@ -1,4 +1,5 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-}
+5 -4
View File
@@ -23,8 +23,9 @@ import qualified Network.HTTP.Client as H1
import qualified Network.HTTP2.Client as H2
import Simplex.FileTransfer.Server.Main (xftpServerCLI)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Server.CLI (simplexmqVersionCommit)
import Simplex.Messaging.Server.Main (smpServerCLI, smpServerCLI_)
import Simplex.Messaging.Transport (TLS (..), TransportPeer (..), defaultSupportedParams, defaultSupportedParamsHTTPS, simplexMQVersion, supportedClientSMPRelayVRange)
import Simplex.Messaging.Transport (TLS (..), TransportPeer (..), defaultSupportedParams, defaultSupportedParamsHTTPS, supportedClientSMPRelayVRange)
import Simplex.Messaging.Transport.Client (TransportClientConfig (..), defaultTransportClientConfig, runTLSTransportClient, smpClientHandshake)
import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..))
import qualified Simplex.Messaging.Transport.HTTP2.Client as HC
@@ -108,7 +109,7 @@ smpServerTest storeLog basicAuth = do
doesFileExist (cfgPath <> "/ca.key") `shouldReturn` True
-- start
r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ()))
r `shouldContain` ["SMP server v" <> simplexMQVersion]
r `shouldContain` ["SMP server v" <> simplexmqVersionCommit]
r `shouldContain` (if storeLog then ["Store log: " <> logPath <> "/smp-server-store.log"] else ["Store log disabled."])
r `shouldContain` ["Serving SMP protocol on port 5223 (TLS)...", "Serving SMP protocol on port 443 (TLS)...", "Serving static site on port 443 (TLS)..."]
r `shouldContain` ["expiring clients inactive for 21600 seconds every 3600 seconds"]
@@ -216,7 +217,7 @@ ntfServerTest storeLog = do
lookupValue "TRANSPORT" "websockets" ini `shouldBe` Right "off"
doesFileExist (ntfCfgPath <> "/ca.key") `shouldReturn` True
r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` ntfServerCLI ntfCfgPath ntfLogPath) `catchAll_` pure (Just ()))
r `shouldContain` ["SMP notifications server v" <> simplexMQVersion]
r `shouldContain` ["SMP notifications server v" <> simplexmqVersionCommit]
r `shouldContain` (if storeLog then ["Store log: " <> ntfLogPath <> "/ntf-server-store.log"] else ["Store log disabled."])
r `shouldContain` ["Serving NTF protocol on port 443 (TLS)..."]
capture_ (withStdin "Y" . withArgs ["delete"] $ ntfServerCLI ntfCfgPath ntfLogPath)
@@ -234,7 +235,7 @@ xftpServerTest storeLog = do
lookupValue "TRANSPORT" "port" ini `shouldBe` Right "443"
doesFileExist (fileCfgPath <> "/ca.key") `shouldReturn` True
r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` xftpServerCLI fileCfgPath fileLogPath) `catchAll_` pure (Just ()))
r `shouldContain` ["SimpleX XFTP server v" <> simplexMQVersion]
r `shouldContain` ["SimpleX XFTP server v" <> simplexmqVersionCommit]
r `shouldContain` (if storeLog then ["Store log: " <> fileLogPath <> "/file-server-store.log"] else ["Store log disabled."])
r `shouldContain` ["Listening on port 443..."]
capture_ (withStdin "Y" . withArgs ["delete"] $ xftpServerCLI fileCfgPath fileLogPath)
+1 -1
View File
@@ -65,6 +65,6 @@ postgresSchemaDumpTest migrations skipComparisonForDownMigrations testDBOpts@DBO
void $ readCreateProcess (shell cmd) ""
threadDelay 20000
let sed = (if ci then "sed -i" else "sed -i ''")
void $ readCreateProcess (shell $ sed <> " '/^--/d' " <> schemaPath) ""
void $ readCreateProcess (shell $ sed <> " '/^--/d; /^\\\\restrict/d; /^\\\\unrestrict/d' " <> schemaPath) ""
sch <- readFile schemaPath
sch `deepseq` pure sch
+12 -7
View File
@@ -954,7 +954,7 @@ testTiming =
forM_ timingTests $ \tst ->
it (testName tst) $ \(ATransport t, msType) ->
smpTest2Cfg (cfgMS msType) (mkVersionRange minServerSMPRelayVersion authCmdsSMPVersion) t $ \rh sh ->
testSameTiming rh sh tst
testSameTiming rh sh tst msType
where
testName :: (C.AuthAlg, C.AuthAlg, Int) -> String
testName (C.AuthAlg goodKeyAlg, C.AuthAlg badKeyAlg, _) = unwords ["queue key:", show goodKeyAlg, "/ used key:", show badKeyAlg]
@@ -971,11 +971,16 @@ testTiming =
(C.AuthAlg C.SX25519, C.AuthAlg C.SX25519, 200) -- correct key type
]
timeRepeat n = fmap fst . timeItT . forM_ (replicate n ()) . const
similarTime t1 t2
| t1 <= t2 = abs (1 - t1 / t2) < 0.3 -- normally the difference between "no queue" and "wrong key" is less than 5%
| otherwise = similarTime t2 t1
testSameTiming :: forall c. Transport c => THandleSMP c 'TClient -> THandleSMP c 'TClient -> (C.AuthAlg, C.AuthAlg, Int) -> Expectation
testSameTiming rh sh (C.AuthAlg goodKeyAlg, C.AuthAlg badKeyAlg, n) = do
similarTime t1 t2 msType
| t1 <= t2 = abs (1 - t1 / t2) < diff
| otherwise = similarTime t2 t1 msType
where
-- normally the difference between "no queue" and "wrong key" is less than 5%, but it's higher on PostgreSQL and on CI
diff = case msType of
ASType SQSPostgres _ -> 0.45
_ -> 0.3
testSameTiming :: forall c. Transport c => THandleSMP c 'TClient -> THandleSMP c 'TClient -> (C.AuthAlg, C.AuthAlg, Int) -> AStoreType -> Expectation
testSameTiming rh sh (C.AuthAlg goodKeyAlg, C.AuthAlg badKeyAlg, n) msType = do
g <- C.newRandom
(rPub, rKey) <- atomically $ C.generateAuthKeyPair goodKeyAlg g
(dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -1010,7 +1015,7 @@ testTiming =
timeNoQueue <- timeRepeat n $ do
Resp "dabc" _ (ERR AUTH) <- signSendRecv h badKey ("dabc", EntityId "1234", cmd)
return ()
let ok = similarTime timeNoQueue timeWrongKey
let ok = similarTime timeNoQueue timeWrongKey msType
unless ok . putStrLn . unwords $
[ show goodKeyAlg,
show badKeyAlg,
+6 -1
View File
@@ -99,6 +99,9 @@ testXFTPLogFile = "tests/tmp/xftp-server-store.log"
testXFTPStatsBackupFile :: FilePath
testXFTPStatsBackupFile = "tests/tmp/xftp-server-stats.log"
xftpTestPrometheusMetricsFile :: FilePath
xftpTestPrometheusMetricsFile = "tests/tmp/xftp-server-metrics.txt"
testXFTPServerConfig :: XFTPServerConfig
testXFTPServerConfig =
XFTPServerConfig
@@ -127,6 +130,8 @@ testXFTPServerConfig =
logStatsStartTime = 0,
serverStatsLogFile = "tests/tmp/xftp-server-stats.daily.log",
serverStatsBackupFile = Nothing,
prometheusInterval = Nothing,
prometheusMetricsFile = xftpTestPrometheusMetricsFile,
transportConfig = mkTransportServerConfig True (Just alpnSupportedXFTPhandshakes) False,
responseDelay = 0
}
@@ -140,6 +145,6 @@ testXFTPClient = testXFTPClientWith testXFTPClientConfig
testXFTPClientWith :: HasCallStack => XFTPClientConfig -> (HasCallStack => XFTPClient -> IO a) -> IO a
testXFTPClientWith cfg client = do
ts <- getCurrentTime
getXFTPClient (1, testXFTPServer, Nothing) cfg ts (\_ -> pure ()) >>= \case
getXFTPClient (1, testXFTPServer, Nothing) cfg [] ts (\_ -> pure ()) >>= \case
Right c -> client c
Left e -> error $ show e
+1 -1
View File
@@ -223,7 +223,7 @@ testInactiveClientExpiration :: Expectation
testInactiveClientExpiration = withXFTPServerCfg testXFTPServerConfig {inactiveClientExpiration} $ \_ -> runRight_ $ do
disconnected <- newEmptyTMVarIO
ts <- liftIO getCurrentTime
c <- ExceptT $ getXFTPClient (1, testXFTPServer, Nothing) testXFTPClientConfig ts (\_ -> atomically $ putTMVar disconnected ())
c <- ExceptT $ getXFTPClient (1, testXFTPServer, Nothing) testXFTPClientConfig [] ts (\_ -> atomically $ putTMVar disconnected ())
pingXFTP c
liftIO $ do
threadDelay 100000