mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-31 09:38:24 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3d2bf6ce43 | ||
|
|
a678758de8 | ||
|
|
b51969c498 |
+15
-34
@@ -58,11 +58,11 @@ jobs:
|
||||
# =============================
|
||||
|
||||
build:
|
||||
name: "ubuntu-${{ matrix.os }}-${{ matrix.arch }}, GHC: ${{ matrix.ghc }}"
|
||||
name: "ubuntu-${{ matrix.os }}, GHC: ${{ matrix.ghc }}"
|
||||
needs: maybe-release
|
||||
env:
|
||||
apps: "smp-server xftp-server ntf-server xftp"
|
||||
runs-on: ${{ matrix.runner }}
|
||||
runs-on: ubuntu-${{ matrix.os }}
|
||||
services:
|
||||
postgres:
|
||||
image: postgres:15
|
||||
@@ -81,34 +81,16 @@ jobs:
|
||||
matrix:
|
||||
include:
|
||||
- os: 22.04
|
||||
os_underscore: 22_04
|
||||
arch: x86-64
|
||||
runner: "ubuntu-22.04"
|
||||
ghc: "8.10.7"
|
||||
platform_name: 22_04-8.10.7
|
||||
should_run: ${{ !(github.ref == 'refs/heads/stable' || startsWith(github.ref, 'refs/tags/v')) }}
|
||||
- os: 22.04
|
||||
os_underscore: 22_04
|
||||
arch: x86-64
|
||||
runner: "ubuntu-22.04"
|
||||
ghc: "9.6.3"
|
||||
platform_name: 22_04-x86-64
|
||||
should_run: true
|
||||
- os: 24.04
|
||||
os_underscore: 24_04
|
||||
arch: x86-64
|
||||
runner: "ubuntu-24.04"
|
||||
ghc: "9.6.3"
|
||||
should_run: true
|
||||
- os: 22.04
|
||||
os_underscore: 22_04
|
||||
arch: aarch64
|
||||
runner: "ubuntu-22.04-arm"
|
||||
ghc: "9.6.3"
|
||||
should_run: true
|
||||
- os: 24.04
|
||||
os_underscore: 24_04
|
||||
arch: aarch64
|
||||
runner: "ubuntu-24.04-arm"
|
||||
ghc: "9.6.3"
|
||||
platform_name: 24_04-x86-64
|
||||
should_run: true
|
||||
steps:
|
||||
- name: Clone project
|
||||
@@ -145,7 +127,11 @@ jobs:
|
||||
context: .
|
||||
load: true
|
||||
file: Dockerfile.build
|
||||
tags: build/${{ matrix.os }}:latest
|
||||
tags: build/${{ matrix.platform_name }}:latest
|
||||
cache-from: |
|
||||
type=gha
|
||||
type=gha,scope=master
|
||||
cache-to: type=gha,mode=max
|
||||
build-args: |
|
||||
TAG=${{ matrix.os }}
|
||||
GHC=${{ matrix.ghc }}
|
||||
@@ -157,28 +143,23 @@ jobs:
|
||||
path: |
|
||||
~/.cabal/store
|
||||
dist-newstyle
|
||||
key: ubuntu-${{ matrix.os }}-${{ matrix.arch }}-ghc${{ matrix.ghc }}-${{ hashFiles('cabal.project', 'simplexmq.cabal') }}
|
||||
key: ${{ matrix.os }}-${{ hashFiles('cabal.project', 'simplexmq.cabal') }}
|
||||
|
||||
- name: Start container
|
||||
if: matrix.should_run == true
|
||||
shell: bash
|
||||
run: |
|
||||
docker run -t -d \
|
||||
--device /dev/fuse \
|
||||
--cap-add SYS_ADMIN \
|
||||
--security-opt apparmor:unconfined \
|
||||
--name builder \
|
||||
-v ~/.cabal:/root/.cabal \
|
||||
-v /home/runner/work/_temp:/home/runner/work/_temp \
|
||||
-v ${{ github.workspace }}:/project \
|
||||
build/${{ matrix.os }}:latest
|
||||
build/${{ matrix.platform_name }}:latest
|
||||
|
||||
- name: Build smp-server (postgresql) and tests
|
||||
if: matrix.should_run == true
|
||||
shell: docker exec -t builder sh -eu {0}
|
||||
run: |
|
||||
chmod -R 777 dist-newstyle ~/.cabal && git config --global --add safe.directory '*'
|
||||
cabal clean
|
||||
cabal update
|
||||
cabal build --jobs=$(nproc) --enable-tests -fserver_postgres
|
||||
mkdir -p /out
|
||||
@@ -200,7 +181,7 @@ jobs:
|
||||
id: prepare-postgres
|
||||
shell: bash
|
||||
run: |
|
||||
name="smp-server-postgres-ubuntu-${{ matrix.os_underscore }}-${{ matrix.arch }}"
|
||||
name="smp-server-postgres-ubuntu-${{ matrix.platform_name }}"
|
||||
docker cp builder:/out/smp-server $name
|
||||
|
||||
path="${{ github.workspace }}/$name"
|
||||
@@ -232,9 +213,9 @@ jobs:
|
||||
printf 'bins<<EOF\n' > bins.output
|
||||
printf 'hashes<<EOF\n' > hashes.output
|
||||
for i in ${{ env.apps }}; do
|
||||
name="$i-ubuntu-${{ matrix.os_underscore }}-${{ matrix.arch }}"
|
||||
mv ./out/$i ./$i-ubuntu-${{ matrix.platform_name }}
|
||||
|
||||
mv ./out/$i ./$name
|
||||
name="$i-ubuntu-${{ matrix.platform_name }}"
|
||||
|
||||
path="${{ github.workspace }}/$name"
|
||||
hash="SHA2-256($name)= $(openssl sha256 $path | cut -d' ' -f 2)"
|
||||
|
||||
@@ -25,7 +25,7 @@ jobs:
|
||||
|
||||
- name: Execute reproduce script
|
||||
run: |
|
||||
${GITHUB_WORKSPACE}/scripts/simplexmq-reproduce-builds.sh "$TAG" || :
|
||||
${GITHUB_WORKSPACE}/scripts/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}-simplexmq/_sha256sums" ]; then
|
||||
if [ -f "${GITHUB_WORKSPACE}/$TAG/_sha256sums" ]; then
|
||||
exit 0
|
||||
else
|
||||
curl --proto '=https' --tlsv1.2 -sSf \
|
||||
|
||||
@@ -1,34 +1,3 @@
|
||||
# 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:
|
||||
|
||||
@@ -223,14 +223,11 @@
|
||||
<table id="public-info">
|
||||
<tr class="text-grey-black dark:text-white text-base">
|
||||
<td>Server version:</td>
|
||||
<td>${version}<x-commit> / <a href="${commitSourceCode}/commit/${commit}" target="_blank">${shortCommit}</a></x-commit></td>
|
||||
<td>${version}</td>
|
||||
</tr>
|
||||
<tr class="text-grey-black dark:text-white text-base">
|
||||
<td>Source code:</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>
|
||||
<td><a href="${sourceCode}" target="_blank">${sourceCode}</a></td>
|
||||
</tr>
|
||||
<x-website>
|
||||
<tr class="text-grey-black dark:text-white text-base">
|
||||
|
||||
@@ -512,8 +512,6 @@
|
||||
element.innerHTML = 'This is a one-time link of the SimpleX network user'
|
||||
} else if (url.includes('/c')) {
|
||||
element.innerHTML = 'This is a public channel address on SimpleX network'
|
||||
} else if (url.includes('/r')) {
|
||||
element.innerHTML = 'This is a chat relay address on SimpleX network'
|
||||
}
|
||||
}
|
||||
</script>
|
||||
|
||||
@@ -12,7 +12,6 @@ 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 (..))
|
||||
@@ -23,9 +22,8 @@ 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 (..), simplexmqSource)
|
||||
import Simplex.Messaging.Server.Main (EmbeddedWebParams (..), WebHttpsParams (..))
|
||||
import Simplex.Messaging.Transport (simplexMQVersion)
|
||||
import Simplex.Messaging.Transport.Client (TransportHost (..))
|
||||
import Simplex.Messaging.Util (tshow)
|
||||
@@ -105,7 +103,6 @@ generateSite si onionHost sitePath = do
|
||||
createLinkPage "a"
|
||||
createLinkPage "c"
|
||||
createLinkPage "g"
|
||||
createLinkPage "r"
|
||||
createLinkPage "i"
|
||||
logInfo $ "Generated static site contents at " <> tshow sitePath
|
||||
where
|
||||
@@ -119,7 +116,7 @@ generateSite si onionHost sitePath = do
|
||||
serverInformation :: ServerInformation -> Maybe TransportHost -> ByteString
|
||||
serverInformation ServerInformation {config, information} onionHost = render E.indexHtml substs
|
||||
where
|
||||
substs = substConfig <> substInfo <> [("onionHost", strEncode <$> onionHost)]
|
||||
substs = substConfig <> maybe [] substInfo information <> [("onionHost", strEncode <$> onionHost)]
|
||||
substConfig =
|
||||
[ ( "persistence",
|
||||
Just $ case persistence config of
|
||||
@@ -134,7 +131,7 @@ serverInformation ServerInformation {config, information} onionHost = render E.i
|
||||
]
|
||||
yesNo True = "Yes"
|
||||
yesNo False = "No"
|
||||
substInfo =
|
||||
substInfo spi =
|
||||
concat
|
||||
[ basic,
|
||||
maybe [("usageConditions", Nothing), ("usageAmendments", Nothing)] conds (usageConditions spi),
|
||||
@@ -146,16 +143,10 @@ serverInformation ServerInformation {config, information} onionHost = render E.i
|
||||
]
|
||||
where
|
||||
basic =
|
||||
[ ("sourceCode", if T.null sc then Nothing else Just (encodeUtf8 sc)),
|
||||
("noSourceCode", if T.null sc then Just "none" else Nothing),
|
||||
[ ("sourceCode", Just . encodeUtf8 $ sourceCode spi),
|
||||
("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)
|
||||
@@ -237,8 +228,8 @@ section_ label content' src =
|
||||
(inside, next') ->
|
||||
let next = B.drop (B.length endMarker) next'
|
||||
in case content' of
|
||||
Just content | not (B.null content) -> before <> item_ label content inside <> section_ label content' next
|
||||
_ -> before <> next -- collapse section
|
||||
Nothing -> before <> next -- collapse section
|
||||
Just content -> before <> item_ label content inside <> section_ label content' next
|
||||
where
|
||||
startMarker = "<x-" <> label <> ">"
|
||||
endMarker = "</x-" <> label <> ">"
|
||||
|
||||
@@ -1,104 +0,0 @@
|
||||
# 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.
|
||||
@@ -133,9 +133,6 @@ done
|
||||
|
||||
# If everything is still okay, compute checksums file
|
||||
if [ "$bad" = 0 ]; then
|
||||
curl -L "${repo}/archive/refs/tags/${TAG}.tar.gz" -o "$init_dir/$TAG-$repo_name/from-source/source_code.tar.gz"
|
||||
curl -L "${repo}/archive/refs/tags/${TAG}.zip" -o "$init_dir/$TAG-$repo_name/from-source/source_code.zip"
|
||||
|
||||
sha256sum "$path_bin"/from-source/* | sed -e "s|$PWD/||g" -e 's|from-source/||g' -e "s|-$repo_name||g" > "$path_bin/_sha256sums"
|
||||
|
||||
printf 'Checksums computed - %s\n' "$path_bin/_sha256sums"
|
||||
|
||||
+1
-5
@@ -1,7 +1,7 @@
|
||||
cabal-version: 1.12
|
||||
|
||||
name: simplexmq
|
||||
version: 6.4.5.2
|
||||
version: 6.4.1.0
|
||||
synopsis: SimpleXMQ message broker
|
||||
description: This package includes <./docs/Simplex-Messaging-Server.html server>,
|
||||
<./docs/Simplex-Messaging-Client.html client> and
|
||||
@@ -224,7 +224,6 @@ 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
|
||||
@@ -234,7 +233,6 @@ 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
|
||||
@@ -355,12 +353,10 @@ 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
|
||||
|
||||
@@ -75,7 +75,7 @@ import Simplex.Messaging.Encoding
|
||||
import Simplex.Messaging.Encoding.String (strDecode, strEncode)
|
||||
import Simplex.Messaging.Protocol (ProtocolServer, ProtocolType (..), XFTPServer)
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
import Simplex.Messaging.Util (allFinally, catchAll_, catchAllErrors, liftError, tshow, unlessM, whenM)
|
||||
import Simplex.Messaging.Util (catchAll_, liftError, tshow, unlessM, whenM)
|
||||
import System.FilePath (takeFileName, (</>))
|
||||
import UnliftIO
|
||||
import UnliftIO.Directory
|
||||
@@ -198,10 +198,10 @@ runXFTPRcvWorker c srv Worker {doWork} = do
|
||||
liftIO $ waitForUserNetwork c
|
||||
atomically $ incXFTPServerStat c userId srv downloadAttempts
|
||||
downloadFileChunk fc replica approvedRelays
|
||||
`catchAllErrors` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e
|
||||
`catchAgentError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e
|
||||
where
|
||||
retryLoop loop e replicaDelay = do
|
||||
flip catchAllErrors (\_ -> pure ()) $ do
|
||||
flip catchAgentError (\_ -> pure ()) $ do
|
||||
when (serverHostError e) $ notify c (fromMaybe rcvFileEntityId redirectEntityId_) (RFWARN e)
|
||||
liftIO $ closeXFTPServerClient c userId server digest
|
||||
withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay
|
||||
@@ -280,7 +280,7 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
|
||||
runXFTPOperation AgentConfig {rcvFilesTTL} =
|
||||
withWork c doWork (`getNextRcvFileToDecrypt` rcvFilesTTL) $
|
||||
\f@RcvFile {rcvFileId, rcvFileEntityId, tmpPath, redirect} ->
|
||||
decryptFile f `catchAllErrors` rcvWorkerInternalError c rcvFileId rcvFileEntityId (redirectEntityId <$> redirect) tmpPath
|
||||
decryptFile f `catchAgentError` rcvWorkerInternalError c rcvFileId rcvFileEntityId (redirectEntityId <$> redirect) tmpPath
|
||||
decryptFile :: RcvFile -> AM ()
|
||||
decryptFile RcvFile {rcvFileId, rcvFileEntityId, size, digest, key, nonce, tmpPath, saveFile, status, chunks, redirect} = do
|
||||
let CryptoFile savePath cfArgs = saveFile
|
||||
@@ -307,7 +307,7 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
|
||||
liftIO $ waitUntilForeground c
|
||||
withStore' c (`updateRcvFileComplete` rcvFileId)
|
||||
-- proceed with redirect
|
||||
yaml <- liftError (FILE . FILE_IO . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `allFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
|
||||
yaml <- liftError (FILE . FILE_IO . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `agentFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
|
||||
next@FileDescription {chunks = nextChunks} <- case strDecode (LB.toStrict yaml) of
|
||||
-- TODO switch to another error constructor
|
||||
Left _ -> throwE . FILE $ REDIRECT "decode error"
|
||||
@@ -399,7 +399,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
|
||||
runXFTPOperation cfg@AgentConfig {sndFilesTTL} =
|
||||
withWork c doWork (`getNextSndFileToPrepare` sndFilesTTL) $
|
||||
\f@SndFile {sndFileId, sndFileEntityId, prefixPath} ->
|
||||
prepareFile cfg f `catchAllErrors` sndWorkerInternalError c sndFileId sndFileEntityId prefixPath
|
||||
prepareFile cfg f `catchAgentError` sndWorkerInternalError c sndFileId sndFileEntityId prefixPath
|
||||
prepareFile :: AgentConfig -> SndFile -> AM ()
|
||||
prepareFile _ SndFile {prefixPath = Nothing} =
|
||||
throwE $ INTERNAL "no prefix path"
|
||||
@@ -468,11 +468,11 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
|
||||
liftIO $ waitForUserNetwork c
|
||||
let triedAllSrvs = n > userSrvCount
|
||||
createWithNextSrv triedHosts
|
||||
`catchAllErrors` \e -> retryOnError "XFTP prepare worker" (retryLoop loop triedAllSrvs e) (throwE e) e
|
||||
`catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop triedAllSrvs e) (throwE e) e
|
||||
where
|
||||
-- we don't do closeXFTPServerClient here to not risk closing connection for concurrent chunk upload
|
||||
retryLoop loop triedAllSrvs e = do
|
||||
flip catchAllErrors (\_ -> pure ()) $ do
|
||||
flip catchAgentError (\_ -> pure ()) $ do
|
||||
when (triedAllSrvs && serverHostError e) $ notify c sndFileEntityId $ SFWARN e
|
||||
liftIO $ assertAgentForeground c
|
||||
loop
|
||||
@@ -508,10 +508,10 @@ runXFTPSndWorker c srv Worker {doWork} = do
|
||||
liftIO $ waitForUserNetwork c
|
||||
atomically $ incXFTPServerStat c userId srv uploadAttempts
|
||||
uploadFileChunk cfg fc replica
|
||||
`catchAllErrors` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e
|
||||
`catchAgentError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e
|
||||
where
|
||||
retryLoop loop e replicaDelay = do
|
||||
flip catchAllErrors (\_ -> pure ()) $ do
|
||||
flip catchAgentError (\_ -> pure ()) $ do
|
||||
when (serverHostError e) $ notify c sndFileEntityId $ SFWARN e
|
||||
liftIO $ closeXFTPServerClient c userId server digest
|
||||
withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay
|
||||
@@ -681,10 +681,10 @@ runXFTPDelWorker c srv Worker {doWork} = do
|
||||
liftIO $ waitForUserNetwork c
|
||||
atomically $ incXFTPServerStat c userId srv deleteAttempts
|
||||
deleteChunkReplica
|
||||
`catchAllErrors` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e
|
||||
`catchAgentError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e
|
||||
where
|
||||
retryLoop loop e replicaDelay = do
|
||||
flip catchAllErrors (\_ -> pure ()) $ do
|
||||
flip catchAgentError (\_ -> pure ()) $ do
|
||||
when (serverHostError e) $ notify c "" $ SFWARN e
|
||||
liftIO $ closeXFTPServerClient c userId server chunkDigest
|
||||
withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay
|
||||
|
||||
@@ -31,7 +31,6 @@ 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
|
||||
@@ -46,7 +45,6 @@ import Simplex.Messaging.Client
|
||||
transportClientConfig,
|
||||
clientSocksCredentials,
|
||||
unexpectedResponse,
|
||||
useWebPort,
|
||||
)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import qualified Simplex.Messaging.Crypto.Lazy as LC
|
||||
@@ -59,8 +57,6 @@ import Simplex.Messaging.Protocol
|
||||
RecipientId,
|
||||
SenderId,
|
||||
pattern NoEntity,
|
||||
NetworkError (..),
|
||||
toNetworkError,
|
||||
)
|
||||
import Simplex.Messaging.Transport (ALPN, CertChainPubKey (..), HandshakeError (..), THandleAuth (..), THandleParams (..), TransportError (..), TransportPeer (..), defaultSupportedParams)
|
||||
import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost)
|
||||
@@ -108,13 +104,12 @@ defaultXFTPClientConfig =
|
||||
clientALPN = Just alpnSupportedXFTPhandshakes
|
||||
}
|
||||
|
||||
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
|
||||
getXFTPClient :: TransportSession FileResponse -> XFTPClientConfig -> UTCTime -> (XFTPClient -> IO ()) -> IO (Either XFTPClientError XFTPClient)
|
||||
getXFTPClient transportSession@(_, srv, _) config@XFTPClientConfig {clientALPN, xftpNetworkConfig, serverVRange} 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 useALPN
|
||||
let tcConfig = transportClientConfig xftpNetworkConfig NRMBackground useHost False clientALPN
|
||||
http2Config = xftpHTTP2Config tcConfig config
|
||||
clientVar <- newTVarIO Nothing
|
||||
let usePort = if null port then "443" else port
|
||||
@@ -126,8 +121,7 @@ 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 alpn | alpn == xftpALPNv1 || alpn == httpALPN11 ->
|
||||
xftpClientHandshakeV1 serverVRange keyHash http2Client thParams0
|
||||
Just "xftp/1" -> xftpClientHandshakeV1 serverVRange keyHash http2Client thParams0
|
||||
_ -> pure thParams0
|
||||
logDebug $ "Client negotiated protocol: " <> tshow thVersion
|
||||
let c = XFTPClient {http2Client, thParams, transportSession, config}
|
||||
@@ -193,7 +187,7 @@ xftpHTTP2Config transportConfig XFTPClientConfig {xftpNetworkConfig = NetworkCon
|
||||
xftpClientError :: HTTP2ClientError -> XFTPClientError
|
||||
xftpClientError = \case
|
||||
HCResponseTimeout -> PCEResponseTimeout
|
||||
HCNetworkError e -> PCENetworkError e
|
||||
HCNetworkError -> PCENetworkError
|
||||
HCIOError e -> PCEIOError e
|
||||
|
||||
sendXFTPCommand :: forall p. FilePartyI p => XFTPClient -> C.APrivateAuthKey -> XFTPFileId -> FileCommand p -> Maybe XFTPChunkSpec -> ExceptT XFTPClientError IO (FileResponse, HTTP2Body)
|
||||
@@ -263,9 +257,9 @@ downloadXFTPChunk g c@XFTPClient {config} rpKey fId chunkSpec@XFTPRcvChunkSpec {
|
||||
ExceptT (sequence <$> (t `timeout` (download cbState `catches` errors))) >>= maybe (throwE PCEResponseTimeout) pure
|
||||
where
|
||||
errors =
|
||||
[ Handler $ \(e :: H.HTTP2Error) -> pure $ Left $ PCENetworkError $ NEConnectError $ displayException e,
|
||||
Handler $ \(e :: IOException) -> pure $ Left $ PCEIOError e,
|
||||
Handler $ \(e :: SomeException) -> pure $ Left $ PCENetworkError $ toNetworkError e
|
||||
[ Handler $ \(_e :: H.HTTP2Error) -> pure $ Left PCENetworkError,
|
||||
Handler $ \(e :: IOException) -> pure $ Left (PCEIOError e),
|
||||
Handler $ \(_e :: SomeException) -> pure $ Left PCENetworkError
|
||||
]
|
||||
download cbState =
|
||||
runExceptT . withExceptT PCEResponseError $
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -45,7 +45,6 @@ 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
|
||||
@@ -70,7 +69,6 @@ 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)
|
||||
@@ -107,14 +105,7 @@ 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
|
||||
<> prometheusMetricsThread_ cfg
|
||||
<> controlPortThread_ cfg
|
||||
)
|
||||
`finally` stopServer
|
||||
raceAny_ (runServer : expireFilesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg) `finally` stopServer
|
||||
where
|
||||
runServer :: M ()
|
||||
runServer = do
|
||||
@@ -133,7 +124,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 alpn | alpn == xftpALPNv1 || alpn == httpALPN11 ->
|
||||
Just "xftp/1" ->
|
||||
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)
|
||||
@@ -249,30 +240,6 @@ 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_ _ = []
|
||||
|
||||
@@ -64,8 +64,6 @@ data XFTPServerConfig = XFTPServerConfig
|
||||
logStatsStartTime :: Int64,
|
||||
serverStatsLogFile :: FilePath,
|
||||
serverStatsBackupFile :: Maybe FilePath,
|
||||
prometheusInterval :: Maybe Int,
|
||||
prometheusMetricsFile :: FilePath,
|
||||
transportConfig :: TransportServerConfig,
|
||||
responseDelay :: Int
|
||||
}
|
||||
|
||||
@@ -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 (eitherToMaybe, safeDecodeUtf8, tshow)
|
||||
import Simplex.Messaging.Util (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" <> simplexmqVersionCommit
|
||||
serverVersion = "SimpleX XFTP server v" <> simplexMQVersion
|
||||
defaultServerPort = "443"
|
||||
executableName = "file-server"
|
||||
storeLogFilePath = combine logPath "file-server-store.log"
|
||||
@@ -89,9 +89,6 @@ 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\
|
||||
@@ -191,12 +188,10 @@ 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 <> httpALPN)
|
||||
(Just alpnSupportedXFTPhandshakes)
|
||||
False,
|
||||
responseDelay = 0
|
||||
}
|
||||
|
||||
@@ -1,134 +0,0 @@
|
||||
{-# 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#-}
|
||||
@@ -14,7 +14,6 @@ module Simplex.FileTransfer.Transport
|
||||
blockedFilesXFTPVersion,
|
||||
xftpClientHandshakeStub,
|
||||
alpnSupportedXFTPhandshakes,
|
||||
xftpALPNv1,
|
||||
XFTPClientHandshake (..),
|
||||
-- xftpClientHandshake,
|
||||
XFTPServerHandshake (..),
|
||||
@@ -106,10 +105,7 @@ xftpClientHandshakeStub :: c 'TClient -> Maybe C.KeyPairX25519 -> C.KeyHash -> V
|
||||
xftpClientHandshakeStub _c _ks _keyHash _xftpVRange _proxyServer _serviceKeys = throwE TEVersion
|
||||
|
||||
alpnSupportedXFTPhandshakes :: [ALPN]
|
||||
alpnSupportedXFTPhandshakes = [xftpALPNv1]
|
||||
|
||||
xftpALPNv1 :: ALPN
|
||||
xftpALPNv1 = "xftp/1"
|
||||
alpnSupportedXFTPhandshakes = ["xftp/1"]
|
||||
|
||||
data XFTPServerHandshake = XFTPServerHandshake
|
||||
{ xftpVersionRange :: VersionRangeXFTP,
|
||||
|
||||
@@ -284,13 +284,13 @@ saveServersStats c@AgentClient {subQ, smpServersStats, xftpServersStats, ntfServ
|
||||
xss <- mapM (liftIO . getAgentXFTPServerStats) =<< readTVarIO xftpServersStats
|
||||
nss <- mapM (liftIO . getAgentNtfServerStats) =<< readTVarIO ntfServersStats
|
||||
let stats = AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss}
|
||||
tryAllErrors' (withStore' c (`updateServersStats` stats)) >>= \case
|
||||
tryAgentError' (withStore' c (`updateServersStats` stats)) >>= \case
|
||||
Left e -> atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
|
||||
Right () -> pure ()
|
||||
|
||||
restoreServersStats :: AgentClient -> AM' ()
|
||||
restoreServersStats c@AgentClient {smpServersStats, xftpServersStats, ntfServersStats, srvStatsStartedAt} = do
|
||||
tryAllErrors' (withStore c getServersStats) >>= \case
|
||||
tryAgentError' (withStore c getServersStats) >>= \case
|
||||
Left e -> atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e)
|
||||
Right (startedAt, Nothing) -> atomically $ writeTVar srvStatsStartedAt startedAt
|
||||
Right (startedAt, Just AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss}) -> do
|
||||
@@ -774,7 +774,7 @@ acceptContactAsync' :: AgentClient -> UserId -> ACorrId -> Bool -> InvitationId
|
||||
acceptContactAsync' c userId corrId enableNtfs invId ownConnInfo pqSupport subMode = do
|
||||
Invitation {connReq} <- withStore c $ \db -> getInvitation db "acceptContactAsync'" invId
|
||||
withStore' c $ \db -> acceptInvitation db invId ownConnInfo
|
||||
joinConnAsync c userId corrId enableNtfs connReq ownConnInfo pqSupport subMode `catchAllErrors` \err -> do
|
||||
joinConnAsync c userId corrId enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
|
||||
withStore' c (`unacceptInvitation` invId)
|
||||
throwE err
|
||||
|
||||
@@ -912,19 +912,12 @@ 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, clData) <- liftEither $ SL.decryptLinkData @c linkKey k d
|
||||
r@(cReq, _) <- 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 $ if srv' == srv then r else (updateConnReqServer srv cReq, clData)
|
||||
pure r
|
||||
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
|
||||
@@ -961,7 +954,7 @@ newRcvConnSrv c nm userId connId enableNtfs cMode userData_ clientData pqInitKey
|
||||
createRcvQueue nonce_ qd e2eKeys = do
|
||||
AgentConfig {smpClientVRange = vr} <- asks config
|
||||
ntfServer_ <- if enableNtfs then newQueueNtfServer else pure Nothing
|
||||
(rq, qUri, tSess, sessId) <- newRcvQueue_ c nm userId connId srvWithAuth vr qd (isJust ntfServer_) subMode nonce_ e2eKeys `catchAllErrors` \e -> liftIO (print e) >> throwE e
|
||||
(rq, qUri, tSess, sessId) <- newRcvQueue_ c nm userId connId srvWithAuth vr qd (isJust ntfServer_) subMode nonce_ e2eKeys `catchAgentError` \e -> liftIO (print e) >> throwE e
|
||||
atomically $ incSMPServerStat c userId srv connCreated
|
||||
rq' <- withStore c $ \db -> updateNewConnRcv db connId rq
|
||||
lift . when (subMode == SMSubscribe) $ addNewQueueSubscription c rq' tSess sessId
|
||||
@@ -1351,7 +1344,7 @@ subscribeClientService' = undefined
|
||||
|
||||
-- requesting messages sequentially, to reduce memory usage
|
||||
getConnectionMessages' :: AgentClient -> NonEmpty ConnMsgReq -> AM' (NonEmpty (Either AgentErrorType (Maybe SMPMsgMeta)))
|
||||
getConnectionMessages' c = mapM $ tryAllErrors' . getConnectionMessage
|
||||
getConnectionMessages' c = mapM $ tryAgentError' . getConnectionMessage
|
||||
where
|
||||
getConnectionMessage :: ConnMsgReq -> AM (Maybe SMPMsgMeta)
|
||||
getConnectionMessage (ConnMsgReq connId dbQueueId msgTs_) = do
|
||||
@@ -1363,7 +1356,7 @@ getConnectionMessages' c = mapM $ tryAllErrors' . getConnectionMessage
|
||||
ContactConnection _ rq -> pure rq
|
||||
SndConnection _ _ -> throwE $ CONN SIMPLEX "getConnectionMessage"
|
||||
NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection"
|
||||
msg_ <- getQueueMessage c rq `catchAllErrors` \e -> atomically (releaseGetLock c rq) >> throwError e
|
||||
msg_ <- getQueueMessage c rq `catchAgentError` \e -> atomically (releaseGetLock c rq) >> throwError e
|
||||
when (isNothing msg_) $ do
|
||||
atomically $ releaseGetLock c rq
|
||||
forM_ msgTs_ $ \msgTs -> withStore' c $ \db -> setLastBrokerTs db connId (DBEntityId dbQueueId) msgTs
|
||||
@@ -1534,7 +1527,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
|
||||
pure CCCompleted
|
||||
-- duplex connection is matched to handle SKEY retries
|
||||
DuplexConnection cData _ (sq :| _) -> do
|
||||
tryAllErrors (mapM_ (connectReplyQueues c cData ownConnInfo (Just sq)) (L.nonEmpty $ smpReplyQueues senderConf)) >>= \case
|
||||
tryAgentError (mapM_ (connectReplyQueues c cData ownConnInfo (Just sq)) (L.nonEmpty $ smpReplyQueues senderConf)) >>= \case
|
||||
Right () -> pure CCCompleted
|
||||
Left e
|
||||
| temporaryOrHostError e && Just server /= server_ -> do
|
||||
@@ -1621,7 +1614,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
|
||||
tryMoveableCommand action = withRetryInterval ri $ \_ loop -> do
|
||||
liftIO $ waitWhileSuspended c
|
||||
liftIO $ waitForUserNetwork c
|
||||
tryAllErrors action >>= \case
|
||||
tryAgentError action >>= \case
|
||||
Left e
|
||||
| temporaryOrHostError e -> retrySndOp c loop
|
||||
| otherwise -> cmdError e
|
||||
@@ -2065,7 +2058,7 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
|
||||
ackQueueMessage :: AgentClient -> RcvQueue -> SMP.MsgId -> AM ()
|
||||
ackQueueMessage c rq@RcvQueue {userId, connId, server} srvMsgId = do
|
||||
atomically $ incSMPServerStat c userId server ackAttempts
|
||||
tryAllErrors (sendAck c rq srvMsgId) >>= \case
|
||||
tryAgentError (sendAck c rq srvMsgId) >>= \case
|
||||
Right _ -> sendMsgNtf ackMsgs
|
||||
Left (SMP _ SMP.NO_MSG) -> sendMsgNtf ackNoMsgErrs
|
||||
Left e -> do
|
||||
@@ -2076,7 +2069,7 @@ ackQueueMessage c rq@RcvQueue {userId, connId, server} srvMsgId = do
|
||||
atomically $ incSMPServerStat c userId server stat
|
||||
whenM (liftIO $ hasGetLock c rq) $ do
|
||||
atomically $ releaseGetLock c rq
|
||||
brokerTs_ <- eitherToMaybe <$> tryAllErrors (withStore c $ \db -> getRcvMsgBrokerTs db connId srvMsgId)
|
||||
brokerTs_ <- eitherToMaybe <$> tryAgentError (withStore c $ \db -> getRcvMsgBrokerTs db connId srvMsgId)
|
||||
atomically $ writeTBQueue (subQ c) ("", connId, AEvt SAEConn $ MSGNTF srvMsgId brokerTs_)
|
||||
|
||||
-- | Suspend SMP agent connection (OFF command) in Reader monad
|
||||
@@ -2307,7 +2300,7 @@ registerNtfToken' c nm suppliedDeviceToken suppliedNtfMode =
|
||||
replaceToken :: NtfTokenId -> AM NtfTknStatus
|
||||
replaceToken tknId = do
|
||||
ns <- asks ntfSupervisor
|
||||
tryReplace ns `catchAllErrors` \e ->
|
||||
tryReplace ns `catchAgentError` \e ->
|
||||
if temporaryOrHostError e
|
||||
then throwE e
|
||||
else do
|
||||
@@ -2564,7 +2557,7 @@ cleanupManager c@AgentClient {subQ} = do
|
||||
where
|
||||
run :: forall e. AEntityI e => (AgentErrorType -> AEvent e) -> AM () -> AM' ()
|
||||
run err a = do
|
||||
waitActive . runExceptT $ a `catchAllErrors` (notify "" . err)
|
||||
waitActive . runExceptT $ a `catchAgentError` (notify "" . err)
|
||||
step <- asks $ cleanupStepInterval . config
|
||||
liftIO $ threadDelay step
|
||||
-- we are catching it to avoid CRITICAL errors in tests when this is the only remaining handle to active
|
||||
@@ -2578,33 +2571,33 @@ cleanupManager c@AgentClient {subQ} = do
|
||||
deleteRcvFilesExpired = do
|
||||
rcvFilesTTL <- asks $ rcvFilesTTL . config
|
||||
rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL)
|
||||
forM_ rcvExpired $ \(dbId, entId, p) -> flip catchAllErrors (notify entId . RFERR) $ do
|
||||
forM_ rcvExpired $ \(dbId, entId, p) -> flip catchAgentError (notify entId . RFERR) $ do
|
||||
lift $ removePath =<< toFSFilePath p
|
||||
withStore' c (`deleteRcvFile'` dbId)
|
||||
deleteRcvFilesDeleted = do
|
||||
rcvDeleted <- withStore' c getCleanupRcvFilesDeleted
|
||||
forM_ rcvDeleted $ \(dbId, entId, p) -> flip catchAllErrors (notify entId . RFERR) $ do
|
||||
forM_ rcvDeleted $ \(dbId, entId, p) -> flip catchAgentError (notify entId . RFERR) $ do
|
||||
lift $ removePath =<< toFSFilePath p
|
||||
withStore' c (`deleteRcvFile'` dbId)
|
||||
deleteRcvFilesTmpPaths = do
|
||||
rcvTmpPaths <- withStore' c getCleanupRcvFilesTmpPaths
|
||||
forM_ rcvTmpPaths $ \(dbId, entId, p) -> flip catchAllErrors (notify entId . RFERR) $ do
|
||||
forM_ rcvTmpPaths $ \(dbId, entId, p) -> flip catchAgentError (notify entId . RFERR) $ do
|
||||
lift $ removePath =<< toFSFilePath p
|
||||
withStore' c (`updateRcvFileNoTmpPath` dbId)
|
||||
deleteSndFilesExpired = do
|
||||
sndFilesTTL <- asks $ sndFilesTTL . config
|
||||
sndExpired <- withStore' c (`getSndFilesExpired` sndFilesTTL)
|
||||
forM_ sndExpired $ \(dbId, entId, p) -> flip catchAllErrors (notify entId . SFERR) $ do
|
||||
forM_ sndExpired $ \(dbId, entId, p) -> flip catchAgentError (notify entId . SFERR) $ do
|
||||
lift . forM_ p $ removePath <=< toFSFilePath
|
||||
withStore' c (`deleteSndFile'` dbId)
|
||||
deleteSndFilesDeleted = do
|
||||
sndDeleted <- withStore' c getCleanupSndFilesDeleted
|
||||
forM_ sndDeleted $ \(dbId, entId, p) -> flip catchAllErrors (notify entId . SFERR) $ do
|
||||
forM_ sndDeleted $ \(dbId, entId, p) -> flip catchAgentError (notify entId . SFERR) $ do
|
||||
lift . forM_ p $ removePath <=< toFSFilePath
|
||||
withStore' c (`deleteSndFile'` dbId)
|
||||
deleteSndFilesPrefixPaths = do
|
||||
sndPrefixPaths <- withStore' c getCleanupSndFilesPrefixPaths
|
||||
forM_ sndPrefixPaths $ \(dbId, entId, p) -> flip catchAllErrors (notify entId . SFERR) $ do
|
||||
forM_ sndPrefixPaths $ \(dbId, entId, p) -> flip catchAgentError (notify entId . SFERR) $ do
|
||||
lift $ removePath =<< toFSFilePath p
|
||||
withStore' c (`updateSndFileNoPrefixPath` dbId)
|
||||
deleteExpiredReplicasForDeletion = do
|
||||
@@ -2652,10 +2645,10 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
|
||||
where
|
||||
withRcvConn :: SMP.RecipientId -> (forall c. RcvQueue -> Connection c -> AM ()) -> AM' ()
|
||||
withRcvConn rId a = do
|
||||
tryAllErrors' (withStore c $ \db -> getRcvConn db srv rId) >>= \case
|
||||
tryAgentError' (withStore c $ \db -> getRcvConn db srv rId) >>= \case
|
||||
Left e -> notify' "" (ERR e)
|
||||
Right (rq@RcvQueue {connId}, SomeConn _ conn) ->
|
||||
tryAllErrors' (a rq conn) >>= \case
|
||||
tryAgentError' (a rq conn) >>= \case
|
||||
Left e -> notify' connId (ERR e)
|
||||
Right () -> pure ()
|
||||
processSubOk :: RcvQueue -> TVar [ConnId] -> AM ()
|
||||
@@ -2739,7 +2732,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
|
||||
_ -> pure ()
|
||||
let encryptedMsgHash = C.sha256Hash encAgentMessage
|
||||
g <- asks random
|
||||
tryAllErrors (agentClientMsg g encryptedMsgHash) >>= \case
|
||||
tryAgentError (agentClientMsg g encryptedMsgHash) >>= \case
|
||||
Right (Just (msgId, msgMeta, aMessage, rcPrev)) -> do
|
||||
conn'' <- resetRatchetSync
|
||||
case aMessage of
|
||||
@@ -2848,7 +2841,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
|
||||
ackDel :: InternalId -> AM ACKd
|
||||
ackDel aId = enqueueCmd (ICAckDel rId srvMsgId aId) $> ACKd
|
||||
handleNotifyAck :: AM ACKd -> AM ACKd
|
||||
handleNotifyAck m = m `catchAllErrors` \e -> notify (ERR e) >> ack
|
||||
handleNotifyAck m = m `catchAgentError` \e -> notify (ERR e) >> ack
|
||||
SMP.END ->
|
||||
atomically (ifM (activeClientSession c tSess sessId) (removeSubscription c connId $> True) (pure False))
|
||||
>>= notifyEnd
|
||||
@@ -3006,7 +2999,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
|
||||
messagesRcvd :: NonEmpty AMessageReceipt -> MsgMeta -> Connection 'CDuplex -> AM ACKd
|
||||
messagesRcvd rcpts msgMeta@MsgMeta {broker = (srvMsgId, _)} _ = do
|
||||
logServer "<--" c srv rId $ "MSG <RCPT>:" <> logSecret' srvMsgId
|
||||
rs <- forM rcpts $ \rcpt -> clientReceipt rcpt `catchAllErrors` \e -> notify (ERR e) $> Nothing
|
||||
rs <- forM rcpts $ \rcpt -> clientReceipt rcpt `catchAgentError` \e -> notify (ERR e) $> Nothing
|
||||
case L.nonEmpty . catMaybes $ L.toList rs of
|
||||
Just rs' -> notify (RCVD msgMeta rs') $> ACKPending
|
||||
Nothing -> ack
|
||||
|
||||
@@ -130,7 +130,6 @@ module Simplex.Messaging.Agent.Client
|
||||
hasWorkToDo,
|
||||
hasWorkToDo',
|
||||
withWork,
|
||||
withWork_,
|
||||
withWorkItems,
|
||||
agentOperations,
|
||||
agentOperationBracket,
|
||||
@@ -250,7 +249,6 @@ import Simplex.Messaging.Protocol
|
||||
EntityId (..),
|
||||
ServiceId,
|
||||
ErrorType,
|
||||
NetworkError (..),
|
||||
MsgFlags (..),
|
||||
MsgId,
|
||||
NtfServer,
|
||||
@@ -332,7 +330,7 @@ data AgentClient = AgentClient
|
||||
xftpServers :: TMap UserId (UserServers 'PXFTP),
|
||||
xftpClients :: TMap XFTPTransportSession XFTPClientVar,
|
||||
useNetworkConfig :: TVar (NetworkConfig, NetworkConfig), -- (slow, fast) networks
|
||||
presetDomains :: [HostName],
|
||||
presetSMPDomains :: [HostName],
|
||||
userNetworkInfo :: TVar UserNetworkInfo,
|
||||
userNetworkUpdated :: TVar (Maybe UTCTime),
|
||||
subscrConns :: TVar (Set ConnId),
|
||||
@@ -373,12 +371,12 @@ data SMPConnectedClient = SMPConnectedClient
|
||||
|
||||
type ProxiedRelayVar = SessionVar (Either AgentErrorType ProxiedRelay)
|
||||
|
||||
getAgentWorker :: (Ord k, Show k, AnyError e, MonadUnliftIO m) => String -> Bool -> AgentClient -> k -> TMap k Worker -> (Worker -> ExceptT e m ()) -> m Worker
|
||||
getAgentWorker :: (Ord k, Show k) => String -> Bool -> AgentClient -> k -> TMap k Worker -> (Worker -> AM ()) -> AM' Worker
|
||||
getAgentWorker = getAgentWorker' id pure
|
||||
{-# INLINE getAgentWorker #-}
|
||||
|
||||
getAgentWorker' :: forall a k e m. (Ord k, Show k, AnyError e, MonadUnliftIO m) => (a -> Worker) -> (Worker -> STM a) -> String -> Bool -> AgentClient -> k -> TMap k a -> (a -> ExceptT e m ()) -> m a
|
||||
getAgentWorker' toW fromW name hasWork c@AgentClient {agentEnv} key ws work = do
|
||||
getAgentWorker' :: forall a k. (Ord k, Show k) => (a -> Worker) -> (Worker -> STM a) -> String -> Bool -> AgentClient -> k -> TMap k a -> (a -> AM ()) -> AM' a
|
||||
getAgentWorker' toW fromW name hasWork c key ws work = do
|
||||
atomically (getWorker >>= maybe createWorker whenExists) >>= \w -> runWorker w $> w
|
||||
where
|
||||
getWorker = TM.lookup key ws
|
||||
@@ -391,12 +389,12 @@ getAgentWorker' toW fromW name hasWork c@AgentClient {agentEnv} key ws work = do
|
||||
| otherwise = pure w
|
||||
runWorker w = runWorkerAsync (toW w) runWork
|
||||
where
|
||||
runWork :: m ()
|
||||
runWork = tryAllErrors' (work w) >>= restartOrDelete
|
||||
restartOrDelete :: Either e () -> m ()
|
||||
runWork :: AM' ()
|
||||
runWork = tryAgentError' (work w) >>= restartOrDelete
|
||||
restartOrDelete :: Either AgentErrorType () -> AM' ()
|
||||
restartOrDelete e_ = do
|
||||
t <- liftIO getSystemTime
|
||||
let maxRestarts = maxWorkerRestartsPerMin $ config agentEnv
|
||||
maxRestarts <- asks $ maxWorkerRestartsPerMin . config
|
||||
-- worker may terminate because it was deleted from the map (getWorker returns Nothing), then it won't restart
|
||||
restart <- atomically $ getWorker >>= maybe (pure False) (shouldRestart e_ (toW w) t maxRestarts)
|
||||
when restart runWork
|
||||
@@ -433,7 +431,7 @@ newWorker c = do
|
||||
restarts <- newTVar $ RestartCount 0 0
|
||||
pure Worker {workerId, doWork, action, restarts}
|
||||
|
||||
runWorkerAsync :: MonadUnliftIO m => Worker -> m () -> m ()
|
||||
runWorkerAsync :: Worker -> AM' () -> AM' ()
|
||||
runWorkerAsync Worker {action} work =
|
||||
E.bracket
|
||||
(atomically $ takeTMVar action) -- get current action, locking to avoid race conditions
|
||||
@@ -539,7 +537,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, presetDomai
|
||||
xftpServers,
|
||||
xftpClients,
|
||||
useNetworkConfig,
|
||||
presetDomains,
|
||||
presetSMPDomains = presetDomains,
|
||||
userNetworkInfo,
|
||||
userNetworkUpdated,
|
||||
subscrConns,
|
||||
@@ -667,7 +665,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq
|
||||
pure (clnt, sess)
|
||||
newProxiedRelay :: SMPConnectedClient -> Maybe SMP.BasicAuth -> ProxiedRelayVar -> AM (Either AgentErrorType ProxiedRelay)
|
||||
newProxiedRelay (SMPConnectedClient smp prs) proxyAuth rv =
|
||||
tryAllErrors (liftClient SMP (clientServer smp) $ connectSMPProxiedRelay smp nm destSrv proxyAuth) >>= \case
|
||||
tryAgentError (liftClient SMP (clientServer smp) $ connectSMPProxiedRelay smp nm destSrv proxyAuth) >>= \case
|
||||
Right sess -> do
|
||||
atomically $ putTMVar (sessionVar rv) (Right sess)
|
||||
pure $ Right sess
|
||||
@@ -688,9 +686,9 @@ 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, presetDomains} nm tSess@(_, srv, _) prs v =
|
||||
smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs} nm tSess@(_, srv, _) prs v =
|
||||
newProtocolClient c tSess smpClients connectClient v
|
||||
`catchAllErrors` \e -> lift (resubscribeSMPSession c tSess) >> throwE e
|
||||
`catchAgentError` \e -> lift (resubscribeSMPSession c tSess) >> throwE e
|
||||
where
|
||||
connectClient :: SMPClientVar -> AM SMPConnectedClient
|
||||
connectClient v' = do
|
||||
@@ -699,7 +697,7 @@ smpConnectClient c@AgentClient {smpClients, msgQ, proxySessTs, presetDomains} nm
|
||||
env <- ask
|
||||
liftError (protocolClientError SMP $ B.unpack $ strEncode srv) $ do
|
||||
ts <- readTVarIO proxySessTs
|
||||
smp <- ExceptT $ getProtocolClient g nm tSess cfg presetDomains (Just msgQ) ts $ smpClientDisconnected c tSess env v' prs
|
||||
smp <- ExceptT $ getProtocolClient g nm tSess cfg (presetSMPDomains c) (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 ()
|
||||
@@ -788,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, presetDomains} nm tSess@(_, srv, _) = do
|
||||
getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs} nm tSess@(_, srv, _) = do
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess ntfClients ts)
|
||||
@@ -802,7 +800,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs, pr
|
||||
g <- asks random
|
||||
ts <- readTVarIO proxySessTs
|
||||
liftError' (protocolClientError NTF $ B.unpack $ strEncode srv) $
|
||||
getProtocolClient g nm tSess cfg presetDomains Nothing ts $
|
||||
getProtocolClient g nm tSess cfg [] Nothing ts $
|
||||
clientDisconnected v
|
||||
|
||||
clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
|
||||
@@ -812,7 +810,7 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq, proxySessTs, pr
|
||||
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
|
||||
|
||||
getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
|
||||
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs, presetDomains} tSess@(_, srv, _) = do
|
||||
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq, proxySessTs} tSess@(_, srv, _) = do
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess xftpClients ts)
|
||||
@@ -826,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} presetDomains ts $
|
||||
X.getXFTPClient tSess cfg {xftpNetworkConfig} ts $
|
||||
clientDisconnected v
|
||||
|
||||
clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
|
||||
@@ -868,7 +866,7 @@ newProtocolClient ::
|
||||
ClientVar msg ->
|
||||
AM (Client msg)
|
||||
newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
|
||||
tryAllErrors (connectClient v) >>= \case
|
||||
tryAgentError (connectClient v) >>= \case
|
||||
Right client -> do
|
||||
logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
|
||||
atomically $ putTMVar (sessionVar v) (Right client)
|
||||
@@ -1029,7 +1027,7 @@ getMapLock locks key = TM.lookup key locks >>= maybe newLock pure
|
||||
withClient_ :: forall a v err msg. ProtocolServerClient v err msg => AgentClient -> NetworkRequestMode -> TransportSession msg -> (Client msg -> AM a) -> AM a
|
||||
withClient_ c nm tSess@(_, srv, _) action = do
|
||||
cl <- getProtocolServerClient c nm tSess
|
||||
action cl `catchAllErrors` logServerError
|
||||
action cl `catchAgentError` logServerError
|
||||
where
|
||||
logServerError :: AgentErrorType -> AM a
|
||||
logServerError e = do
|
||||
@@ -1042,7 +1040,7 @@ withProxySession c nm proxySrv_ destSess@(_, destSrv, _) entId cmdStr action = d
|
||||
logServer ("--> " <> proxySrv cl <> " >") c destSrv entId cmdStr
|
||||
case sess_ of
|
||||
Right sess -> do
|
||||
r <- action (cl, sess) `catchAllErrors` logServerError cl
|
||||
r <- action (cl, sess) `catchAgentError` logServerError cl
|
||||
logServer ("<-- " <> proxySrv cl <> " <") c destSrv entId "OK"
|
||||
pure r
|
||||
Left e -> logServerError cl e
|
||||
@@ -1119,7 +1117,7 @@ sendOrProxySMPCommand c nm userId destSrv@ProtocolServer {host = destHosts} conn
|
||||
unknownServer = liftIO $ maybe True (\srvs -> all (`S.notMember` knownHosts srvs) destHosts) <$> TM.lookupIO userId (smpServers c)
|
||||
sendViaProxy :: Maybe SMPServerWithAuth -> SMPTransportSession -> AM (Maybe SMPServer, a)
|
||||
sendViaProxy proxySrv_ destSess@(_, _, connId_) = do
|
||||
r <- tryAllErrors . withProxySession c nm proxySrv_ destSess entId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess@ProxiedRelay {prBasicAuth}) -> do
|
||||
r <- tryAgentError . withProxySession c nm proxySrv_ destSess entId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess@ProxiedRelay {prBasicAuth}) -> do
|
||||
r' <- liftClient SMP (clientServer smp) $ sendCmdViaProxy smp proxySess
|
||||
let proxySrv = protocolClientServer' smp
|
||||
case r' of
|
||||
@@ -1166,7 +1164,7 @@ sendOrProxySMPCommand c nm userId destSrv@ProtocolServer {host = destHosts} conn
|
||||
| otherwise -> throwE e
|
||||
sendDirectly tSess =
|
||||
withLogClient_ c nm tSess (unEntityId entId) ("SEND " <> cmdStr) $ \(SMPConnectedClient smp _) -> do
|
||||
tryAllErrors (liftClient SMP (clientServer smp) $ sendCmdDirectly smp) >>= \case
|
||||
tryAgentError (liftClient SMP (clientServer smp) $ sendCmdDirectly smp) >>= \case
|
||||
Right r -> r <$ atomically (incSMPServerStat c userId destSrv sentDirect)
|
||||
Left e -> throwE e
|
||||
|
||||
@@ -1200,12 +1198,12 @@ protocolClientError protocolError_ host = \case
|
||||
PCEResponseError e -> BROKER host $ RESPONSE $ B.unpack $ smpEncode e
|
||||
PCEUnexpectedResponse e -> BROKER host $ UNEXPECTED $ B.unpack e
|
||||
PCEResponseTimeout -> BROKER host TIMEOUT
|
||||
PCENetworkError e -> BROKER host $ NETWORK e
|
||||
PCENetworkError -> BROKER host NETWORK
|
||||
PCEIncompatibleHost -> BROKER host HOST
|
||||
PCETransportError e -> BROKER host $ TRANSPORT e
|
||||
e@PCECryptoError {} -> INTERNAL $ show e
|
||||
PCEServiceUnavailable {} -> BROKER host NO_SERVICE
|
||||
PCEIOError e -> BROKER host $ NETWORK $ NEConnectError $ E.displayException e
|
||||
PCEIOError {} -> BROKER host NETWORK
|
||||
|
||||
data ProtocolTestStep
|
||||
= TSConnect
|
||||
@@ -1229,7 +1227,7 @@ data ProtocolTestFailure = ProtocolTestFailure
|
||||
deriving (Eq, Show)
|
||||
|
||||
runSMPServerTest :: AgentClient -> NetworkRequestMode -> UserId -> SMPServerWithAuth -> AM' (Maybe ProtocolTestFailure)
|
||||
runSMPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth srv auth) = do
|
||||
runSMPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
|
||||
cfg <- getClientConfig c smpCfg
|
||||
C.AuthAlg ra <- asks $ rcvAuthAlg . config
|
||||
C.AuthAlg sa <- asks $ sndAuthAlg . config
|
||||
@@ -1237,7 +1235,7 @@ runSMPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth sr
|
||||
liftIO $ do
|
||||
let tSess = (userId, srv, Nothing)
|
||||
ts <- readTVarIO $ proxySessTs c
|
||||
getProtocolClient g nm tSess cfg presetDomains Nothing ts (\_ -> pure ()) >>= \case
|
||||
getProtocolClient g nm tSess cfg (presetSMPDomains c) Nothing ts (\_ -> pure ()) >>= \case
|
||||
Right smp -> do
|
||||
rKeys@(_, rpKey) <- atomically $ C.generateAuthKeyPair ra g
|
||||
(sKey, spKey) <- atomically $ C.generateAuthKeyPair sa g
|
||||
@@ -1258,7 +1256,7 @@ runSMPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth sr
|
||||
testErr step = ProtocolTestFailure step . protocolClientError SMP addr
|
||||
|
||||
runXFTPServerTest :: AgentClient -> NetworkRequestMode -> UserId -> XFTPServerWithAuth -> AM' (Maybe ProtocolTestFailure)
|
||||
runXFTPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth srv auth) = do
|
||||
runXFTPServerTest c nm userId (ProtoServerWithAuth srv auth) = do
|
||||
cfg <- asks $ xftpCfg . config
|
||||
g <- asks random
|
||||
xftpNetworkConfig <- getNetworkConfig c
|
||||
@@ -1268,7 +1266,7 @@ runXFTPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth s
|
||||
liftIO $ do
|
||||
let tSess = (userId, srv, Nothing)
|
||||
ts <- readTVarIO $ proxySessTs c
|
||||
X.getXFTPClient tSess cfg {xftpNetworkConfig} presetDomains ts (\_ -> pure ()) >>= \case
|
||||
X.getXFTPClient tSess cfg {xftpNetworkConfig} ts (\_ -> pure ()) >>= \case
|
||||
Right xftp -> withTestChunk filePath $ do
|
||||
(sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
|
||||
(rcvKey, rpKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
|
||||
@@ -1306,14 +1304,14 @@ runXFTPServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth s
|
||||
createTestChunk fp = B.writeFile fp =<< atomically . C.randomBytes chSize =<< C.newRandom
|
||||
|
||||
runNTFServerTest :: AgentClient -> NetworkRequestMode -> UserId -> NtfServerWithAuth -> AM' (Maybe ProtocolTestFailure)
|
||||
runNTFServerTest c@AgentClient {presetDomains} nm userId (ProtoServerWithAuth srv _) = do
|
||||
runNTFServerTest c 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 presetDomains Nothing ts (\_ -> pure ()) >>= \case
|
||||
getProtocolClient g nm tSess cfg [] Nothing ts (\_ -> pure ()) >>= \case
|
||||
Right ntf -> do
|
||||
(nKey, npKey) <- atomically $ C.generateAuthKeyPair a g
|
||||
(dhKey, _) <- atomically $ C.generateKeyPair g
|
||||
@@ -1479,7 +1477,7 @@ temporaryAgentError = \case
|
||||
_ -> False
|
||||
where
|
||||
tempBrokerError = \case
|
||||
NETWORK _ -> True
|
||||
NETWORK -> True
|
||||
TIMEOUT -> True
|
||||
_ -> False
|
||||
|
||||
@@ -1519,7 +1517,7 @@ subscribeQueues c qs = do
|
||||
subscribeQueues_ env session smp qs' = do
|
||||
let (userId, srv, _) = transportSession' smp
|
||||
atomically $ incSMPServerStat' c userId srv connSubAttempts $ length qs'
|
||||
rs <- sendBatch (\smp' _ -> subscribeSMPQueues smp') smp NRMBackground qs'
|
||||
rs <- sendBatch (\smp' _ -> subscribeSMPQueues smp') smp NRMBackground qs'
|
||||
active <-
|
||||
atomically $
|
||||
ifM
|
||||
@@ -1530,8 +1528,7 @@ subscribeQueues c qs = do
|
||||
then when (hasTempErrors rs) resubscribe $> rs
|
||||
else do
|
||||
logWarn "subcription batch result for replaced SMP client, resubscribing"
|
||||
-- TODO we probably use PCENetworkError here instead of the original error, so it becomes temporary.
|
||||
resubscribe $> L.map (second $ Left . PCENetworkError . NESubscribeError . show) rs
|
||||
resubscribe $> L.map (second $ \_ -> Left PCENetworkError) rs
|
||||
where
|
||||
tSess = transportSession' smp
|
||||
sessId = sessionId $ thParams smp
|
||||
@@ -1565,7 +1562,7 @@ sendTSessionBatches statCmd toRQ action c nm qs =
|
||||
in M.alter (Just . maybe [q] (q <|)) tSess m
|
||||
sendClientBatch :: (SMPTransportSession, NonEmpty q) -> AM' (BatchResponses q AgentErrorType r)
|
||||
sendClientBatch (tSess@(_, srv, _), qs') =
|
||||
tryAllErrors' (getSMPServerClient c nm tSess) >>= \case
|
||||
tryAgentError' (getSMPServerClient c nm tSess) >>= \case
|
||||
Left e -> pure $ L.map (,Left e) qs'
|
||||
Right (SMPConnectedClient smp _) -> liftIO $ do
|
||||
logServer' "-->" c srv (bshow (length qs') <> " queues") statCmd
|
||||
@@ -1870,7 +1867,7 @@ withNtfBatch ::
|
||||
AM' (NonEmpty (Either AgentErrorType r))
|
||||
withNtfBatch cmdStr action c NtfToken {ntfServer, ntfPrivKey} subs = do
|
||||
let tSess = (0, ntfServer, Nothing)
|
||||
tryAllErrors' (getNtfServerClient c NRMBackground tSess) >>= \case
|
||||
tryAgentError' (getNtfServerClient c NRMBackground tSess) >>= \case
|
||||
Left e -> pure $ L.map (\_ -> Left e) subs
|
||||
Right ntf -> liftIO $ do
|
||||
logServer' "-->" c ntfServer (bshow (length subs) <> " subscriptions") cmdStr
|
||||
@@ -1971,42 +1968,40 @@ waitForWork = void . atomically . readTMVar
|
||||
{-# INLINE waitForWork #-}
|
||||
|
||||
withWork :: AgentClient -> TMVar () -> (DB.Connection -> IO (Either StoreError (Maybe a))) -> (a -> AM ()) -> AM ()
|
||||
withWork c doWork = withWork_ c doWork . withStore' c
|
||||
{-# INLINE withWork #-}
|
||||
|
||||
withWork_ :: (AnyStoreError e', MonadIO m) => AgentClient -> TMVar () -> ExceptT e m (Either e' (Maybe a)) -> (a -> ExceptT e m ()) -> ExceptT e m ()
|
||||
withWork_ c doWork getWork action =
|
||||
getWork >>= \case
|
||||
withWork c doWork getWork action =
|
||||
withStore' c getWork >>= \case
|
||||
Right (Just r) -> action r
|
||||
Right Nothing -> noWork
|
||||
-- worker is stopped here (noWork) because the next iteration is likely to produce the same result
|
||||
Left e
|
||||
| isWorkItemError e -> noWork >> notifyErr (CRITICAL False) e
|
||||
| otherwise -> notifyErr INTERNAL e
|
||||
Left e@SEWorkItemError {} -> noWork >> notifyErr (CRITICAL False) e
|
||||
Left e -> notifyErr INTERNAL e
|
||||
where
|
||||
noWork = liftIO $ noWorkToDo doWork
|
||||
notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
|
||||
withWorkItems :: (AnyStoreError e', MonadIO m) => AgentClient -> TMVar () -> ExceptT e m (Either e' [Either e' a]) -> (NonEmpty a -> ExceptT e m ()) -> ExceptT e m ()
|
||||
withWorkItems :: AgentClient -> TMVar () -> (DB.Connection -> IO (Either StoreError [Either StoreError a])) -> (NonEmpty a -> AM ()) -> AM ()
|
||||
withWorkItems c doWork getWork action = do
|
||||
getWork >>= \case
|
||||
withStore' c getWork >>= \case
|
||||
Right [] -> noWork
|
||||
Right rs -> do
|
||||
let (errs, items) = partitionEithers rs
|
||||
case L.nonEmpty items of
|
||||
Just items' -> action items'
|
||||
Nothing -> do
|
||||
let criticalErr = find isWorkItemError errs
|
||||
let criticalErr = find workItemError errs
|
||||
forM_ criticalErr $ \err -> do
|
||||
notifyErr (CRITICAL False) err
|
||||
when (all isWorkItemError errs) noWork
|
||||
when (all workItemError errs) noWork
|
||||
unless (null errs) $
|
||||
atomically $
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAENone $ ERRS $ map (\e -> ("", INTERNAL $ show e)) errs)
|
||||
Left e
|
||||
| isWorkItemError e -> noWork >> notifyErr (CRITICAL False) e
|
||||
| workItemError e -> noWork >> notifyErr (CRITICAL False) e
|
||||
| otherwise -> notifyErr INTERNAL e
|
||||
where
|
||||
workItemError = \case
|
||||
SEWorkItemError {} -> True
|
||||
_ -> False
|
||||
noWork = liftIO $ noWorkToDo doWork
|
||||
notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
|
||||
|
||||
@@ -27,6 +27,11 @@ module Simplex.Messaging.Agent.Env.SQLite
|
||||
serverHosts,
|
||||
defaultAgentConfig,
|
||||
defaultReconnectInterval,
|
||||
tryAgentError,
|
||||
tryAgentError',
|
||||
catchAgentError,
|
||||
catchAgentError',
|
||||
agentFinally,
|
||||
Env (..),
|
||||
newSMPAgentEnv,
|
||||
createAgentStore,
|
||||
@@ -40,6 +45,7 @@ module Simplex.Messaging.Agent.Env.SQLite
|
||||
where
|
||||
|
||||
import Control.Concurrent (ThreadId)
|
||||
import Control.Exception (BlockedIndefinitelyOnSTM (..), SomeException, fromException)
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.IO.Unlift
|
||||
import Control.Monad.Reader
|
||||
@@ -77,6 +83,7 @@ import Simplex.Messaging.TMap (TMap)
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
import Simplex.Messaging.Transport (SMPVersion)
|
||||
import Simplex.Messaging.Transport.Client (TransportHost)
|
||||
import Simplex.Messaging.Util (allFinally, catchAllErrors, catchAllErrors', tryAllErrors, tryAllErrors')
|
||||
import System.Mem.Weak (Weak)
|
||||
import System.Random (StdGen, newStdGen)
|
||||
import UnliftIO.STM
|
||||
@@ -305,6 +312,33 @@ newXFTPAgent = do
|
||||
xftpDelWorkers <- TM.emptyIO
|
||||
pure XFTPAgent {xftpWorkDir, xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers}
|
||||
|
||||
tryAgentError :: AM a -> AM (Either AgentErrorType a)
|
||||
tryAgentError = tryAllErrors mkInternal
|
||||
{-# INLINE tryAgentError #-}
|
||||
|
||||
-- unlike runExceptT, this ensures we catch IO exceptions as well
|
||||
tryAgentError' :: AM a -> AM' (Either AgentErrorType a)
|
||||
tryAgentError' = tryAllErrors' mkInternal
|
||||
{-# INLINE tryAgentError' #-}
|
||||
|
||||
catchAgentError :: AM a -> (AgentErrorType -> AM a) -> AM a
|
||||
catchAgentError = catchAllErrors mkInternal
|
||||
{-# INLINE catchAgentError #-}
|
||||
|
||||
catchAgentError' :: AM a -> (AgentErrorType -> AM' a) -> AM' a
|
||||
catchAgentError' = catchAllErrors' mkInternal
|
||||
{-# INLINE catchAgentError' #-}
|
||||
|
||||
agentFinally :: AM a -> AM b -> AM a
|
||||
agentFinally = allFinally mkInternal
|
||||
{-# INLINE agentFinally #-}
|
||||
|
||||
mkInternal :: SomeException -> AgentErrorType
|
||||
mkInternal e = case fromException e of
|
||||
Just BlockedIndefinitelyOnSTM -> CRITICAL True "Thread blocked indefinitely in STM transaction"
|
||||
_ -> INTERNAL $ show e
|
||||
{-# INLINE mkInternal #-}
|
||||
|
||||
data Worker = Worker
|
||||
{ workerId :: Int,
|
||||
doWork :: TMVar (),
|
||||
|
||||
@@ -52,7 +52,7 @@ import Simplex.Messaging.Notifications.Protocol
|
||||
import Simplex.Messaging.Notifications.Types
|
||||
import Simplex.Messaging.Protocol (NtfServer, sameSrvAddr)
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Util (catchAllErrors, diffToMicroseconds, threadDelay', tryAllErrors, tshow, whenM)
|
||||
import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, whenM)
|
||||
import System.Random (randomR)
|
||||
import UnliftIO
|
||||
import UnliftIO.Concurrent (forkIO)
|
||||
@@ -217,7 +217,7 @@ runNtfWorker c srv Worker {doWork} =
|
||||
runNtfOperation :: AM ()
|
||||
runNtfOperation = do
|
||||
ntfBatchSize <- asks $ ntfBatchSize . config
|
||||
withWorkItems c doWork (withStore' c $ \db -> getNextNtfSubNTFActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
withWorkItems c doWork (\db -> getNextNtfSubNTFActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
logInfo $ "runNtfWorker - length nextSubs = " <> tshow (length nextSubs)
|
||||
currTs <- liftIO getCurrentTime
|
||||
let (creates, checks, deletes, rotates) = splitActions currTs nextSubs
|
||||
@@ -357,7 +357,7 @@ runNtfWorker c srv Worker {doWork} =
|
||||
runCatching :: (NtfSubscription -> AM (Maybe NtfSubscription)) -> NtfSubscription -> AM' (Maybe NtfSubscription)
|
||||
runCatching action sub@NtfSubscription {connId} =
|
||||
fromRight Nothing
|
||||
<$> runExceptT (action sub `catchAllErrors` \e -> workerInternalError c connId (show e) $> Nothing)
|
||||
<$> runExceptT (action sub `catchAgentError` \e -> workerInternalError c connId (show e) $> Nothing)
|
||||
-- deleteNtfSub is only used in NSADelete and NSARotate, so also deprecated
|
||||
deleteNtfSub :: NtfSubscription -> AM () -> AM (Maybe NtfSubscription)
|
||||
deleteNtfSub sub@NtfSubscription {userId, ntfSubId} continue = case ntfSubId of
|
||||
@@ -365,7 +365,7 @@ runNtfWorker c srv Worker {doWork} =
|
||||
lift getNtfToken >>= \case
|
||||
Just tkn@NtfToken {ntfServer} -> do
|
||||
atomically $ incNtfServerStat c userId ntfServer ntfDelAttempts
|
||||
tryAllErrors (agentNtfDeleteSubscription c nSubId tkn) >>= \case
|
||||
tryAgentError (agentNtfDeleteSubscription c nSubId tkn) >>= \case
|
||||
Right _ -> do
|
||||
atomically $ incNtfServerStat c userId ntfServer ntfDeleted
|
||||
continue'
|
||||
@@ -385,7 +385,7 @@ runNtfSMPWorker c srv Worker {doWork} = forever $ do
|
||||
runNtfSMPOperation :: AM ()
|
||||
runNtfSMPOperation = do
|
||||
ntfBatchSize <- asks $ ntfBatchSize . config
|
||||
withWorkItems c doWork (withStore' c $ \db -> getNextNtfSubSMPActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
withWorkItems c doWork (\db -> getNextNtfSubSMPActions db srv ntfBatchSize) $ \nextSubs -> do
|
||||
logInfo $ "runNtfSMPWorker - length nextSubs = " <> tshow (length nextSubs)
|
||||
let (creates, deletes) = splitActions nextSubs
|
||||
retrySubActions c creates createNotifierKeys
|
||||
@@ -567,7 +567,7 @@ runNtfTknDelWorker c srv Worker {doWork} =
|
||||
withRetryInterval ri $ \_ loop -> do
|
||||
liftIO $ waitWhileSuspended c
|
||||
liftIO $ waitForUserNetwork c
|
||||
processTknToDelete nextTknToDelete `catchAllErrors` retryTmpError loop nextTknToDelete
|
||||
processTknToDelete nextTknToDelete `catchAgentError` retryTmpError loop nextTknToDelete
|
||||
retryTmpError :: AM () -> NtfTokenToDelete -> AgentErrorType -> AM ()
|
||||
retryTmpError loop (tknDbId, _, _) e = do
|
||||
logError $ "ntf tkn del error: " <> tshow e
|
||||
|
||||
@@ -173,7 +173,6 @@ module Simplex.Messaging.Agent.Protocol
|
||||
where
|
||||
|
||||
import Control.Applicative (optional, (<|>))
|
||||
import Control.Exception (BlockedIndefinitelyOnSTM (..), fromException)
|
||||
import Data.Aeson (FromJSON (..), ToJSON (..), Value (..), (.:), (.:?))
|
||||
import qualified Data.Aeson as J'
|
||||
import qualified Data.Aeson.Encoding as JE
|
||||
@@ -1130,7 +1129,7 @@ instance StrEncoding AConnectionRequestUri where
|
||||
|
||||
connReqUriP :: Maybe ServiceScheme -> Parser AConnectionRequestUri
|
||||
connReqUriP overrideScheme = do
|
||||
crScheme <- (`fromMaybe` overrideScheme) <$> strP -- always parse, but use the passed one if any
|
||||
crScheme <- (`fromMaybe` overrideScheme) <$> strP
|
||||
crMode <- A.char '/' *> crModeP <* optional (A.char '/') <* "#/?"
|
||||
query <- strP
|
||||
aVRange <- queryParam "v" query
|
||||
@@ -1446,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 | CCTRelay deriving (Eq, Show)
|
||||
data ContactConnType = CCTContact | CCTChannel | CCTGroup deriving (Eq, Show)
|
||||
|
||||
data AConnShortLink = forall m. ConnectionModeI m => ACSL (SConnectionMode m) (ConnShortLink m)
|
||||
|
||||
@@ -1594,7 +1593,6 @@ ctTypeP = \case
|
||||
'A' -> pure CCTContact
|
||||
'C' -> pure CCTChannel
|
||||
'G' -> pure CCTGroup
|
||||
'R' -> pure CCTRelay
|
||||
_ -> fail "unknown contact address type"
|
||||
{-# INLINE ctTypeP #-}
|
||||
|
||||
@@ -1603,7 +1601,6 @@ 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.
|
||||
@@ -1867,12 +1864,6 @@ data AgentErrorType
|
||||
INACTIVE
|
||||
deriving (Eq, Show, Exception)
|
||||
|
||||
instance AnyError AgentErrorType where
|
||||
fromSomeException e = case fromException e of
|
||||
Just BlockedIndefinitelyOnSTM -> CRITICAL True "Thread blocked indefinitely in STM transaction"
|
||||
_ -> INTERNAL $ show e
|
||||
{-# INLINE fromSomeException #-}
|
||||
|
||||
-- | SMP agent protocol command or response error.
|
||||
data CommandErrorType
|
||||
= -- | command is prohibited in this context
|
||||
|
||||
@@ -29,7 +29,6 @@ import Data.Time (UTCTime)
|
||||
import Data.Type.Equality
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.RetryInterval (RI2State)
|
||||
import Simplex.Messaging.Agent.Store.Entity
|
||||
import Simplex.Messaging.Agent.Store.Common
|
||||
import Simplex.Messaging.Agent.Store.Interface (createDBStore)
|
||||
import Simplex.Messaging.Agent.Store.Migrations.App (appMigrations)
|
||||
@@ -53,7 +52,7 @@ import Simplex.Messaging.Protocol
|
||||
VersionSMPC,
|
||||
)
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Util (AnyError (..), bshow)
|
||||
import Simplex.Messaging.Agent.Store.Entity
|
||||
|
||||
createStore :: DBOpts -> MigrationConfirmation -> IO (Either MigrationError DBStore)
|
||||
createStore dbOpts = createDBStore dbOpts appMigrations
|
||||
@@ -673,7 +672,7 @@ data StoreError
|
||||
| -- | Invitation not found
|
||||
SEInvitationNotFound String InvitationId
|
||||
| -- | Message not found
|
||||
SEMsgNotFound String
|
||||
SEMsgNotFound
|
||||
| -- | Command not found
|
||||
SECmdNotFound
|
||||
| -- | Currently not used. The intention was to pass current expected queue status in methods,
|
||||
@@ -693,20 +692,7 @@ data StoreError
|
||||
| -- | XFTP Deleted snd chunk replica not found.
|
||||
SEDeletedSndChunkReplicaNotFound
|
||||
| -- | Error when reading work item that suspends worker - do not use!
|
||||
SEWorkItemError {errContext :: String}
|
||||
SEWorkItemError ByteString
|
||||
| -- | Servers stats not found.
|
||||
SEServersStatsNotFound
|
||||
deriving (Eq, Show, Exception)
|
||||
|
||||
instance AnyError StoreError where
|
||||
fromSomeException = SEInternal . bshow
|
||||
|
||||
class (Show e, AnyError e) => AnyStoreError e where
|
||||
isWorkItemError :: e -> Bool
|
||||
mkWorkItemError :: String -> e
|
||||
|
||||
instance AnyStoreError StoreError where
|
||||
isWorkItemError = \case
|
||||
SEWorkItemError {} -> True
|
||||
_ -> False
|
||||
mkWorkItemError errContext = SEWorkItemError {errContext}
|
||||
|
||||
@@ -237,8 +237,6 @@ module Simplex.Messaging.Agent.Store.AgentStore
|
||||
firstRow',
|
||||
maybeFirstRow,
|
||||
fromOnlyBI,
|
||||
getWorkItem,
|
||||
getWorkItems,
|
||||
)
|
||||
where
|
||||
|
||||
@@ -893,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 "getSndMsgViaRcpt") $
|
||||
firstRow toSndMsg SEMsgNotFound $
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
@@ -968,25 +966,28 @@ getPendingQueueMsg db connId SndQueue {dbQueueId} =
|
||||
_ -> Left $ SEInternal "unexpected snd msg data"
|
||||
markMsgFailed msgId = DB.execute db "UPDATE snd_message_deliveries SET failed = 1 WHERE conn_id = ? AND internal_id = ?" (connId, msgId)
|
||||
|
||||
getWorkItem :: (Show i, AnyStoreError e) => String -> IO (Maybe i) -> (i -> IO (Either e a)) -> (i -> IO ()) -> IO (Either e (Maybe a))
|
||||
getWorkItem :: Show i => ByteString -> IO (Maybe i) -> (i -> IO (Either StoreError a)) -> (i -> IO ()) -> IO (Either StoreError (Maybe a))
|
||||
getWorkItem itemName getId getItem markFailed =
|
||||
runExceptT $ handleWrkErr itemName "getId" getId >>= mapM (tryGetItem itemName getItem markFailed)
|
||||
|
||||
getWorkItems :: (Show i, AnyStoreError e) => String -> IO [i] -> (i -> IO (Either e a)) -> (i -> IO ()) -> IO (Either e [Either e a])
|
||||
getWorkItems :: Show i => ByteString -> IO [i] -> (i -> IO (Either StoreError a)) -> (i -> IO ()) -> IO (Either StoreError [Either StoreError a])
|
||||
getWorkItems itemName getIds getItem markFailed =
|
||||
runExceptT $ handleWrkErr itemName "getIds" getIds >>= mapM (tryE . tryGetItem itemName getItem markFailed)
|
||||
|
||||
tryGetItem :: (Show i, AnyStoreError e) => String -> (i -> IO (Either e a)) -> (i -> IO ()) -> i -> ExceptT e IO a
|
||||
tryGetItem itemName getItem markFailed itemId = ExceptT (getItem itemId) `catchAllErrors` \e -> mark >> throwE e
|
||||
tryGetItem :: Show i => ByteString -> (i -> IO (Either StoreError a)) -> (i -> IO ()) -> i -> ExceptT StoreError IO a
|
||||
tryGetItem itemName getItem markFailed itemId = ExceptT (getItem itemId) `catchStoreError` \e -> mark >> throwE e
|
||||
where
|
||||
mark = handleWrkErr itemName ("markFailed ID " <> show itemId) $ markFailed itemId
|
||||
mark = handleWrkErr itemName ("markFailed ID " <> bshow itemId) $ markFailed itemId
|
||||
|
||||
catchStoreError :: ExceptT StoreError IO a -> (StoreError -> ExceptT StoreError IO a) -> ExceptT StoreError IO a
|
||||
catchStoreError = catchAllErrors (SEInternal . bshow)
|
||||
|
||||
-- Errors caught by this function will suspend worker as if there is no more work,
|
||||
handleWrkErr :: forall e a. AnyStoreError e => String -> String -> IO a -> ExceptT e IO a
|
||||
handleWrkErr :: ByteString -> ByteString -> IO a -> ExceptT StoreError IO a
|
||||
handleWrkErr itemName opName action = ExceptT $ first mkError <$> E.try action
|
||||
where
|
||||
mkError :: E.SomeException -> e
|
||||
mkError e = mkWorkItemError $ itemName <> " " <> opName <> " error: " <> show e
|
||||
mkError :: E.SomeException -> StoreError
|
||||
mkError e = SEWorkItemError $ itemName <> " " <> opName <> " error: " <> bshow e
|
||||
|
||||
updatePendingMsgRIState :: DB.Connection -> ConnId -> InternalId -> RI2State -> IO ()
|
||||
updatePendingMsgRIState db connId msgId RI2State {slowInterval, fastInterval} =
|
||||
@@ -1026,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 "setMsgUserAck") $
|
||||
ExceptT . firstRow id SEMsgNotFound $
|
||||
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)
|
||||
@@ -1034,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 "getRcvMsg") $
|
||||
firstRow toRcvMsg SEMsgNotFound $
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
@@ -1084,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 "getRcvMsgBrokerTs") $
|
||||
firstRow fromOnly SEMsgNotFound $
|
||||
DB.query db "SELECT broker_ts FROM rcv_messages WHERE conn_id = ? AND broker_id = ?" (connId, Binary msgId)
|
||||
|
||||
deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
|
||||
|
||||
@@ -11,21 +11,23 @@ module Simplex.Messaging.Agent.Store.Postgres.DB
|
||||
execute,
|
||||
execute_,
|
||||
executeMany,
|
||||
PSQL.query,
|
||||
PSQL.query_,
|
||||
query,
|
||||
query_,
|
||||
blobFieldDecoder,
|
||||
fromTextField_,
|
||||
)
|
||||
where
|
||||
|
||||
import qualified Control.Exception as E
|
||||
import Control.Monad (void)
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Int (Int64)
|
||||
import Data.Text (Text)
|
||||
import Data.Text.Encoding (decodeUtf8)
|
||||
import Data.Typeable (Typeable)
|
||||
import Data.Word (Word16, Word32)
|
||||
import Database.PostgreSQL.Simple (ResultError (..))
|
||||
import Database.PostgreSQL.Simple (ResultError (..), SqlError (..))
|
||||
import qualified Database.PostgreSQL.Simple as PSQL
|
||||
import Database.PostgreSQL.Simple.FromField (Field (..), FieldParser, FromField (..), returnError)
|
||||
import Database.PostgreSQL.Simple.ToField (ToField (..))
|
||||
@@ -42,17 +44,49 @@ instance ToField BoolInt where
|
||||
{-# INLINE toField #-}
|
||||
|
||||
execute :: PSQL.ToRow q => PSQL.Connection -> PSQL.Query -> q -> IO ()
|
||||
execute db q qs = void $ PSQL.execute db q qs
|
||||
execute db q qs = withLoggedErrors q $ void $ PSQL.execute db q qs
|
||||
{-# INLINE execute #-}
|
||||
|
||||
execute_ :: PSQL.Connection -> PSQL.Query -> IO ()
|
||||
execute_ db q = void $ PSQL.execute_ db q
|
||||
execute_ db q = withLoggedErrors q $ void $ PSQL.execute_ db q
|
||||
{-# INLINE execute_ #-}
|
||||
|
||||
executeMany :: PSQL.ToRow q => PSQL.Connection -> PSQL.Query -> [q] -> IO ()
|
||||
executeMany db q qs = void $ PSQL.executeMany db q qs
|
||||
executeMany db q qs = withLoggedErrors q $ void $ PSQL.executeMany db q qs
|
||||
{-# INLINE executeMany #-}
|
||||
|
||||
query :: (PSQL.ToRow q, PSQL.FromRow r) => PSQL.Connection -> PSQL.Query -> q -> IO [r]
|
||||
query db q qs = withLoggedErrors q $ PSQL.query db q qs
|
||||
{-# INLINE query #-}
|
||||
|
||||
query_ :: PSQL.FromRow r => PSQL.Connection -> PSQL.Query -> IO [r]
|
||||
query_ db q = withLoggedErrors q $ PSQL.query_ db q
|
||||
{-# INLINE query_ #-}
|
||||
|
||||
withLoggedErrors :: Show q => q -> IO a -> IO a
|
||||
withLoggedErrors q action =
|
||||
action
|
||||
`E.catch` (\(e :: SqlError) -> logSqlErrorAndRethrow e)
|
||||
`E.catch`
|
||||
(\(e :: E.SomeException) ->
|
||||
case E.fromException e :: Maybe SqlError of
|
||||
Just sqlErr -> E.throwIO sqlErr -- rethrow SqlError without logging
|
||||
Nothing -> logGenericErrorAndRethrow e
|
||||
)
|
||||
where
|
||||
logSqlErrorAndRethrow :: SqlError -> IO a
|
||||
logSqlErrorAndRethrow e = do
|
||||
putStrLn "Caught SqlError"
|
||||
putStrLn $ "Message: " <> B.unpack (sqlErrorMsg e)
|
||||
putStrLn $ "SQL State: " <> B.unpack (sqlState e)
|
||||
putStrLn $ "Query: " <> show q
|
||||
E.throwIO e
|
||||
logGenericErrorAndRethrow :: E.SomeException -> IO a
|
||||
logGenericErrorAndRethrow e = do
|
||||
putStrLn $ "Caught generic exception: " <> show e
|
||||
putStrLn $ "Query: " <> show q
|
||||
E.throwIO e
|
||||
|
||||
-- orphan instances
|
||||
|
||||
-- used in FileSize
|
||||
|
||||
@@ -52,7 +52,7 @@ import Simplex.Messaging.Util (diffToMicroseconds, tshow)
|
||||
newtype BoolInt = BI {unBI :: Bool}
|
||||
deriving newtype (FromField, ToField)
|
||||
|
||||
newtype Binary a = Binary {fromBinary :: a}
|
||||
newtype Binary = Binary {fromBinary :: ByteString}
|
||||
deriving newtype (FromField, ToField)
|
||||
|
||||
data Connection = Connection
|
||||
|
||||
@@ -38,7 +38,6 @@ module Simplex.Messaging.Client
|
||||
protocolClientServer',
|
||||
transportHost',
|
||||
transportSession',
|
||||
useWebPort,
|
||||
|
||||
-- * SMP protocol command functions
|
||||
createSMPQueue,
|
||||
@@ -161,7 +160,6 @@ 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
|
||||
@@ -562,7 +560,7 @@ getProtocolClient g nm transportSession@(_, srv, _) cfg@ProtocolClientConfig {qS
|
||||
`catch` \(e :: IOException) -> pure . Left $ PCEIOError e
|
||||
Left e -> pure $ Left e
|
||||
where
|
||||
NetworkConfig {tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
|
||||
NetworkConfig {smpWebPortServers, tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
|
||||
mkProtocolClient :: TransportHost -> UTCTime -> IO (PClient v err msg)
|
||||
mkProtocolClient transportHost ts = do
|
||||
connected <- newTVarIO False
|
||||
@@ -593,30 +591,30 @@ 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 useALPN) {clientCredentials = serviceCreds <$> serviceCredentials}
|
||||
let tcConfig = (transportClientConfig networkConfig nm useHost useSNI clientALPN) {clientCredentials = serviceCreds <$> serviceCredentials}
|
||||
socksCreds = clientSocksCredentials networkConfig proxySessTs transportSession
|
||||
tId <-
|
||||
runTransportClient tcConfig socksCreds useHost port' (Just $ keyHash srv) (client t c cVar)
|
||||
`forkFinally` \r ->
|
||||
let err = either toNetworkError (const NEFailedError) r
|
||||
in void $ atomically $ tryPutTMVar cVar $ Left $ PCENetworkError err
|
||||
`forkFinally` \_ -> void (atomically . tryPutTMVar cVar $ Left PCENetworkError)
|
||||
c_ <- netTimeoutInt tcpConnectTimeout nm `timeout` atomically (takeTMVar cVar)
|
||||
case c_ of
|
||||
Just (Right c') -> mkWeakThreadId tId >>= \tId' -> pure $ Right c' {action = Just tId'}
|
||||
Just (Left e) -> pure $ Left e
|
||||
Nothing -> killThread tId $> Left (PCENetworkError NETimeoutError)
|
||||
Nothing -> killThread tId $> Left PCENetworkError
|
||||
|
||||
useTransport :: (ServiceName, ATransport 'TClient)
|
||||
useTransport = case port srv of
|
||||
"" -> case protocolTypeI @(ProtoType msg) of
|
||||
SPSMP | web -> ("443", transport @TLS)
|
||||
SPSMP | smpWebPort -> ("443", transport @TLS)
|
||||
_ -> defaultTransport cfg
|
||||
p -> (p, transport @TLS)
|
||||
|
||||
useALPN :: Maybe [ALPN]
|
||||
useALPN = if web then Just [httpALPN11] else clientALPN
|
||||
|
||||
web = useWebPort networkConfig presetDomains srv
|
||||
where
|
||||
smpWebPort = case smpWebPortServers of
|
||||
SWPAll -> True
|
||||
SWPPreset -> case srv of
|
||||
ProtocolServer {host = THDomainName h :| _} -> any (`isSuffixOf` h) presetDomains
|
||||
_ -> False
|
||||
SWPOff -> False
|
||||
|
||||
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
|
||||
@@ -711,14 +709,6 @@ 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
|
||||
|
||||
@@ -745,7 +735,7 @@ data ProtocolClientError err
|
||||
PCEResponseTimeout
|
||||
| -- | Failure to establish TCP connection.
|
||||
-- Forwarded to the agent client as `ERR BROKER NETWORK`.
|
||||
PCENetworkError NetworkError
|
||||
PCENetworkError
|
||||
| -- | No host compatible with network configuration
|
||||
PCEIncompatibleHost
|
||||
| -- | Service is unavailable for command that requires service connection
|
||||
@@ -763,7 +753,7 @@ type SMPClientError = ProtocolClientError ErrorType
|
||||
|
||||
temporaryClientError :: ProtocolClientError err -> Bool
|
||||
temporaryClientError = \case
|
||||
PCENetworkError _ -> True
|
||||
PCENetworkError -> True
|
||||
PCEResponseTimeout -> True
|
||||
PCEIOError _ -> True
|
||||
_ -> False
|
||||
@@ -784,7 +774,7 @@ smpProxyError = \case
|
||||
PCEResponseError e -> PROXY $ BROKER $ RESPONSE $ B.unpack $ strEncode e
|
||||
PCEUnexpectedResponse e -> PROXY $ BROKER $ UNEXPECTED $ B.unpack e
|
||||
PCEResponseTimeout -> PROXY $ BROKER TIMEOUT
|
||||
PCENetworkError e -> PROXY $ BROKER $ NETWORK e
|
||||
PCENetworkError -> PROXY $ BROKER NETWORK
|
||||
PCEIncompatibleHost -> PROXY $ BROKER HOST
|
||||
PCEServiceUnavailable -> PROXY $ BROKER $ NO_SERVICE -- for completeness, it cannot happen.
|
||||
PCETransportError t -> PROXY $ BROKER $ TRANSPORT t
|
||||
|
||||
@@ -391,7 +391,7 @@ withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPE
|
||||
where
|
||||
logSMPError :: SMPClientError -> ExceptT SMPClientError IO a
|
||||
logSMPError e = do
|
||||
logInfo $ "SMP error (" <> safeDecodeUtf8 (strEncode srv) <> "): " <> tshow e
|
||||
logInfo $ "SMP error (" <> safeDecodeUtf8 (strEncode $ host srv) <> "): " <> tshow e
|
||||
throwE e
|
||||
|
||||
subscribeQueuesNtfs :: SMPClientAgent 'NotifierService -> SMPServer -> NonEmpty (NotifierId, NtfPrivateAuthKey) -> IO ()
|
||||
|
||||
@@ -613,7 +613,7 @@ ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} =
|
||||
PCEIncompatibleHost -> Just $ NSErr "IncompatibleHost"
|
||||
PCEServiceUnavailable -> Just NSService -- this error should not happen on individual subscriptions
|
||||
PCEResponseTimeout -> Nothing
|
||||
PCENetworkError _ -> Nothing
|
||||
PCENetworkError -> Nothing
|
||||
PCEIOError _ -> Nothing
|
||||
where
|
||||
-- Note on moving to PostgreSQL: the idea of logging errors without e is removed here
|
||||
|
||||
@@ -46,9 +46,8 @@ 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)
|
||||
import Simplex.Messaging.Transport (ASrvTransport, simplexMQVersion)
|
||||
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)
|
||||
@@ -137,7 +136,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" <> simplexmqVersionCommit
|
||||
serverVersion = "SMP notifications server v" <> simplexMQVersion
|
||||
defaultServerPort = "443"
|
||||
executableName = "ntf-server"
|
||||
storeLogFilePath = combine logPath "ntf-server-store.log"
|
||||
@@ -168,9 +167,6 @@ 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\
|
||||
@@ -284,7 +280,7 @@ ntfServerCLI cfgPath logPath =
|
||||
transportConfig =
|
||||
mkTransportServerConfig
|
||||
(fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini)
|
||||
(Just $ alpnSupportedNTFHandshakes <> httpALPN)
|
||||
(Just alpnSupportedNTFHandshakes)
|
||||
False,
|
||||
startOptions
|
||||
}
|
||||
|
||||
@@ -81,7 +81,6 @@ module Simplex.Messaging.Protocol
|
||||
CommandError (..),
|
||||
ProxyError (..),
|
||||
BrokerErrorType (..),
|
||||
NetworkError (..),
|
||||
BlockingInfo (..),
|
||||
BlockingReason (..),
|
||||
RawTransmission,
|
||||
@@ -169,7 +168,6 @@ module Simplex.Messaging.Protocol
|
||||
noMsgFlags,
|
||||
messageId,
|
||||
messageTs,
|
||||
toNetworkError,
|
||||
|
||||
-- * Parse and serialize
|
||||
ProtocolMsgTag (..),
|
||||
@@ -214,7 +212,7 @@ module Simplex.Messaging.Protocol
|
||||
where
|
||||
|
||||
import Control.Applicative (optional, (<|>))
|
||||
import Control.Exception (Exception, SomeException, displayException, fromException)
|
||||
import Control.Exception (Exception)
|
||||
import Control.Monad.Except
|
||||
import Data.Aeson (FromJSON (..), ToJSON (..))
|
||||
import qualified Data.Aeson.TH as J
|
||||
@@ -243,7 +241,6 @@ import GHC.TypeLits (ErrorMessage (..), TypeError, type (+))
|
||||
import qualified GHC.TypeLits as TE
|
||||
import qualified GHC.TypeLits as Type
|
||||
import Network.Socket (ServiceName)
|
||||
import qualified Network.TLS as TLS
|
||||
import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..))
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Encoding
|
||||
@@ -1558,7 +1555,7 @@ data BrokerErrorType
|
||||
| -- | unexpected response
|
||||
UNEXPECTED {respErr :: String}
|
||||
| -- | network error
|
||||
NETWORK {networkError :: NetworkError}
|
||||
NETWORK
|
||||
| -- | no compatible server host (e.g. onion when public is required, or vice versa)
|
||||
HOST
|
||||
| -- | service unavailable client-side - used in agent errors
|
||||
@@ -1569,24 +1566,6 @@ data BrokerErrorType
|
||||
TIMEOUT
|
||||
deriving (Eq, Read, Show, Exception)
|
||||
|
||||
data NetworkError
|
||||
= NEConnectError {connectError :: String}
|
||||
| NETLSError {tlsError :: String}
|
||||
| NEUnknownCAError
|
||||
| NEFailedError
|
||||
| NETimeoutError
|
||||
| NESubscribeError {subscribeError :: String}
|
||||
deriving (Eq, Read, Show)
|
||||
|
||||
toNetworkError :: SomeException -> NetworkError
|
||||
toNetworkError e = maybe (NEConnectError err) fromTLSError (fromException e)
|
||||
where
|
||||
err = displayException e
|
||||
fromTLSError :: TLS.TLSException -> NetworkError
|
||||
fromTLSError = \case
|
||||
TLS.HandshakeFailed (TLS.Error_Protocol _ TLS.UnknownCa) -> NEUnknownCAError
|
||||
_ -> NETLSError err
|
||||
|
||||
data BlockingInfo = BlockingInfo
|
||||
{ reason :: BlockingReason
|
||||
}
|
||||
@@ -2022,7 +2001,7 @@ instance Encoding BrokerErrorType where
|
||||
RESPONSE e -> "RESPONSE " <> smpEncode e
|
||||
UNEXPECTED e -> "UNEXPECTED " <> smpEncode e
|
||||
TRANSPORT e -> "TRANSPORT " <> smpEncode e
|
||||
NETWORK _e -> "NETWORK" -- TODO once all upgrade: "NETWORK " <> smpEncode e
|
||||
NETWORK -> "NETWORK"
|
||||
TIMEOUT -> "TIMEOUT"
|
||||
HOST -> "HOST"
|
||||
NO_SERVICE -> "NO_SERVICE"
|
||||
@@ -2031,7 +2010,7 @@ instance Encoding BrokerErrorType where
|
||||
"RESPONSE" -> RESPONSE <$> _smpP
|
||||
"UNEXPECTED" -> UNEXPECTED <$> _smpP
|
||||
"TRANSPORT" -> TRANSPORT <$> _smpP
|
||||
"NETWORK" -> NETWORK <$> (_smpP <|> pure NEFailedError)
|
||||
"NETWORK" -> pure NETWORK
|
||||
"TIMEOUT" -> pure TIMEOUT
|
||||
"HOST" -> pure HOST
|
||||
"NO_SERVICE" -> pure NO_SERVICE
|
||||
@@ -2042,7 +2021,7 @@ instance StrEncoding BrokerErrorType where
|
||||
RESPONSE e -> "RESPONSE " <> encodeUtf8 (T.pack e)
|
||||
UNEXPECTED e -> "UNEXPECTED " <> encodeUtf8 (T.pack e)
|
||||
TRANSPORT e -> "TRANSPORT " <> smpEncode e
|
||||
NETWORK _e -> "NETWORK" -- TODO once all upgrade: "NETWORK " <> strEncode e
|
||||
NETWORK -> "NETWORK"
|
||||
TIMEOUT -> "TIMEOUT"
|
||||
HOST -> "HOST"
|
||||
NO_SERVICE -> "NO_SERVICE"
|
||||
@@ -2051,50 +2030,13 @@ instance StrEncoding BrokerErrorType where
|
||||
"RESPONSE" -> RESPONSE <$> _textP
|
||||
"UNEXPECTED" -> UNEXPECTED <$> _textP
|
||||
"TRANSPORT" -> TRANSPORT <$> _smpP
|
||||
"NETWORK" -> NETWORK <$> (_strP <|> pure NEFailedError)
|
||||
"NETWORK" -> pure NETWORK
|
||||
"TIMEOUT" -> pure TIMEOUT
|
||||
"HOST" -> pure HOST
|
||||
"NO_SERVICE" -> pure NO_SERVICE
|
||||
_ -> fail "bad BrokerErrorType"
|
||||
|
||||
instance Encoding NetworkError where
|
||||
smpEncode = \case
|
||||
NEConnectError e -> "CONNECT " <> smpEncode e
|
||||
NETLSError e -> "TLS " <> smpEncode e
|
||||
NEUnknownCAError -> "UNKNOWNCA"
|
||||
NEFailedError -> "FAILED"
|
||||
NETimeoutError -> "TIMEOUT"
|
||||
NESubscribeError e -> "SUBSCRIBE " <> smpEncode e
|
||||
smpP =
|
||||
A.takeTill (== ' ') >>= \case
|
||||
"CONNECT" -> NEConnectError <$> _smpP
|
||||
"TLS" -> NETLSError <$> _smpP
|
||||
"UNKNOWNCA" -> pure NEUnknownCAError
|
||||
"FAILED" -> pure NEFailedError
|
||||
"TIMEOUT" -> pure NETimeoutError
|
||||
"SUBSCRIBE" -> NESubscribeError <$> _smpP
|
||||
_ -> fail "bad NetworkError"
|
||||
|
||||
instance StrEncoding NetworkError where
|
||||
strEncode = \case
|
||||
NEConnectError e -> "CONNECT " <> encodeUtf8 (T.pack e)
|
||||
NETLSError e -> "TLS " <> encodeUtf8 (T.pack e)
|
||||
NEUnknownCAError -> "UNKNOWNCA"
|
||||
NEFailedError -> "FAILED"
|
||||
NETimeoutError -> "TIMEOUT"
|
||||
NESubscribeError e -> "SUBSCRIBE " <> encodeUtf8 (T.pack e)
|
||||
strP =
|
||||
A.takeTill (== ' ') >>= \case
|
||||
"CONNECT" -> NEConnectError <$> _textP
|
||||
"TLS" -> NETLSError <$> _textP
|
||||
"UNKNOWNCA" -> pure NEUnknownCAError
|
||||
"FAILED" -> pure NEFailedError
|
||||
"TIMEOUT" -> pure NETimeoutError
|
||||
"SUBSCRIBE" -> NESubscribeError <$> _textP
|
||||
_ -> fail "bad NetworkError"
|
||||
|
||||
_textP :: Parser String
|
||||
_textP = A.space *> (T.unpack . safeDecodeUtf8 <$> A.takeByteString)
|
||||
where
|
||||
_textP = A.space *> (T.unpack . safeDecodeUtf8 <$> A.takeByteString)
|
||||
|
||||
-- | Send signed SMP transmission to TCP transport.
|
||||
tPut :: Transport c => THandle v c p -> NonEmpty (Either TransportError SentRawTransmission) -> IO [Either TransportError ()]
|
||||
@@ -2258,12 +2200,6 @@ $(J.deriveJSON defaultJSON ''MsgFlags)
|
||||
|
||||
$(J.deriveJSON (sumTypeJSON id) ''CommandError)
|
||||
|
||||
$(J.deriveToJSON (sumTypeJSON $ dropPrefix "NE") ''NetworkError)
|
||||
|
||||
instance FromJSON NetworkError where
|
||||
parseJSON = $(J.mkParseJSON (sumTypeJSON $ dropPrefix "NE") ''NetworkError)
|
||||
omittedField = Just NEFailedError
|
||||
|
||||
$(J.deriveJSON (sumTypeJSON id) ''BrokerErrorType)
|
||||
|
||||
$(J.deriveJSON defaultJSON ''BlockingInfo)
|
||||
|
||||
@@ -203,12 +203,15 @@ 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 $ \s (sniUsed, h) ->
|
||||
runTransportServerState_ ss started tcpPort defaultSupportedParamsHTTPS combinedCreds tCfg {serverALPN = Just combinedALPNs} $ \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
|
||||
|
||||
@@ -960,24 +963,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 qId -> withAdminRole $ unliftIO u $ do
|
||||
CPDelete sId -> withAdminRole $ unliftIO u $ do
|
||||
st <- asks msgStore
|
||||
r <- liftIO $ runExceptT $ do
|
||||
(q, _) <- ExceptT $ getSenderQueue st qId
|
||||
q <- ExceptT $ getQueue st SSender sId
|
||||
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 qId -> withUserRole $ unliftIO u $ do
|
||||
CPStatus sId -> withUserRole $ unliftIO u $ do
|
||||
st <- asks msgStore
|
||||
q <- liftIO $ getSenderQueue st qId
|
||||
q <- liftIO $ getQueueRec st SSender sId
|
||||
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 qId info -> withUserRole $ unliftIO u $ do
|
||||
CPBlock sId info -> withUserRole $ unliftIO u $ do
|
||||
st <- asks msgStore
|
||||
stats <- asks serverStats
|
||||
blocked <- liftIO $ readIORef $ qBlocked stats
|
||||
@@ -986,27 +989,21 @@ 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, QueueRec {status}) <- ExceptT $ getSenderQueue st qId
|
||||
when (status == EntityActive) $ ExceptT $ blockQueue (queueStore st) q info
|
||||
pure status
|
||||
q <- ExceptT $ getQueue st SSender sId
|
||||
ExceptT $ blockQueue (queueStore st) q info
|
||||
case r of
|
||||
Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
|
||||
Right EntityActive -> do
|
||||
Right () -> do
|
||||
incStat $ qBlocked stats
|
||||
liftIO $ hPutStrLn h "ok, queue blocked"
|
||||
Right status -> liftIO $ hPutStrLn h $ "ok, already inactive: " <> show status
|
||||
CPUnblock qId -> withUserRole $ unliftIO u $ do
|
||||
liftIO $ hPutStrLn h "ok"
|
||||
CPUnblock sId -> withUserRole $ unliftIO u $ do
|
||||
st <- asks msgStore
|
||||
r <- liftIO $ runExceptT $ do
|
||||
(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
|
||||
q <- ExceptT $ getQueue st SSender sId
|
||||
ExceptT $ unblockQueue (queueStore st) q
|
||||
liftIO $ hPutStrLn h $ case r of
|
||||
Left e -> "error: " <> show e
|
||||
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"
|
||||
Right () -> "ok"
|
||||
CPSave -> withAdminRole $ withLock' (savingLock srv) "control" $ do
|
||||
hPutStrLn h "saving server state..."
|
||||
unliftIO u $ saveServer False
|
||||
@@ -1015,11 +1012,6 @@ 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
|
||||
@@ -1514,8 +1506,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
|
||||
@@ -1848,10 +1840,10 @@ client
|
||||
Right body -> do
|
||||
when (isJust (queueData qr) && isSecuredMsgQueue qr) $ void $ liftIO $
|
||||
deleteQueueLinkData (queueStore ms) q
|
||||
ServerConfig {messageExpiration, expireMessagesOnSend, msgIdBytes} <- asks config
|
||||
ServerConfig {messageExpiration, msgIdBytes} <- asks config
|
||||
msgId <- randomId' msgIdBytes
|
||||
msg_ <- liftIO $ runExceptT $ do
|
||||
when expireMessagesOnSend $ mapM_ (expireMessages stats) messageExpiration
|
||||
expireMessages messageExpiration stats
|
||||
msg <- liftIO $ mkMessage msgId body
|
||||
writeMsg ms q True msg
|
||||
case msg_ of
|
||||
@@ -1875,9 +1867,9 @@ client
|
||||
msgTs <- getSystemTime
|
||||
pure $! Message msgId msgTs msgFlags body
|
||||
|
||||
expireMessages :: ServerStats -> ExpirationConfig -> ExceptT ErrorType IO ()
|
||||
expireMessages stats msgExp = do
|
||||
deleted <- deleteExpiredMsgs ms q =<< liftIO (expireBeforeEpoch msgExp)
|
||||
expireMessages :: Maybe ExpirationConfig -> ServerStats -> ExceptT ErrorType IO ()
|
||||
expireMessages msgExp stats = do
|
||||
deleted <- maybe (pure 0) (deleteExpiredMsgs ms q <=< liftIO . expireBeforeEpoch) msgExp
|
||||
liftIO $ when (deleted > 0) $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
|
||||
|
||||
-- The condition for delivery of the message is:
|
||||
|
||||
@@ -7,7 +7,6 @@
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE TemplateHaskell #-}
|
||||
{-# LANGUAGE TupleSections #-}
|
||||
{-# LANGUAGE TypeApplications #-}
|
||||
|
||||
@@ -34,9 +33,8 @@ 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 (..), simplexMQVersion)
|
||||
import Simplex.Messaging.Transport (ASrvTransport, ATransport (..), TLS, Transport (..))
|
||||
import Simplex.Messaging.Transport.Server (AddHTTP, loadFileFingerprint)
|
||||
import Simplex.Messaging.Transport.WebSockets (WS)
|
||||
import Simplex.Messaging.Util (eitherToMaybe, whenM)
|
||||
@@ -99,12 +97,6 @@ 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
|
||||
|
||||
|
||||
@@ -153,7 +153,6 @@ data ServerConfig s = ServerConfig
|
||||
-- | time after which the messages can be removed from the queues and check interval, seconds
|
||||
messageExpiration :: Maybe ExpirationConfig,
|
||||
expireMessagesOnStart :: Bool,
|
||||
expireMessagesOnSend :: Bool,
|
||||
-- | interval of inactivity after which journal queue is closed
|
||||
idleQueueInterval :: Int64,
|
||||
-- | notification expiration interval (seconds)
|
||||
|
||||
@@ -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 (ConnectionLink, ConnectionMode (..))
|
||||
import Simplex.Messaging.Agent.Protocol (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 (ConnectionLink 'CMContact),
|
||||
{ simplex :: Maybe (ConnectionRequestUri 'CMContact),
|
||||
email :: Maybe Text, -- it is recommended that it matches DNS email address, if either is present
|
||||
pgp :: Maybe PGPKey
|
||||
}
|
||||
|
||||
@@ -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 (ConnectionLink (..), connReqUriP')
|
||||
import Simplex.Messaging.Agent.Protocol (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,9 +56,8 @@ 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 (supportedProxyClientSMPRelayVRange, alpnSupportedSMPHandshakes, supportedServerSMPRelayVRange)
|
||||
import Simplex.Messaging.Transport (simplexMQVersion, 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)
|
||||
@@ -235,7 +234,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" <> simplexmqVersionCommit
|
||||
serverVersion = "SMP server v" <> simplexMQVersion
|
||||
executableName = "smp-server"
|
||||
storeLogFilePath = combine logPath "smp-server-store.log"
|
||||
storeMsgsFilePath = combine logPath "smp-server-messages.log"
|
||||
@@ -428,7 +427,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
|
||||
{ ttl = 86400 * readIniDefault defMsgExpirationDays "STORE_LOG" "expire_messages_days" ini
|
||||
},
|
||||
expireMessagesOnStart = fromMaybe True $ iniOnOff "STORE_LOG" "expire_messages_on_start" ini,
|
||||
expireMessagesOnSend = fromMaybe True $ iniOnOff "STORE_LOG" "expire_messages_on_send" ini,
|
||||
idleQueueInterval = defaultIdleQueueInterval,
|
||||
notificationExpiration =
|
||||
defaultNtfExpiration
|
||||
@@ -452,7 +450,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
|
||||
transportConfig =
|
||||
mkTransportServerConfig
|
||||
(fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini)
|
||||
(Just $ alpnSupportedSMPHandshakes <> httpALPN)
|
||||
(Just alpnSupportedSMPHandshakes)
|
||||
(fromMaybe True $ iniOnOff "TRANSPORT" "accept_service_credentials" ini), -- TODO [certs] remove this option
|
||||
controlPort = eitherToMaybe $ T.unpack <$> lookupValue "TRANSPORT" "control_port" ini,
|
||||
smpAgentCfg =
|
||||
@@ -640,8 +638,7 @@ 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 linkP . encodeUtf8 <$!> eitherToMaybe (lookupValue "INFORMATION" simplexField ini)
|
||||
linkP = CLFull <$> connReqUriP' Nothing <|> CLShort <$> strP
|
||||
let simplex = either error id . parseAll (connReqUriP' Nothing) . encodeUtf8 <$!> eitherToMaybe (lookupValue "INFORMATION" simplexField ini)
|
||||
email = infoValue emailField
|
||||
pkURI_ = infoValue pgpKeyUriField
|
||||
pkFingerprint_ = infoValue pgpKeyFingerprintField
|
||||
@@ -810,7 +807,7 @@ cliCommandP cfgPath logPath iniFile =
|
||||
sourceCode = T.pack <$> sourceCode,
|
||||
serverInfo =
|
||||
ServerPublicInfo
|
||||
{ sourceCode = T.pack $ fromMaybe simplexmqSource sourceCode,
|
||||
{ sourceCode = T.pack simplexmqSource,
|
||||
usageConditions = Nothing,
|
||||
operator = fst operator_,
|
||||
website,
|
||||
|
||||
@@ -1,17 +0,0 @@
|
||||
{-# 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
|
||||
_ -> ""
|
||||
@@ -87,13 +87,12 @@ iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
|
||||
<> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
|
||||
<> "# Messages and notifications expiration periods.\n"
|
||||
<> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
|
||||
<> "expire_messages_on_start: on\n\
|
||||
\expire_messages_on_send: off\n"
|
||||
<> "expire_messages_on_start: on\n"
|
||||
<> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
|
||||
<> "# Log daily server statistics to CSV file\n"
|
||||
<> ("log_stats: " <> onOff logStats <> "\n\n")
|
||||
<> "# Log interval for real-time Prometheus metrics\n\
|
||||
\# prometheus_interval: 60\n\n\
|
||||
\# prometheus_interval: 300\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 =
|
||||
withTransaction (dbStore st) $ \db -> do
|
||||
withConnection (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 =
|
||||
withTransaction (dbStore toStore) $ \db ->
|
||||
withConnection (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 <-
|
||||
withTransaction st $ \db -> do
|
||||
withConnection 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 : _ <- withTransaction st (`DB.query_` "SELECT count(*) FROM msg_queues")
|
||||
Only qCnt : _ <- withConnection 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 =
|
||||
withTransaction (dbStore st) $ \db ->
|
||||
withConnection (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) <- withTransaction (dbStore st) $ \db ->
|
||||
(n, r) <- withConnection (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 (withTransaction (dbStore st) action) >>= either logErr pure
|
||||
ExceptT $ E.try (withConnection (dbStore st) action) >>= either logErr pure
|
||||
where
|
||||
logErr :: E.SomeException -> IO (Either ErrorType a)
|
||||
logErr e = logError ("STORE: " <> err) $> Left (STORE err)
|
||||
|
||||
@@ -114,7 +114,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
|
||||
serviceQueuesCount serviceSel = foldM (\n s -> (n +) . S.size <$> readTVarIO (serviceSel s)) 0
|
||||
|
||||
addQueue_ :: STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q)
|
||||
addQueue_ st mkQ rId qr@QueueRec {senderId = sId, notifier, queueData, rcvServiceId} = do
|
||||
addQueue_ st mkQ rId qr@QueueRec {senderId = sId, notifier, queueData} = do
|
||||
sq <- mkQ rId qr
|
||||
add sq $>> withLog "addStoreQueue" st (\s -> logCreateQueue s rId qr) $> Right sq
|
||||
where
|
||||
@@ -122,11 +122,8 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
|
||||
add q = atomically $ ifM hasId (pure $ Left DUPLICATE_) $ Right () <$ do
|
||||
TM.insert rId q queues
|
||||
TM.insert sId rId senders
|
||||
forM_ notifier $ \NtfCreds {notifierId = nId, ntfServiceId} -> do
|
||||
TM.insert nId rId notifiers
|
||||
mapM_ (addServiceQueue st serviceNtfQueues nId) ntfServiceId
|
||||
forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId notifiers
|
||||
forM_ queueData $ \(lnkId, _) -> TM.insert lnkId rId links
|
||||
mapM_ (addServiceQueue st serviceRcvQueues rId) rcvServiceId
|
||||
hasId = anyM [TM.member rId queues, TM.member sId senders, hasNotifier, hasLink]
|
||||
hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId notifiers) notifier
|
||||
hasLink = maybe (pure False) (\(lnkId, _) -> TM.member lnkId links) queueData
|
||||
|
||||
@@ -30,14 +30,12 @@ where
|
||||
|
||||
import Control.Applicative (optional, (<|>))
|
||||
import Control.Logger.Simple (logError)
|
||||
import Control.Monad
|
||||
import Data.Aeson (FromJSON (..), ToJSON (..))
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Char (isAsciiLower, isDigit, isHexDigit)
|
||||
import Data.Default (def)
|
||||
import Data.Functor (($>))
|
||||
import Data.IORef
|
||||
import Data.IP
|
||||
import Data.List.NonEmpty (NonEmpty (..))
|
||||
@@ -60,7 +58,7 @@ import Simplex.Messaging.Parsers (parseAll, parseString)
|
||||
import Simplex.Messaging.Transport
|
||||
import Simplex.Messaging.Transport.KeepAlive
|
||||
import Simplex.Messaging.Transport.Shared
|
||||
import Simplex.Messaging.Util (bshow, catchAll, catchAll_, tshow, (<$?>))
|
||||
import Simplex.Messaging.Util (bshow, catchAll, tshow, (<$?>))
|
||||
import System.IO.Error
|
||||
import Text.Read (readMaybe)
|
||||
import UnliftIO.Exception (IOException)
|
||||
@@ -158,11 +156,6 @@ clientTransportConfig TransportClientConfig {logTLSErrors} =
|
||||
runTransportClient :: Transport c => TransportClientConfig -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c 'TClient -> IO a) -> IO a
|
||||
runTransportClient = runTLSTransportClient defaultSupportedParams Nothing
|
||||
|
||||
data ConnectionHandle c
|
||||
= CHSocket Socket
|
||||
| CHContext T.Context
|
||||
| CHTransport (c 'TClient)
|
||||
|
||||
runTLSTransportClient :: Transport c => T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> Maybe SocksCredentials -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c 'TClient -> IO a) -> IO a
|
||||
runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy, tcpKeepAlive, clientCredentials, clientALPN, useSNI} socksCreds host port keyHash client = do
|
||||
serverCert <- newEmptyTMVarIO
|
||||
@@ -172,22 +165,17 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
|
||||
connectTCP = case socksProxy of
|
||||
Just proxy -> connectSocksClient proxy socksCreds (hostAddr host)
|
||||
_ -> connectTCPClient hostName
|
||||
h <- newIORef Nothing
|
||||
let set hc = (>>= \c -> writeIORef h (Just $ hc c) $> c)
|
||||
E.bracket (set CHSocket $ connectTCP port) (\_ -> closeConn h) $ \sock -> do
|
||||
mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive " <> tshow e)
|
||||
c <- do
|
||||
sock <- connectTCP port
|
||||
mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
|
||||
let tCfg = clientTransportConfig cfg
|
||||
-- No TLS timeout to avoid failing connections via SOCKS
|
||||
tls <- set CHContext $ connectTLS (Just hostName) tCfg clientParams sock
|
||||
chain <- takePeerCertChain serverCert
|
||||
tls <- connectTLS (Just hostName) tCfg clientParams sock
|
||||
chain <- takePeerCertChain serverCert `E.onException` closeTLS tls
|
||||
sent <- readIORef clientCredsSent
|
||||
client =<< set CHTransport (getTransportConnection tCfg sent chain tls)
|
||||
getTransportConnection tCfg sent chain tls
|
||||
client c `E.finally` closeConnection c
|
||||
where
|
||||
closeConn = readIORef >=> mapM_ (\c -> E.uninterruptibleMask_ $ closeConn_ c `catchAll_` pure ())
|
||||
closeConn_ = \case
|
||||
CHSocket sock -> close sock
|
||||
CHContext tls -> closeTLS tls
|
||||
CHTransport c -> closeConnection c
|
||||
hostAddr = \case
|
||||
THIPv4 addr -> SocksAddrIPV4 $ tupleToHostAddress addr
|
||||
THIPv6 addr -> SocksAddrIPV6 addr
|
||||
@@ -211,11 +199,10 @@ connectTCPClient host port = withSocketsDo $ resolve >>= tryOpen err
|
||||
E.try (open addr) >>= either (`tryOpen` as) pure
|
||||
|
||||
open :: AddrInfo -> IO Socket
|
||||
open addr =
|
||||
E.bracketOnError
|
||||
(socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr))
|
||||
close
|
||||
(\sock -> connect sock (addrAddress addr) $> sock)
|
||||
open addr = do
|
||||
sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
|
||||
connect sock $ addrAddress addr
|
||||
pure sock
|
||||
|
||||
defaultSMPPort :: PortNumber
|
||||
defaultSMPPort = 5223
|
||||
|
||||
@@ -25,6 +25,14 @@ 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
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
|
||||
module Simplex.Messaging.Transport.HTTP2 where
|
||||
|
||||
@@ -16,7 +15,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 (ALPN, TLS, Transport (cGet, cPut))
|
||||
import Simplex.Messaging.Transport (TLS, Transport (cGet, cPut))
|
||||
import Simplex.Messaging.Transport.Buffer
|
||||
import qualified System.TimeManager as TI
|
||||
|
||||
@@ -82,9 +81,3 @@ 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"
|
||||
|
||||
@@ -27,7 +27,6 @@ import qualified Network.TLS as T
|
||||
import Numeric.Natural (Natural)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Protocol (NetworkError (..), toNetworkError)
|
||||
import Simplex.Messaging.Transport (ALPN, STransportPeer (..), SessionId, TLS (tlsALPN, tlsPeerCert, tlsUniq), TransportPeer (..), TransportPeerI (..), getServerVerifyKey)
|
||||
import Simplex.Messaging.Transport.Client (TransportClientConfig (..), TransportHost (..), defaultTcpConnectTimeout, runTLSTransportClient)
|
||||
import Simplex.Messaging.Transport.HTTP2
|
||||
@@ -90,7 +89,7 @@ defaultHTTP2ClientConfig =
|
||||
suportedTLSParams = http2TLSParams
|
||||
}
|
||||
|
||||
data HTTP2ClientError = HCResponseTimeout | HCNetworkError NetworkError | HCIOError IOException
|
||||
data HTTP2ClientError = HCResponseTimeout | HCNetworkError | HCIOError IOException
|
||||
deriving (Show)
|
||||
|
||||
getHTTP2Client :: HostName -> ServiceName -> Maybe XS.CertificateStore -> HTTP2ClientConfig -> IO () -> IO (Either HTTP2ClientError HTTP2Client)
|
||||
@@ -122,15 +121,12 @@ getVerifiedHTTP2ClientWith config host port disconnected setup =
|
||||
runClient :: HClient -> IO (Either HTTP2ClientError HTTP2Client)
|
||||
runClient c = do
|
||||
cVar <- newEmptyTMVarIO
|
||||
action <-
|
||||
async $ setup (client c cVar) `E.catch` \e -> do
|
||||
atomically $ putTMVar cVar $ Left $ HCNetworkError $ toNetworkError e
|
||||
E.throwIO e
|
||||
action <- async $ setup (client c cVar) `E.finally` atomically (putTMVar cVar $ Left HCNetworkError)
|
||||
c_ <- connTimeout config `timeout` atomically (takeTMVar cVar)
|
||||
case c_ of
|
||||
Just (Right c') -> pure $ Right c' {action = Just action}
|
||||
Just (Left e) -> pure $ Left e
|
||||
Nothing -> cancel action $> Left (HCNetworkError NETimeoutError)
|
||||
Nothing -> cancel action $> Left HCNetworkError
|
||||
|
||||
client :: HClient -> TMVar (Either HTTP2ClientError HTTP2Client) -> TLS p -> H.Client HTTP2Response
|
||||
client c cVar tls sendReq = do
|
||||
@@ -180,7 +176,7 @@ sendRequestDirect HTTP2Client {client_ = HClient {config, disconnected}, sendReq
|
||||
reqTimeout `timeout` try (sendReq req process) >>= \case
|
||||
Just (Right r) -> pure $ Right r
|
||||
Just (Left e) -> disconnected $> Left (HCIOError e)
|
||||
Nothing -> pure $ Left HCResponseTimeout
|
||||
Nothing -> pure $ Left HCNetworkError
|
||||
where
|
||||
process r = do
|
||||
respBody <- getHTTP2Body r $ bodyHeadSize config
|
||||
|
||||
@@ -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, unlessM)
|
||||
import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
|
||||
import System.Exit (exitFailure)
|
||||
import System.IO.Error (tryIOError)
|
||||
import System.Mem.Weak (Weak, deRefWeak)
|
||||
@@ -172,13 +172,12 @@ 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 $ writeTVar closed True >> modifyTVar' clients (IM.delete cId)
|
||||
atomically $ 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 $ unlessM (readTVar closed) $ modifyTVar' clients $ IM.insert cId tId
|
||||
atomically $ 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.
|
||||
|
||||
@@ -171,30 +171,28 @@ catchAll_ :: IO a -> IO a -> IO a
|
||||
catchAll_ a = catchAll a . const
|
||||
{-# INLINE catchAll_ #-}
|
||||
|
||||
class Show e => AnyError e where fromSomeException :: E.SomeException -> e
|
||||
|
||||
tryAllErrors :: (AnyError e, MonadUnliftIO m) => ExceptT e m a -> ExceptT e m (Either e a)
|
||||
tryAllErrors action = ExceptT $ Right <$> runExceptT action `UE.catch` (pure . Left . fromSomeException)
|
||||
tryAllErrors :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> ExceptT e m (Either e a)
|
||||
tryAllErrors err action = ExceptT $ Right <$> runExceptT action `UE.catch` (pure . Left . err)
|
||||
{-# INLINE tryAllErrors #-}
|
||||
|
||||
tryAllErrors' :: (AnyError e, MonadUnliftIO m) => ExceptT e m a -> m (Either e a)
|
||||
tryAllErrors' action = runExceptT action `UE.catch` (pure . Left . fromSomeException)
|
||||
tryAllErrors' :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> m (Either e a)
|
||||
tryAllErrors' err action = runExceptT action `UE.catch` (pure . Left . err)
|
||||
{-# INLINE tryAllErrors' #-}
|
||||
|
||||
catchAllErrors :: (AnyError e, MonadUnliftIO m) => ExceptT e m a -> (e -> ExceptT e m a) -> ExceptT e m a
|
||||
catchAllErrors action handler = tryAllErrors action >>= either handler pure
|
||||
catchAllErrors :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> (e -> ExceptT e m a) -> ExceptT e m a
|
||||
catchAllErrors err action handler = tryAllErrors err action >>= either handler pure
|
||||
{-# INLINE catchAllErrors #-}
|
||||
|
||||
catchAllErrors' :: (AnyError e, MonadUnliftIO m) => ExceptT e m a -> (e -> m a) -> m a
|
||||
catchAllErrors' action handler = tryAllErrors' action >>= either handler pure
|
||||
catchAllErrors' :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> (e -> m a) -> m a
|
||||
catchAllErrors' err action handler = tryAllErrors' err action >>= either handler pure
|
||||
{-# INLINE catchAllErrors' #-}
|
||||
|
||||
catchThrow :: MonadUnliftIO m => ExceptT e m a -> (SomeException -> e) -> ExceptT e m a
|
||||
action `catchThrow` err = ExceptT $ runExceptT action `UE.catch` (pure . Left . err)
|
||||
catchThrow :: MonadUnliftIO m => ExceptT e m a -> (E.SomeException -> e) -> ExceptT e m a
|
||||
catchThrow action err = catchAllErrors err action throwE
|
||||
{-# INLINE catchThrow #-}
|
||||
|
||||
allFinally :: (AnyError e, MonadUnliftIO m) => ExceptT e m a -> ExceptT e m b -> ExceptT e m a
|
||||
allFinally action final = tryAllErrors action >>= \r -> final >> except r
|
||||
allFinally :: MonadUnliftIO m => (E.SomeException -> e) -> ExceptT e m a -> ExceptT e m b -> ExceptT e m a
|
||||
allFinally err action final = tryAllErrors err action >>= \r -> final >> except r
|
||||
{-# INLINE allFinally #-}
|
||||
|
||||
eitherToMaybe :: Either a b -> Maybe b
|
||||
|
||||
@@ -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) (1, 24 * 999999) "localhost" -- session-signing cert
|
||||
leaf <- genCredentials drg (Just caCreds) (0, 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 (1, 24 * 999999) "ca"
|
||||
((_, caKey), caCert) <- genCredentials drg Nothing (0, 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
|
||||
@@ -306,8 +306,14 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
|
||||
atomically $ takeTMVar endSession
|
||||
logDebug "Session ended"
|
||||
|
||||
catchRCError :: ExceptT RCErrorType IO a -> (RCErrorType -> ExceptT RCErrorType IO a) -> ExceptT RCErrorType IO a
|
||||
catchRCError = catchAllErrors $ \e -> case fromException e of
|
||||
Just (TLS.Terminated _ _ (TLS.Error_Protocol _ TLS.UnknownCa)) -> RCEIdentity
|
||||
_ -> RCEException $ show e
|
||||
{-# INLINE catchRCError #-}
|
||||
|
||||
putRCError :: ExceptT RCErrorType IO a -> TMVar (Either RCErrorType b) -> ExceptT RCErrorType IO a
|
||||
a `putRCError` r = a `catchAllErrors` \e -> atomically (tryPutTMVar r $ Left e) >> throwE e
|
||||
a `putRCError` r = a `catchRCError` \e -> atomically (tryPutTMVar r $ Left e) >> throwE e
|
||||
|
||||
sendRCPacket :: Encoding a => TLS p -> a -> ExceptT RCErrorType IO ()
|
||||
sendRCPacket tls pkt = do
|
||||
@@ -389,7 +395,7 @@ discoverRCCtrl subscribers pairings =
|
||||
pure r
|
||||
where
|
||||
loop :: ExceptT RCErrorType IO a -> ExceptT RCErrorType IO a
|
||||
loop action = action `catchAllErrors` \e -> logError (tshow e) >> loop action
|
||||
loop action = action `catchRCError` \e -> logError (tshow e) >> loop action
|
||||
|
||||
findRCCtrlPairing :: NonEmpty RCCtrlPairing -> RCEncInvitation -> ExceptT RCErrorType IO (RCCtrlPairing, RCVerifiedInvitation)
|
||||
findRCCtrlPairing pairings RCEncInvitation {dhPubKey, nonce, encInvitation} = do
|
||||
|
||||
@@ -19,7 +19,6 @@ import qualified Data.Text as T
|
||||
import Data.Text.Encoding (encodeUtf8)
|
||||
import Data.Word (Word16)
|
||||
import qualified Data.X509 as X
|
||||
import qualified Network.TLS as TLS
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Crypto.SNTRUP761.Bindings
|
||||
import Simplex.Messaging.Encoding
|
||||
@@ -27,7 +26,7 @@ import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, sumTypeJSON)
|
||||
import Simplex.Messaging.Transport (TLS, TSbChainKeys, TransportPeer (..))
|
||||
import Simplex.Messaging.Transport.Client (TransportHost)
|
||||
import Simplex.Messaging.Util (AnyError (..), safeDecodeUtf8)
|
||||
import Simplex.Messaging.Util (safeDecodeUtf8)
|
||||
import Simplex.Messaging.Version (VersionRange, VersionScope, mkVersionRange)
|
||||
import Simplex.Messaging.Version.Internal
|
||||
import UnliftIO
|
||||
@@ -51,12 +50,6 @@ data RCErrorType
|
||||
| RCESyntax {syntaxErr :: String}
|
||||
deriving (Eq, Show, Exception)
|
||||
|
||||
instance AnyError RCErrorType where
|
||||
fromSomeException e = case fromException e of
|
||||
Just (TLS.Terminated _ _ (TLS.Error_Protocol _ TLS.UnknownCa)) -> RCEIdentity
|
||||
_ -> RCEException $ show e
|
||||
{-# INLINE fromSomeException #-}
|
||||
|
||||
instance StrEncoding RCErrorType where
|
||||
strEncode = \case
|
||||
RCEInternal err -> "INTERNAL" <> text err
|
||||
|
||||
@@ -98,7 +98,7 @@ import qualified Simplex.Messaging.Crypto.Ratchet as CR
|
||||
import Simplex.Messaging.Encoding
|
||||
import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Notifications.Transport (NTFVersion, pattern VersionNTF)
|
||||
import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, NetworkError (..), ProtocolServer (..), SubscriptionMode (..), initialSMPClientVersion, srvHostnamesSMPClientVersion, supportedSMPClientVRange)
|
||||
import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolServer (..), SubscriptionMode (..), initialSMPClientVersion, srvHostnamesSMPClientVersion, supportedSMPClientVRange)
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (..), ServerStoreCfg (..), StorePaths (..))
|
||||
import Simplex.Messaging.Server.Expiration
|
||||
@@ -177,7 +177,7 @@ pGet' c skipWarn = do
|
||||
case cmd of
|
||||
CONNECT {} -> pGet c
|
||||
DISCONNECT {} -> pGet c
|
||||
ERR (BROKER _ (NETWORK _)) -> pGet c
|
||||
ERR (BROKER _ NETWORK) -> pGet c
|
||||
MWARN {} | skipWarn -> pGet c
|
||||
RFWARN {} | skipWarn -> pGet c
|
||||
SFWARN {} | skipWarn -> pGet c
|
||||
@@ -516,7 +516,7 @@ functionalAPITests ps = do
|
||||
it "should pass without basic auth" $ testSMPServerConnectionTest ps Nothing (noAuthSrv testSMPServer2) `shouldReturn` Nothing
|
||||
let srv1 = testSMPServer2 {keyHash = "1234"}
|
||||
it "should fail with incorrect fingerprint" $ do
|
||||
testSMPServerConnectionTest ps Nothing (noAuthSrv srv1) `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) $ NETWORK NEUnknownCAError)
|
||||
testSMPServerConnectionTest ps Nothing (noAuthSrv srv1) `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) NETWORK)
|
||||
describe "server with password" $ do
|
||||
let auth = Just "abcd"
|
||||
srv = ProtoServerWithAuth testSMPServer2
|
||||
@@ -1105,7 +1105,7 @@ testAsyncServerOffline ps = withAgentClients2 $ \alice bob -> do
|
||||
(bobId, cReq) <- withSmpServerStoreLogOn ps testPort $ \_ ->
|
||||
runRight $ createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
-- connection fails
|
||||
Left (BROKER _ (NETWORK _)) <- runExceptT $ joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
|
||||
Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
|
||||
("", "", DOWN srv conns) <- nGet alice
|
||||
srv `shouldBe` testSMPServer
|
||||
conns `shouldBe` [bobId]
|
||||
@@ -1172,13 +1172,13 @@ testInvitationErrors ps restart = do
|
||||
("", "", DOWN _ [_]) <- nGet a
|
||||
aId <- runRight $ A.prepareConnectionToJoin b 1 True cReq PQSupportOn
|
||||
-- fails to secure the queue on testPort
|
||||
BROKER srv (NETWORK _) <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
BROKER srv NETWORK <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
(testPort `isSuffixOf` srv) `shouldBe` True
|
||||
withServer1 ps $ do
|
||||
("", "", UP _ [_]) <- nGet a
|
||||
let loopSecure = do
|
||||
-- secures the queue on testPort, but fails to create reply queue on testPort2
|
||||
BROKER srv2 (NETWORK _) <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
BROKER srv2 NETWORK <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
unless (testPort2 `isSuffixOf` srv2) $ putStrLn "retrying secure" >> threadDelay 200000 >> loopSecure
|
||||
loopSecure
|
||||
("", "", DOWN _ [_]) <- nGet a
|
||||
@@ -1186,7 +1186,7 @@ testInvitationErrors ps restart = do
|
||||
threadDelay 200000
|
||||
let loopCreate = do
|
||||
-- creates the reply queue on testPort2, but fails to send it to testPort
|
||||
BROKER srv' (NETWORK _) <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
BROKER srv' NETWORK <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
unless (testPort `isSuffixOf` srv') $ putStrLn "retrying create" >> threadDelay 200000 >> loopCreate
|
||||
loopCreate
|
||||
restartAgentB restart b [aId]
|
||||
@@ -1242,12 +1242,12 @@ testContactErrors ps restart = do
|
||||
("", "", DOWN _ [_]) <- nGet a
|
||||
aId <- runRight $ A.prepareConnectionToJoin b 1 True cReq PQSupportOn
|
||||
-- fails to create queue on testPort2
|
||||
BROKER srv2 (NETWORK _) <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
BROKER srv2 NETWORK <- runLeft $ A.joinConnection b NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
(testPort2 `isSuffixOf` srv2) `shouldBe` True
|
||||
b' <- restartAgentB restart b [aId]
|
||||
let loopCreate2 = do
|
||||
-- creates the reply queue on testPort2, but fails to send invitation to testPort
|
||||
BROKER srv' (NETWORK _) <- runLeft $ A.joinConnection b' NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
BROKER srv' NETWORK <- runLeft $ A.joinConnection b' NRMInteractive 1 aId True cReq "bob's connInfo" PQSupportOn SMSubscribe
|
||||
unless (testPort `isSuffixOf` srv') $ putStrLn "retrying create 2" >> threadDelay 200000 >> loopCreate2
|
||||
b'' <- withServer2 ps $ do
|
||||
loopCreate2
|
||||
@@ -1270,7 +1270,7 @@ testContactErrors ps restart = do
|
||||
("", "", UP _ [_]) <- nGet b''
|
||||
let loopSecure = do
|
||||
-- secures the queue on testPort2, but fails to create reply queue on testPort
|
||||
BROKER srv (NETWORK _) <- runLeft $ acceptContact a 1 bId True invId "alice's connInfo" PQSupportOn SMSubscribe
|
||||
BROKER srv NETWORK <- runLeft $ acceptContact a 1 bId True invId "alice's connInfo" PQSupportOn SMSubscribe
|
||||
unless (testPort `isSuffixOf` srv) $ putStrLn "retrying secure" >> threadDelay 200000 >> loopSecure
|
||||
loopSecure
|
||||
("", "", DOWN _ [_]) <- nGet b''
|
||||
@@ -1278,7 +1278,7 @@ testContactErrors ps restart = do
|
||||
("", "", UP _ [_]) <- nGet a
|
||||
let loopCreate = do
|
||||
-- creates the reply queue on testPort, but fails to send confirmation to testPort2
|
||||
BROKER srv2' (NETWORK _) <- runLeft $ acceptContact a 1 bId True invId "alice's connInfo" PQSupportOn SMSubscribe
|
||||
BROKER srv2' NETWORK <- runLeft $ acceptContact a 1 bId True invId "alice's connInfo" PQSupportOn SMSubscribe
|
||||
unless (testPort2 `isSuffixOf` srv2') $ putStrLn "retrying create" >> threadDelay 200000 >> loopCreate
|
||||
loopCreate
|
||||
restartAgentA restart a [contactId, bId]
|
||||
@@ -1743,7 +1743,7 @@ testDuplicateMessage ps = do
|
||||
-- commenting two lines below and uncommenting further two lines would also runRight_,
|
||||
-- it is the scenario tested above, when the message was not acknowledged by the user
|
||||
threadDelay 200000
|
||||
Left (BROKER _ (NETWORK _)) <- runExceptT $ ackMessage bob1 aliceId 3 Nothing
|
||||
Left (BROKER _ NETWORK) <- runExceptT $ ackMessage bob1 aliceId 3 Nothing
|
||||
|
||||
disposeAgentClient alice
|
||||
disposeAgentClient bob1
|
||||
@@ -1827,8 +1827,8 @@ testDeliveryAfterSubscriptionError ps = do
|
||||
pure (aId, bId)
|
||||
|
||||
withAgentClients2 $ \a b -> do
|
||||
Left (BROKER _ (NETWORK _)) <- runExceptT $ subscribeConnection a bId
|
||||
Left (BROKER _ (NETWORK _)) <- runExceptT $ subscribeConnection b aId
|
||||
Left (BROKER _ NETWORK) <- runExceptT $ subscribeConnection a bId
|
||||
Left (BROKER _ NETWORK) <- runExceptT $ subscribeConnection b aId
|
||||
withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
|
||||
withUP a bId $ \case ("", c, SENT 2) -> c == bId; _ -> False
|
||||
withUP b aId $ \case ("", c, Msg "hello") -> c == aId; _ -> False
|
||||
@@ -1872,7 +1872,7 @@ testExpireMessage ps =
|
||||
2 <- runRight $ sendMessage a bId SMP.noMsgFlags "1"
|
||||
threadDelay 1500000
|
||||
3 <- runRight $ sendMessage a bId SMP.noMsgFlags "2" -- this won't expire
|
||||
get a =##> \case ("", c, MERR 2 (BROKER _ e)) -> bId == c && networkOrTimeoutError e; _ -> False
|
||||
get a =##> \case ("", c, MERR 2 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
withSmpServerStoreLogOn ps testPort $ \_ -> runRight_ $ do
|
||||
withUP a bId $ \case ("", _, SENT 3) -> True; _ -> False
|
||||
withUP b aId $ \case ("", _, MsgErr 2 (MsgSkipped 2 2) "2") -> True; _ -> False
|
||||
@@ -1891,8 +1891,8 @@ testExpireManyMessages ps =
|
||||
4 <- sendMessage a bId SMP.noMsgFlags "3"
|
||||
liftIO $ threadDelay 2000000
|
||||
5 <- sendMessage a bId SMP.noMsgFlags "4" -- this won't expire
|
||||
get a =##> \case ("", c, MERR 2 (BROKER _ e)) -> bId == c && networkOrTimeoutError e; _ -> False
|
||||
let expected c e = bId == c && networkOrTimeoutError e
|
||||
get a =##> \case ("", c, MERR 2 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
let expected c e = bId == c && (e == TIMEOUT || e == NETWORK)
|
||||
get a >>= \case
|
||||
("", c, MERR 3 (BROKER _ e)) -> do
|
||||
liftIO $ expected c e `shouldBe` True
|
||||
@@ -2633,7 +2633,7 @@ testDeleteConnectionAsync ps =
|
||||
runRight_ $ do
|
||||
deleteConnectionsAsync a False connIds
|
||||
nGet a =##> \case ("", "", DOWN {}) -> True; _ -> False
|
||||
let delOk = \case (c, _, _, Just (BROKER _ e)) -> c `elem` connIds && networkOrTimeoutError e; _ -> False
|
||||
let delOk = \case (c, _, _, Just (BROKER _ e)) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
get a =##> \case ("", "", DEL_RCVQS rs) -> length rs == 3 && all delOk rs; _ -> False
|
||||
get a =##> \case ("", "", DEL_CONNS cs) -> length cs == 3 && all (`elem` connIds) cs; _ -> False
|
||||
liftIO $ noMessages a "nothing else should be delivered to alice"
|
||||
@@ -2691,7 +2691,7 @@ testWaitDelivery ps =
|
||||
3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
|
||||
4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
|
||||
deleteConnectionsAsync alice True [bobId]
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && networkOrTimeoutError e; _ -> False
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
liftIO $ noMessages alice "nothing else should be delivered to alice"
|
||||
liftIO $ noMessages bob "nothing else should be delivered to bob"
|
||||
|
||||
@@ -2748,7 +2748,7 @@ testWaitDeliveryAUTHErr ps =
|
||||
3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
|
||||
4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
|
||||
deleteConnectionsAsync alice True [bobId]
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && networkOrTimeoutError e; _ -> False
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
liftIO $ noMessages alice "nothing else should be delivered to alice"
|
||||
liftIO $ noMessages bob "nothing else should be delivered to bob"
|
||||
|
||||
@@ -2788,7 +2788,7 @@ testWaitDeliveryTimeout ps =
|
||||
3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
|
||||
4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
|
||||
deleteConnectionsAsync alice True [bobId]
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && networkOrTimeoutError e; _ -> False
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False
|
||||
liftIO $ noMessages alice "nothing else should be delivered to alice"
|
||||
liftIO $ noMessages bob "nothing else should be delivered to bob"
|
||||
@@ -2828,7 +2828,7 @@ testWaitDeliveryTimeout2 ps =
|
||||
3 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
|
||||
4 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "message 1"
|
||||
deleteConnectionsAsync alice True [bobId]
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && networkOrTimeoutError e; _ -> False
|
||||
get alice =##> \case ("", "", DEL_RCVQS [(cId, _, _, Just (BROKER _ e))]) -> cId == bobId && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
get alice =##> \case ("", "", DEL_CONNS [cId]) -> cId == bobId; _ -> False
|
||||
liftIO $ noMessages alice "nothing else should be delivered to alice"
|
||||
liftIO $ noMessages bob "nothing else should be delivered to bob"
|
||||
@@ -2849,12 +2849,6 @@ testWaitDeliveryTimeout2 ps =
|
||||
baseId = 1
|
||||
msgId = subtract baseId
|
||||
|
||||
networkOrTimeoutError :: BrokerErrorType -> Bool
|
||||
networkOrTimeoutError = \case
|
||||
TIMEOUT -> True
|
||||
NETWORK _ -> True
|
||||
_ -> False
|
||||
|
||||
testJoinConnectionAsyncReplyErrorV8 :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
|
||||
testJoinConnectionAsyncReplyErrorV8 ps@(t, ASType qsType _) = do
|
||||
let initAgentServersSrv2 = initAgentServers {smp = userServers [testSMPServer2]}
|
||||
@@ -2981,7 +2975,7 @@ testUsersNoServer ps = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
|
||||
nGet b =##> \case ("", "", DOWN _ cs) -> length cs == 2; _ -> False
|
||||
runRight_ $ do
|
||||
deleteUser a auId True
|
||||
get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Just (BROKER _ e))]) -> c == bId' && networkOrTimeoutError e;; _ -> False
|
||||
get a =##> \case ("", "", DEL_RCVQS [(c, _, _, Just (BROKER _ e))]) -> c == bId' && (e == TIMEOUT || e == NETWORK); _ -> False
|
||||
get a =##> \case ("", "", DEL_CONNS [c]) -> c == bId'; _ -> False
|
||||
nGet a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False
|
||||
liftIO $ noMessages a "nothing else should be delivered to alice"
|
||||
@@ -3645,7 +3639,7 @@ testServerMultipleIdentities =
|
||||
exchangeGreetings alice bobId bob aliceId
|
||||
-- this saves queue with second server identity
|
||||
bob' <- liftIO $ do
|
||||
Left (BROKER _ (NETWORK _)) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo" SMSubscribe
|
||||
Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo" SMSubscribe
|
||||
disposeAgentClient bob
|
||||
threadDelay 250000
|
||||
getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
|
||||
@@ -79,7 +79,7 @@ import Simplex.Messaging.Notifications.Server.Push.APNS
|
||||
import Simplex.Messaging.Notifications.Server.Store.Postgres (closeNtfDbStore, newNtfDbStore, withDB')
|
||||
import Simplex.Messaging.Notifications.Types (NtfTknAction (..), NtfToken (..))
|
||||
import Simplex.Messaging.Parsers (parseAll)
|
||||
import Simplex.Messaging.Protocol (ErrorType (AUTH), NetworkError (..), MsgFlags (MsgFlags), NMsgMeta (..), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
|
||||
import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NMsgMeta (..), NtfServer, ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Server.Env.STM (AStoreType (..), ServerConfig (..))
|
||||
import Simplex.Messaging.Transport (ASrvTransport)
|
||||
@@ -137,7 +137,7 @@ notificationTests ps@(t, _) = do
|
||||
it "should pass" $ testRunNTFServerTests t testNtfServer `shouldReturn` Nothing
|
||||
let srv1 = testNtfServer {keyHash = "1234"}
|
||||
it "should fail with incorrect fingerprint" $ do
|
||||
testRunNTFServerTests t srv1 `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) $ NETWORK NEUnknownCAError)
|
||||
testRunNTFServerTests t srv1 `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) NETWORK)
|
||||
describe "Managing notification subscriptions" $ do
|
||||
describe "should create notification subscription for existing connection" $
|
||||
testNtfMatrix ps testNotificationSubscriptionExistingConnection
|
||||
@@ -321,7 +321,7 @@ testNtfTokenServerRestartReverify t apns = do
|
||||
runRight_ $ do
|
||||
verification <- ntfData .-> "verification"
|
||||
nonce <- C.cbNonce <$> ntfData .-> "nonce"
|
||||
Left (BROKER _ (NETWORK _)) <- tryE $ verifyNtfToken a tkn nonce verification
|
||||
Left (BROKER _ NETWORK) <- tryE $ verifyNtfToken a tkn nonce verification
|
||||
pure ()
|
||||
threadDelay 1500000
|
||||
withAgent 2 agentCfg initAgentServers testDB $ \a' ->
|
||||
@@ -478,7 +478,7 @@ testNtfTokenChangeServers t apns =
|
||||
tkn2 <- registerTestToken a "xyzw" NMInstant apns
|
||||
getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort -- not yet changed
|
||||
deleteNtfToken a tkn2 -- force server switch
|
||||
Left BROKER {brokerErr = (NETWORK _)} <- tryError $ registerTestToken a "qwer" NMInstant apns -- ok, it's down for now
|
||||
Left BROKER {brokerErr = NETWORK} <- tryError $ registerTestToken a "qwer" NMInstant apns -- ok, it's down for now
|
||||
getTestNtfTokenPort a >>= \port2 -> liftIO $ port2 `shouldBe` ntfTestPort2 -- but the token got updated
|
||||
killThread ntf
|
||||
withNtfServerOn t ntfTestPort2 ntfTestDBCfg2 $ runRight_ $ do
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE DuplicateRecordFields #-}
|
||||
{-# LANGUAGE OverloadedLists #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
|
||||
|
||||
+4
-5
@@ -23,9 +23,8 @@ 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, supportedClientSMPRelayVRange)
|
||||
import Simplex.Messaging.Transport (TLS (..), TransportPeer (..), defaultSupportedParams, defaultSupportedParamsHTTPS, simplexMQVersion, 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
|
||||
@@ -109,7 +108,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" <> simplexmqVersionCommit]
|
||||
r `shouldContain` ["SMP server v" <> simplexMQVersion]
|
||||
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"]
|
||||
@@ -217,7 +216,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" <> simplexmqVersionCommit]
|
||||
r `shouldContain` ["SMP notifications server v" <> simplexMQVersion]
|
||||
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)
|
||||
@@ -235,7 +234,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" <> simplexmqVersionCommit]
|
||||
r `shouldContain` ["SimpleX XFTP server v" <> simplexMQVersion]
|
||||
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)
|
||||
|
||||
@@ -45,32 +45,56 @@ utilTests = do
|
||||
runExceptT (throwTestException `catchError` handleCatch) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
|
||||
describe "tryAllErrors" $ do
|
||||
it "should return ExceptT error as Left" $
|
||||
runExceptT (tryAllErrors throwTestError) `shouldReturn` Right (Left (TestError "error"))
|
||||
runExceptT (tryAllErrors testErr throwTestError) `shouldReturn` Right (Left (TestError "error"))
|
||||
it "should return SomeException as Left" $
|
||||
runExceptT (tryAllErrors throwTestException) `shouldReturn` Right (Left (TestException "user error (error)"))
|
||||
runExceptT (tryAllErrors testErr throwTestException) `shouldReturn` Right (Left (TestException "user error (error)"))
|
||||
it "should return no errors as Right" $
|
||||
runExceptT (tryAllErrors noErrors) `shouldReturn` Right (Right "no errors")
|
||||
runExceptT (tryAllErrors testErr noErrors) `shouldReturn` Right (Right "no errors")
|
||||
describe "tryAllErrors specialized as tryTestError" $ do
|
||||
let tryTestError = tryAllErrors testErr
|
||||
it "should return ExceptT error as Left" $
|
||||
runExceptT (tryTestError throwTestError) `shouldReturn` Right (Left (TestError "error"))
|
||||
it "should return SomeException as Left" $
|
||||
runExceptT (tryTestError throwTestException) `shouldReturn` Right (Left (TestException "user error (error)"))
|
||||
it "should return no errors as Right" $
|
||||
runExceptT (tryTestError noErrors) `shouldReturn` Right (Right "no errors")
|
||||
describe "catchAllErrors" $ do
|
||||
it "should catch ExceptT error" $
|
||||
runExceptT (throwTestError `catchAllErrors` handleCatch) `shouldReturn` Right "caught TestError \"error\""
|
||||
runExceptT (catchAllErrors testErr throwTestError handleCatch) `shouldReturn` Right "caught TestError \"error\""
|
||||
it "should catch SomeException" $
|
||||
runExceptT (throwTestException `catchAllErrors` handleCatch) `shouldReturn` Right "caught TestException \"user error (error)\""
|
||||
runExceptT (catchAllErrors testErr throwTestException handleCatch) `shouldReturn` Right "caught TestException \"user error (error)\""
|
||||
it "should not throw if there are no errors" $
|
||||
runExceptT (noErrors `catchAllErrors` throwError) `shouldReturn` Right "no errors"
|
||||
runExceptT (catchAllErrors testErr noErrors throwError) `shouldReturn` Right "no errors"
|
||||
describe "catchAllErrors specialized as catchTestError" $ do
|
||||
let catchTestError = catchAllErrors testErr
|
||||
it "should catch ExceptT error" $
|
||||
runExceptT (throwTestError `catchTestError` handleCatch) `shouldReturn` Right "caught TestError \"error\""
|
||||
it "should catch SomeException" $
|
||||
runExceptT (throwTestException `catchTestError` handleCatch) `shouldReturn` Right "caught TestException \"user error (error)\""
|
||||
it "should not throw if there are no errors" $
|
||||
runExceptT (noErrors `catchTestError` throwError) `shouldReturn` Right "no errors"
|
||||
describe "catchThrow" $ do
|
||||
it "should re-throw ExceptT error" $
|
||||
runExceptT (throwTestError `catchThrow` fromSomeException) `shouldReturn` Left (TestError "error")
|
||||
runExceptT (throwTestError `catchThrow` testErr) `shouldReturn` Left (TestError "error")
|
||||
it "should catch SomeException and throw as ExceptT error" $
|
||||
runExceptT (throwTestException `catchThrow` fromSomeException) `shouldReturn` Left (TestException "user error (error)")
|
||||
runExceptT (throwTestException `catchThrow` testErr) `shouldReturn` Left (TestException "user error (error)")
|
||||
it "should not throw if there are no exceptions" $
|
||||
runExceptT (noErrors `catchThrow` fromSomeException) `shouldReturn` Right "no errors"
|
||||
runExceptT (noErrors `catchThrow` testErr) `shouldReturn` Right "no errors"
|
||||
describe "allFinally should run final action" $ do
|
||||
it "then throw ExceptT error" $ withFinal $ \final ->
|
||||
runExceptT (throwTestError `allFinally` final) `shouldReturn` Left (TestError "error")
|
||||
runExceptT (allFinally testErr throwTestError final) `shouldReturn` Left (TestError "error")
|
||||
it "then throw SomeException as ExceptT error" $ withFinal $ \final ->
|
||||
runExceptT (throwTestException `allFinally` final) `shouldReturn` Left (TestException "user error (error)")
|
||||
runExceptT (allFinally testErr throwTestException final) `shouldReturn` Left (TestException "user error (error)")
|
||||
it "and should not throw if there are no exceptions" $ withFinal $ \final ->
|
||||
runExceptT (noErrors `allFinally` final) `shouldReturn` Right "no errors"
|
||||
runExceptT (allFinally testErr noErrors final) `shouldReturn` Right "no errors"
|
||||
describe "allFinally specialized as testFinally should run final action" $ do
|
||||
let testFinally = allFinally testErr
|
||||
it "then throw ExceptT error" $ withFinal $ \final ->
|
||||
runExceptT (throwTestError `testFinally` final) `shouldReturn` Left (TestError "error")
|
||||
it "then throw SomeException as ExceptT error" $ withFinal $ \final ->
|
||||
runExceptT (throwTestException `testFinally` final) `shouldReturn` Left (TestException "user error (error)")
|
||||
it "and should not throw if there are no exceptions" $ withFinal $ \final ->
|
||||
runExceptT (noErrors `testFinally` final) `shouldReturn` Right "no errors"
|
||||
where
|
||||
throwTestError :: ExceptT TestError IO String
|
||||
throwTestError = throwError $ TestError "error"
|
||||
@@ -78,6 +102,8 @@ utilTests = do
|
||||
throwTestException = liftIO $ throwIO $ userError "error"
|
||||
noErrors :: ExceptT TestError IO String
|
||||
noErrors = pure "no errors"
|
||||
testErr :: SomeException -> TestError
|
||||
testErr = TestException . show
|
||||
handleCatch :: TestError -> ExceptT TestError IO String
|
||||
handleCatch e = pure $ "caught " <> show e
|
||||
handleException :: SomeException -> ExceptT TestError IO String
|
||||
@@ -93,6 +119,3 @@ data TestError = TestError String | TestException String
|
||||
deriving (Eq, Show)
|
||||
|
||||
instance Exception TestError
|
||||
|
||||
instance AnyError TestError where
|
||||
fromSomeException = TestException . show
|
||||
|
||||
@@ -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; /^\\\\restrict/d; /^\\\\unrestrict/d' " <> schemaPath) ""
|
||||
void $ readCreateProcess (shell $ sed <> " '/^--/d' " <> schemaPath) ""
|
||||
sch <- readFile schemaPath
|
||||
sch `deepseq` pure sch
|
||||
|
||||
@@ -229,7 +229,6 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
|
||||
dailyBlockQueueQuota = 20,
|
||||
messageExpiration = Just defaultMessageExpiration,
|
||||
expireMessagesOnStart = True,
|
||||
expireMessagesOnSend = False,
|
||||
idleQueueInterval = defaultIdleQueueInterval,
|
||||
notificationExpiration = defaultNtfExpiration,
|
||||
inactiveClientExpiration = Just defaultInactiveClientExpiration,
|
||||
|
||||
+9
-59
@@ -86,7 +86,6 @@ serverTests = do
|
||||
describe "Message notifications" $ do
|
||||
testMessageNotifications
|
||||
testMessageServiceNotifications
|
||||
testServiceNotificationsTwoRestarts
|
||||
describe "Message expiration" $ do
|
||||
testMsgExpireOnSend
|
||||
testMsgExpireOnInterval
|
||||
@@ -955,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 msType
|
||||
testSameTiming rh sh tst
|
||||
where
|
||||
testName :: (C.AuthAlg, C.AuthAlg, Int) -> String
|
||||
testName (C.AuthAlg goodKeyAlg, C.AuthAlg badKeyAlg, _) = unwords ["queue key:", show goodKeyAlg, "/ used key:", show badKeyAlg]
|
||||
@@ -972,16 +971,11 @@ 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 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
|
||||
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
|
||||
g <- C.newRandom
|
||||
(rPub, rKey) <- atomically $ C.generateAuthKeyPair goodKeyAlg g
|
||||
(dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
|
||||
@@ -1016,7 +1010,7 @@ testTiming =
|
||||
timeNoQueue <- timeRepeat n $ do
|
||||
Resp "dabc" _ (ERR AUTH) <- signSendRecv h badKey ("dabc", EntityId "1234", cmd)
|
||||
return ()
|
||||
let ok = similarTime timeNoQueue timeWrongKey msType
|
||||
let ok = similarTime timeNoQueue timeWrongKey
|
||||
unless ok . putStrLn . unwords $
|
||||
[ show goodKeyAlg,
|
||||
show badKeyAlg,
|
||||
@@ -1092,6 +1086,7 @@ testMessageServiceNotifications =
|
||||
(rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
|
||||
Resp "1" _ (NID nId _) <- signSendRecv rh rKey ("1", rId, NKEY nPub rcvNtfPubDhKey)
|
||||
serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g
|
||||
-- TODO [certs] we need to get certificate fingerprint and include it into signed over for NSUB commands
|
||||
testNtfServiceClient t serviceKeys $ \nh1 -> do
|
||||
-- can't subscribe without service signature in service connection
|
||||
Resp "2a" _ (ERR SERVICE) <- signSendRecv nh1 nKey ("2a", nId, NSUB)
|
||||
@@ -1155,57 +1150,12 @@ testMessageServiceNotifications =
|
||||
Resp "" _ (NMSG _ _) <- tGet1 nh
|
||||
pure ()
|
||||
|
||||
testServiceNotificationsTwoRestarts :: SpecWith (ASrvTransport, AStoreType)
|
||||
testServiceNotificationsTwoRestarts =
|
||||
it "subscribe notifier as service and deliver notifications after two restarts" $ \ps@(ATransport t, _) -> do
|
||||
g <- C.newRandom
|
||||
(sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
|
||||
(nPub, nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
|
||||
serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g
|
||||
(rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
|
||||
(rId, rKey, sId, dec, serviceId) <- withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh -> do
|
||||
(sId, rId, rKey, dhShared) <- createAndSecureQueue rh sPub
|
||||
let dec = decryptMsgV3 dhShared
|
||||
Resp "0" _ (NID nId _) <- signSendRecv rh rKey ("0", rId, NKEY nPub rcvNtfPubDhKey)
|
||||
testNtfServiceClient t serviceKeys $ \nh -> do
|
||||
Resp "1" _ (SOK (Just serviceId)) <- serviceSignSendRecv nh nKey servicePK ("1", nId, NSUB)
|
||||
deliverMessage rh rId rKey sh sId sKey nh "hello" dec
|
||||
pure (rId, rKey, sId, dec, serviceId)
|
||||
threadDelay 250000
|
||||
withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh ->
|
||||
testNtfServiceClient t serviceKeys $ \nh -> do
|
||||
Resp "2.1" serviceId' (SOKS n) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("2.1", serviceId, NSUBS)
|
||||
n `shouldBe` 1
|
||||
Resp "2.2" _ (SOK Nothing) <- signSendRecv rh rKey ("2.2", rId, SUB)
|
||||
serviceId' `shouldBe` serviceId
|
||||
deliverMessage rh rId rKey sh sId sKey nh "hello 2" dec
|
||||
threadDelay 250000
|
||||
withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh ->
|
||||
testNtfServiceClient t serviceKeys $ \nh -> do
|
||||
Resp "3.1" _ (SOKS n) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("3.1", serviceId, NSUBS)
|
||||
n `shouldBe` 1
|
||||
Resp "3.2" _ (SOK Nothing) <- signSendRecv rh rKey ("3.2", rId, SUB)
|
||||
deliverMessage rh rId rKey sh sId sKey nh "hello 3" dec
|
||||
where
|
||||
runTest2 :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO a) -> ThreadId -> IO a
|
||||
runTest2 _ test' server = do
|
||||
a <- testSMPClient $ \h1 -> testSMPClient $ \h2 -> test' h1 h2
|
||||
killThread server
|
||||
pure a
|
||||
deliverMessage rh rId rKey sh sId sKey nh msgText dec = do
|
||||
Resp "msg-1" _ OK <- signSendRecv sh sKey ("msg-1", sId, _SEND' msgText)
|
||||
Resp "" _ (Msg mId msg) <- tGet1 rh
|
||||
Resp "msg-2" _ OK <- signSendRecv rh rKey ("msg-2", rId, ACK mId)
|
||||
(dec mId msg, Right msgText) #== "delivered from queue"
|
||||
Resp "" _ (NMSG _ _) <- tGet1 nh
|
||||
pure ()
|
||||
|
||||
testMsgExpireOnSend :: SpecWith (ASrvTransport, AStoreType)
|
||||
testMsgExpireOnSend =
|
||||
it "should expire messages that are not received before messageTTL on SEND" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do
|
||||
g <- C.newRandom
|
||||
(sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
|
||||
let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {expireMessagesOnSend = True, messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
|
||||
let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
|
||||
withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
|
||||
testSMPClient @c $ \sh -> do
|
||||
(sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
|
||||
|
||||
+4
-4
@@ -102,7 +102,7 @@ main = do
|
||||
] -- skipComparisonForDownMigrations
|
||||
testStoreDBOpts
|
||||
"src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql"
|
||||
around_ (postgressBracket testServerDBConnectInfo) $
|
||||
aroundAll_ (postgressBracket testServerDBConnectInfo) $
|
||||
describe "SMP server via TLS, postgres+jornal message store" $
|
||||
before (pure (transport @TLS, ASType SQSPostgres SMSJournal)) serverTests
|
||||
#endif
|
||||
@@ -122,13 +122,13 @@ main = do
|
||||
[] -- skipComparisonForDownMigrations
|
||||
ntfTestStoreDBOpts
|
||||
"src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql"
|
||||
around_ (postgressBracket ntfTestServerDBConnectInfo) $ do
|
||||
aroundAll_ (postgressBracket ntfTestServerDBConnectInfo) $ do
|
||||
describe "Notifications server (SMP server: jornal store)" $
|
||||
ntfServerTests (transport @TLS, ASType SQSMemory SMSJournal)
|
||||
around_ (postgressBracket testServerDBConnectInfo) $
|
||||
aroundAll_ (postgressBracket testServerDBConnectInfo) $
|
||||
describe "Notifications server (SMP server: postgres+jornal store)" $
|
||||
ntfServerTests (transport @TLS, ASType SQSPostgres SMSJournal)
|
||||
around_ (postgressBracket testServerDBConnectInfo) $ do
|
||||
aroundAll_ (postgressBracket testServerDBConnectInfo) $ do
|
||||
describe "SMP client agent, postgres+jornal message store" $ agentTests (transport @TLS, ASType SQSPostgres SMSJournal)
|
||||
describe "SMP proxy, postgres+jornal message store" $
|
||||
before (pure $ ASType SQSPostgres SMSJournal) smpProxyTests
|
||||
|
||||
+2
-2
@@ -38,7 +38,7 @@ import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs)
|
||||
import qualified Simplex.Messaging.Crypto.File as CF
|
||||
import Simplex.Messaging.Encoding.String (StrEncoding (..))
|
||||
import Simplex.Messaging.Protocol (BasicAuth, NetworkError (..), ProtoServerWithAuth (..), ProtocolServer (..), XFTPServerWithAuth)
|
||||
import Simplex.Messaging.Protocol (BasicAuth, ProtoServerWithAuth (..), ProtocolServer (..), XFTPServerWithAuth)
|
||||
import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
|
||||
import Simplex.Messaging.Util (tshow)
|
||||
import System.Directory (doesDirectoryExist, doesFileExist, getFileSize, listDirectory, removeFile)
|
||||
@@ -84,7 +84,7 @@ xftpAgentTests =
|
||||
it "should pass without basic auth" $ testXFTPServerTest Nothing (noAuthSrv testXFTPServer2) `shouldReturn` Nothing
|
||||
let srv1 = testXFTPServer2 {keyHash = "1234"}
|
||||
it "should fail with incorrect fingerprint" $ do
|
||||
testXFTPServerTest Nothing (noAuthSrv srv1) `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) $ NETWORK NEUnknownCAError)
|
||||
testXFTPServerTest Nothing (noAuthSrv srv1) `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) NETWORK)
|
||||
describe "server with password" $ do
|
||||
let auth = Just "abcd"
|
||||
srv = ProtoServerWithAuth testXFTPServer2
|
||||
|
||||
+1
-6
@@ -99,9 +99,6 @@ 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
|
||||
@@ -130,8 +127,6 @@ 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
|
||||
}
|
||||
@@ -145,6 +140,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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user