Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 3 additions & 4 deletions src-tauri/src/sync2/live.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,14 @@ use tokio_tungstenite::tungstenite::{Bytes, Message};
use super::actions::{local, stopped};
use super::engine::Result;
use super::live_hub::{announce, receive, set_server_online, subscribe};
use super::live_seal::MAX_LIVE_MESSAGE;
use super::sealed::random_bytes;
use super::server::{self, ServerStore, StoreError};
use super::service::sync_after;
use super::state::{StoreKind, load_group};
use crate::sync::error::SyncError;

const PROTOCOL: &str = "flow-sync.v1";
/// The most a live message or a notice may hold (spec/api.md, Limits).
const MAX_MESSAGE: usize = 64 * 1024;
const PING_EVERY: Duration = Duration::from_secs(15);
const RECONNECT_AFTER: Duration = Duration::from_secs(20);
/// Several devices may store at once: one sync shortly after covers them all.
Expand Down Expand Up @@ -77,8 +76,8 @@ impl ServerStore {
}
let upgraded = response.upgrade().await.map_err(StoreError::Network)?;
let config = WebSocketConfig::default()
.max_message_size(Some(MAX_MESSAGE))
.max_frame_size(Some(MAX_MESSAGE));
.max_message_size(Some(MAX_LIVE_MESSAGE))
.max_frame_size(Some(MAX_LIVE_MESSAGE));
Ok(WebSocketStream::from_raw_socket(upgraded, Role::Client, Some(config)).await)
}
}
Expand Down
36 changes: 36 additions & 0 deletions src-tauri/src/sync2/live_seal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,9 @@ use super::roster::Role;
use super::store::Store;
use crate::sync::canonical::to_canonical_json;

/// The most a live message or a notice may hold, on any transport (spec/store.md, Untrusted input).
pub const MAX_LIVE_MESSAGE: usize = 64 * 1024;

/// What sealing and opening live messages needs, taken from a session when it opened.
#[derive(Clone)]
pub struct LiveKeys {
Expand Down Expand Up @@ -83,6 +86,9 @@ impl LiveKeys {
/// Checks and opens a message from another current device. The caller still checks that its
/// counter is above the last one accepted from that device.
pub fn open(&self, bytes: &[u8]) -> Result<Received> {
if bytes.len() > MAX_LIVE_MESSAGE {
return Err(Error::Length.into());
}
let object = parse_object(bytes, &self.group)?;
let from = hex::encode(object.header.sender);
let key = self.others.get(&from).ok_or(Error::Sender)?;
Expand Down Expand Up @@ -113,6 +119,14 @@ mod tests {
use super::*;
use crate::sync2::fuzz::{json, member, mutations, seq};
use crate::sync2::live_message::{Answer, DeviceKind};
use crate::sync2::sealed::random_bytes;

fn random_bytes_of(len: usize) -> Vec<u8> {
std::iter::repeat_with(random_bytes::<32>)
.flatten()
.take(len)
.collect()
}

/// Two devices of one group, each with the other's signing key.
static PAIR: LazyLock<(LiveKeys, LiveKeys)> = LazyLock::new(|| {
Expand Down Expand Up @@ -152,6 +166,28 @@ mod tests {
]
}

#[test]
fn a_live_message_over_the_limit_is_refused_on_any_transport() {
let (a, b) = &*PAIR;
let with_reason = |bytes: usize| {
a.seal(
1,
&LiveMessage::Result(Answer {
to: b.me(),
reference: 1,
ok: false,
reason: Some(hex::encode(random_bytes_of(bytes))),
}),
)
.unwrap()
};
let small = with_reason(1024);
assert!(b.open(&small).is_ok());
let large = with_reason(128 * 1024);
assert!(large.len() > MAX_LIVE_MESSAGE);
assert!(b.open(&large).is_err());
}

proptest! {
#[test]
fn live_messages_round_trip(seq in seq(), message in message()) {
Expand Down
Loading