diff --git a/docs/testing/determinism-audit.md b/docs/testing/determinism-audit.md index 1ba6693bf85..e3c3dd0f542 100644 --- a/docs/testing/determinism-audit.md +++ b/docs/testing/determinism-audit.md @@ -403,7 +403,7 @@ and the simulation guide. | `RemoteMap`, `RemoteStateActor`, net-report, protocol router, and relay actor `JoinSet::spawn` calls | **Architectural problem** | Replace direct Tokio `JoinSet` ownership in simulator-supported paths with an environment-owned task group abstraction. | | `krikos-relay/src/client/tls.rs:200`, `quic.rs:124`, `server.rs:772,810,860`, `server/client.rs:146`, `server/http_server/listener.rs:323,350`, `server/http_server/upgrade.rs:113`, `server/resolver.rs:79` | **Architectural problem** | Add relay runtime/task capabilities before claiming deterministic relay coverage. | | DNS/HTTP server task sets and `krikos-dns-server/src/http/transport.rs` | **Acceptable nondeterminism** in the real-server backend | Listener and connection work is now owned by supervisors, bounded before spawn, and cancelled/drained on shutdown. Runtime/I/O injection remains required before full DNS-server simulation. | -| Direct spawns in `#[cfg(test)]` regions, `tests/`, `examples/`, and `krikos/bench` | **Acceptable nondeterminism** | Do not route through production environment unless the scenario runner reuses that code. | +| Direct spawns in `#[cfg(test)]` regions, `tests/`, `examples/`, and `krikos/bench` | **Acceptable nondeterminism** | Do not route through production environment unless the scenario runner reuses that code. Includes `krikos-relay/src/server/client.rs::tests::undeliverable_packet_does_not_end_the_session`, which drives one client actor over a `tokio::io::duplex` pair: the spawn owns only the actor under test, the test joins it before returning, and no simulator backend executes this module. | | `krikos-relay/src/main.rs:627` certificate file `spawn_blocking` | **Acceptable nondeterminism** | Keep in production binary; exclude from in-process simulation. | | DNS-server bounded per-IP token buckets | **Behavioral randomness** with explicit transition input | The LRU has a validated hard capacity; token transitions receive `Instant` explicitly. Only the HTTP middleware samples the production clock. There is no GC thread. | diff --git a/krikos-dns-server/src/metrics.rs b/krikos-dns-server/src/metrics.rs index 23e4e0d9825..d3812d87147 100644 --- a/krikos-dns-server/src/metrics.rs +++ b/krikos-dns-server/src/metrics.rs @@ -67,6 +67,11 @@ pub struct Metrics { pub store_corrupt_rows: Counter, /// Number of packet-store actor or eviction failures. pub store_background_failures: Counter, + /// Number of DHT items discarded because they were not signed by the queried key. + /// + /// Anything but zero means someone is publishing forged items under a key this + /// server is asked to resolve. + pub dht_packets_rejected: Counter, /// Current number of zones in the main cache pub cache_zones: Gauge, /// Current number of zones in the DHT cache diff --git a/krikos-dns-server/src/store.rs b/krikos-dns-server/src/store.rs index fbb76f39f73..5b000d45478 100644 --- a/krikos-dns-server/src/store.rs +++ b/krikos-dns-server/src/store.rs @@ -163,17 +163,30 @@ impl ZoneStore { .as_async() .get_mutable_most_recent(pubkey.as_bytes(), None) .await; - if let Some(item) = maybe_item - && let Ok(packet) = mutable_item_to_signed_packet(&item) - { - debug!("DHT resolve successful {:?}", packet); - return self - .cache - .lock() - .await - .insert_and_resolve_dht(&packet, name, record_type); + match maybe_item { + // Keep the rejection distinguishable from a plain miss: a packet that + // was returned but does not verify against the key we asked for is the + // observable trace of an attempt to poison this resolver, and it is the + // only signal the verification produces. + Some(item) => match mutable_item_to_signed_packet(&item, pubkey) { + Ok(packet) => { + debug!("DHT resolve successful {:?}", packet); + return self.cache.lock().await.insert_and_resolve_dht( + &packet, + name, + record_type, + ); + } + Err(err) => { + self.metrics.dht_packets_rejected.inc(); + warn!( + pubkey = %pubkey.to_z32(), + "DHT returned a packet that does not verify against the requested key: {err:#}" + ); + } + }, + None => debug!("DHT resolve failed"), } - debug!("DHT resolve failed"); } Ok(None) } @@ -215,17 +228,27 @@ fn finish_mainline_build(result: std::io::Result) -> Result { result.anyerr() } -/// Convert a mainline [`MutableItem`] to a [`SignedPacket`]. +/// Convert a mainline [`MutableItem`] to a [`SignedPacket`], for the key we asked for. +/// +/// `expected` is the key the lookup was issued for, and the packet is verified against +/// it. This binding cannot be left to `mainline`: it checks a mutable item's signature +/// against the key carried in the *response* and then stores the queried target +/// verbatim, so all it establishes is that some key signed the value. Without the check +/// here, any DHT node could answer a lookup with a packet signed under a key it +/// controls, and those records would be re-served under the queried name. fn mutable_item_to_signed_packet( item: &MutableItem, + expected: &PublicKeyBytes, ) -> Result { let timestamp = u64::try_from(item.seq()).map_err(|_| { n0_error::e!(SignedPacketVerifyError::InvalidTimestamp { timestamp: item.seq(), }) })?; - SignedPacket::from_parts_unchecked( - item.key(), + // Built from `expected` rather than `item.key()`, so a packet signed by any other + // key fails signature verification instead of being silently re-attributed. + SignedPacket::from_parts( + expected.as_bytes(), item.signature(), Timestamp::from_micros(timestamp), item.value(), @@ -404,6 +427,41 @@ mod tests { Ok(()) } + /// A DHT response must be bound to the key we asked for. + /// + /// `mainline` verifies a mutable item's signature against the key carried in the + /// response, not against the queried target, and stores the target verbatim — so + /// "signed by somebody" is all it guarantees. Any DHT node can therefore answer a + /// lookup for one key with a packet correctly signed under a key it controls. Were + /// that accepted, its records would be re-served under the queried name, which is + /// exactly the binding pkarr exists to provide. + #[test] + fn dht_item_signed_by_another_key_is_rejected() -> Result { + let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(1); + // A perfectly valid packet — just not the one that was asked for. + let attacker_packet = random_signed_packet(&mut rng)?; + let attacker_key = PublicKeyBytes::from_signed_packet(&attacker_packet); + // pkarr carries the packet timestamp as the BEP-0044 sequence number. + let item = MutableItem::new_signed_unchecked( + *attacker_key.as_bytes(), + attacker_packet.signature().to_bytes(), + attacker_packet.encoded_packet(), + i64::try_from(attacker_packet.timestamp().as_micros()) + .expect("pkarr timestamp fits in a mainline sequence number"), + None, + ); + + // Resolved under the key that actually signed it: accepted. + mutable_item_to_signed_packet(&item, &attacker_key) + .expect("a packet signed by the key we asked for must be accepted"); + + // Resolved under someone else's key: refused. + let victim = PublicKeyBytes::from_signed_packet(&random_signed_packet(&mut rng)?); + mutable_item_to_signed_packet(&item, &victim) + .expect_err("a packet signed by an unrelated key must be refused"); + Ok(()) + } + #[test] fn mainline_construction_failure_is_reported() { let error = finish_mainline_build(Err(std::io::Error::other( @@ -419,16 +477,19 @@ mod tests { } #[test] - fn negative_mainline_sequence_is_rejected() { - let item = MutableItem::new_signed_unchecked([0; 32], [0; 64], &[], -1, None); + fn negative_mainline_sequence_is_rejected() -> Result { + let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(2); + let key = PublicKeyBytes::from_signed_packet(&random_signed_packet(&mut rng)?); + let item = MutableItem::new_signed_unchecked(*key.as_bytes(), [0; 64], &[], -1, None); - let error = mutable_item_to_signed_packet(&item) + let error = mutable_item_to_signed_packet(&item, &key) .expect_err("negative mainline sequence must not wrap into a timestamp"); assert!(matches!( error, SignedPacketVerifyError::InvalidTimestamp { timestamp: -1, .. } )); + Ok(()) } fn random_signed_packet(rng: &mut R) -> Result { diff --git a/krikos-dns/src/pkarr.rs b/krikos-dns/src/pkarr.rs index ed267ba7ce7..4a03feb7411 100644 --- a/krikos-dns/src/pkarr.rs +++ b/krikos-dns/src/pkarr.rs @@ -130,6 +130,29 @@ impl SignedPacket { Self::from_bytes(&bytes) } + /// Assemble a signed packet from its parts, verifying it against `public_key`. + /// + /// Use this over [`SignedPacket::from_parts_unchecked`] whenever the parts come from + /// somewhere that does not itself bind them to the key being resolved, such as a DHT + /// response. + /// Both fixed-size arrays: the wire layout is `public_key || signature || timestamp + /// || packet` at fixed offsets, so a signature of any other length would silently + /// carve the timestamp and the packet out of the wrong bytes and report a + /// signature-verification failure instead of a length error. + pub fn from_parts( + public_key: &[u8; 32], + signature: &[u8; 64], + timestamp: Timestamp, + encoded_packet: &[u8], + ) -> Result { + let mut bytes = Vec::with_capacity(HEADER_SIZE + encoded_packet.len()); + bytes.extend_from_slice(public_key); + bytes.extend_from_slice(signature); + bytes.extend_from_slice(×tamp.to_be_bytes()); + bytes.extend_from_slice(encoded_packet); + Self::from_bytes(&bytes) + } + /// Parse a signed packet without verifying the signature. /// /// Still validates minimum length and DNS parsing. @@ -253,8 +276,12 @@ impl SignedPacket { /// Reconstruct a signed packet from its raw parts without verifying the signature. /// - /// This is useful for reconstructing a packet from storage or DHT mutable items - /// where the components are stored separately. + /// Only for parts that are already bound to the key they will be served under, such + /// as rows of a store this node wrote itself. Never for anything a peer supplied: + /// the parts of a DHT mutable item carry the *publisher's* key and signature, not + /// the key that was queried, so reassembling one here re-serves an attacker-signed + /// packet under the name that was looked up. Use [`SignedPacket::from_parts`] for + /// those, which verifies against the key you pass in. pub fn from_parts_unchecked( public_key: &[u8], signature: &[u8], diff --git a/krikos-relay/src/protos/relay.rs b/krikos-relay/src/protos/relay.rs index 3dbebcf89b1..72476a2f75a 100644 --- a/krikos-relay/src/protos/relay.rs +++ b/krikos-relay/src/protos/relay.rs @@ -549,10 +549,30 @@ impl ClientToRelayMsg { ensure!(content.len() >= EndpointId::LENGTH, Error::InvalidFrame); let dst_endpoint_id = cache.key_from_slice(&content[..EndpointId::LENGTH])?; - let datagrams = Datagrams::from_bytes( - content.slice(EndpointId::LENGTH..), - frame_type == FrameType::ClientToRelayDatagramBatch, - )?; + let is_batch = frame_type == FrameType::ClientToRelayDatagramBatch; + let datagrams = + Datagrams::from_bytes(content.slice(EndpointId::LENGTH..), is_batch)?; + + // Only accept what can be forwarded. The relay re-encodes this as a + // `RelayToClientMsg::Datagrams`, which carries its own frame type byte on + // top of the same payload, so the egress frame is larger than the one + // checked above. The egress encoder also refuses empty contents. Either + // refusal would surface on the *receiving* client's stream, i.e. on a + // peer that did nothing, so both have to be caught here at ingress. + ensure!(!datagrams.contents.is_empty(), Error::InvalidFrame); + let forwarded_len = if is_batch { + FrameType::RelayToClientDatagramBatch.encoded_len() + } else { + FrameType::RelayToClientDatagram.encoded_len() + } + EndpointId::LENGTH + + datagrams.encoded_len(); + ensure!( + forwarded_len <= MAX_PACKET_SIZE, + Error::FrameTooLarge { + frame_len: forwarded_len + } + ); + Self::Datagrams { dst_endpoint_id, datagrams, @@ -880,6 +900,77 @@ mod tests { assert_eq!(&encoded[5..9], &u32::MAX.to_be_bytes()); } + /// Anything the ingress parser accepts must be re-encodable towards its destination. + /// + /// The relay forwards a `ClientToRelayMsg::Datagrams` on as a + /// `RelayToClientMsg::Datagrams`, which is one frame-type byte larger because it + /// carries the sender's id in place of the destination's. `RelayedStream::start_send` + /// refuses to encode an empty or oversized frame — and that refusal lands on the + /// *receiving* client, whose actor treats a send failure as fatal. So any frame the + /// ingress parser lets through but the egress encoder rejects is a frame one client + /// can aim at another. Accept only what can be forwarded. + #[test] + fn accepted_client_datagrams_are_re_encodable() { + let dst = SecretKey::from_bytes(&[7u8; 32]).public(); + let src = SecretKey::from_bytes(&[8u8; 32]).public(); + // Number of frames that made it past the ingress parser, to prove the boundary + // lengths below were actually exercised rather than skipped by the `continue`. + let mut accepted = 0; + for is_batch in [false, true] { + // The largest payload whose *ingress* frame still fits, and its neighbours: + // the egress frame is one byte longer, so the boundary cannot be the same. A + // batch also carries a two-byte segment size, so its boundary sits two bytes + // lower; using the packed layout's boundary for both would push the batch + // cases past the plain `frame_len` check and never reach the new one. + let max_payload = MAX_PACKET_SIZE + - EndpointId::LENGTH + - 1 /* ECN */ + - if is_batch { 2 /* segment size */ } else { 0 }; + for len in [0, 1, max_payload - 1, max_payload] { + let datagrams = Datagrams { + ecn: None, + segment_size: is_batch.then(|| NonZeroU16::new(1).unwrap()), + contents: vec![0u8; len].into(), + }; + let encoded = ClientToRelayMsg::Datagrams { + dst_endpoint_id: dst, + datagrams, + } + .write_to(Vec::new()); + let Ok(ClientToRelayMsg::Datagrams { datagrams, .. }) = + ClientToRelayMsg::from_bytes(encoded.into(), &KeyCache::test()) + else { + // Refused at ingress: nothing to forward, which is the safe outcome. + continue; + }; + accepted += 1; + assert!( + !datagrams.contents.is_empty(), + "accepted an empty datagram (len {len}, batch {is_batch}); \ + forwarding it fails on the receiver's stream" + ); + let forwarded = RelayToClientMsg::Datagrams { + remote_endpoint_id: src, + datagrams, + }; + assert!( + forwarded.encoded_len() <= MAX_PACKET_SIZE, + "accepted a datagram that does not fit once forwarded: \ + {} bytes (len {len}, batch {is_batch})", + forwarded.encoded_len() + ); + } + } + // Per layout: `1` and `max_payload - 1` are forwardable, `0` is empty and + // `max_payload` overflows by the egress frame-type byte. If this count drops, + // the boundary arithmetic above has drifted and the cases that matter are being + // skipped instead of checked. + assert_eq!( + accepted, 4, + "expected both forwardable boundary lengths of both layouts to be accepted" + ); + } + /// A datagram frame must contain at least an EndpointId (32 bytes) after /// the frame type. A frame consisting only of the frame type byte used to /// panic when slicing the destination endpoint id. @@ -926,13 +1017,31 @@ mod proptests { }) } + /// Any datagram, including an empty one. + /// + /// The relay-to-client direction has no non-empty rule — a relay can send a + /// zero-length datagram and the client will decode it — so that direction has to + /// keep generating them. fn datagrams() -> impl Strategy { + datagrams_of_len(0) + } + + /// Datagrams a client is allowed to send to a relay. + /// + /// Non-empty: an empty datagram carries nothing and cannot be encoded towards a + /// client, so the client-to-relay parser rejects it rather than handing the + /// receiver's stream a frame it will refuse. + fn forwardable_datagrams() -> impl Strategy { + datagrams_of_len(1) + } + + fn datagrams_of_len(min_len: usize) -> impl Strategy { // The max payload size (conservatively, since with segment_size = 0 we'd have slightly more space) const MAX_PAYLOAD_SIZE: usize = MAX_PACKET_SIZE - EndpointId::LENGTH - 1 /* ECN bytes */ - 2 /* segment size */; ( ecn(), prop::option::of(MAX_PAYLOAD_SIZE / 20..MAX_PAYLOAD_SIZE), - vec(any::(), 0..MAX_PAYLOAD_SIZE), + vec(any::(), min_len..MAX_PAYLOAD_SIZE), ) .prop_map(|(ecn, segment_size, data)| Datagrams { ecn, @@ -982,12 +1091,13 @@ mod proptests { } fn client_relay_frame() -> impl Strategy { - let send_packet = (key(), datagrams()).prop_map(|(dst_endpoint_id, datagrams)| { - ClientToRelayMsg::Datagrams { - dst_endpoint_id, - datagrams, - } - }); + let send_packet = + (key(), forwardable_datagrams()).prop_map(|(dst_endpoint_id, datagrams)| { + ClientToRelayMsg::Datagrams { + dst_endpoint_id, + datagrams, + } + }); let ping = prop::array::uniform8(any::()).prop_map(ClientToRelayMsg::Ping); let pong = prop::array::uniform8(any::()).prop_map(ClientToRelayMsg::Pong); prop_oneof![send_packet, ping, pong] diff --git a/krikos-relay/src/server/client.rs b/krikos-relay/src/server/client.rs index ff1eeb3bab4..084c376383d 100644 --- a/krikos-relay/src/server/client.rs +++ b/krikos-relay/src/server/client.rs @@ -510,7 +510,27 @@ where } Err(err) => { self.metrics.send_packets_dropped.inc(); - Err(err) + // A packet the encoder refuses is not this client's doing: packets on + // this queue come from other clients. Dropping it keeps the blast radius + // at one datagram, matching how every other forwarding failure is + // handled; returning the error would end *this* client's session and let + // any peer disconnect any other. Only genuine stream errors, which mean + // this connection is already broken, are fatal. + // + // Both enums are `#[non_exhaustive]`, so the arms are arranged to make + // that invariant hold for variants that do not exist yet: a new + // `WriteFrameError` defaults to fatal, a new encode-side `SendError` + // defaults to a drop. + match &err { + WriteFrameError::Stream { source, .. } => match source { + RelaySendError::StreamError { .. } => Err(err), + _ => { + warn!("dropping undeliverable packet: {err:#}"); + Ok(()) + } + }, + _ => Err(err), + } } } } @@ -784,6 +804,87 @@ mod tests { Ok(()) } + /// A packet that cannot be encoded must not end the receiving client's session. + /// + /// Packets on this queue were put there by *other* clients, so a packet the encoder + /// refuses is not this client's doing. Every other forwarding failure already + /// logs-and-continues; treating this one as fatal would let any peer disconnect any + /// other by sending it something unencodable. Drop the packet, keep the session. + #[tokio::test] + #[traced_test] + async fn undeliverable_packet_does_not_end_the_session() -> Result { + let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(11); + let (send_queue_s, send_queue_r) = mpsc::channel(10); + let (message_s, message_r) = mpsc::channel(10); + let endpoint_id = SecretKey::from_bytes(&rng.random()).public(); + let (io, io_rw) = tokio::io::duplex(1024); + let mut io_rw = Conn::test(io_rw, Default::default()); + let stream = RelayedStream::test(io); + let clients = Clients::default(); + let metrics = Arc::new(Metrics::default()); + let actor = Actor { + stream, + timeout: Duration::from_secs(1), + packet_send_queue: send_queue_r, + message_send_queue: message_r, + guard: Some(OnDisconnectGuard::empty(endpoint_id)), + endpoint_id, + registered: false, + clients: clients.clone(), + client_counter: ClientCounter::new(clients.runtime().wall_clock()), + ping_tracker: PingTracker::default(), + metrics: metrics.clone(), + clock: clients.runtime().clock(), + }; + let done = CancellationToken::new(); + let io_done = done.clone(); + let handle = tokio::task::spawn(async move { actor.run(io_done).await }); + + // `RelayedStream` refuses to encode an empty datagram. + send_queue_s + .send(Packet { + src: endpoint_id, + data: Datagrams::from(&[][..]), + }) + .await + .std_context("send")?; + // Queued behind it: if the actor survived, this still arrives. + let data = b"still connected"; + send_queue_s + .send(Packet { + src: endpoint_id, + data: Datagrams::from(&data[..]), + }) + .await + .std_context("send")?; + + let frame = recv_frame(FrameType::RelayToClientDatagram, &mut io_rw) + .await + .anyerr()?; + assert_eq!( + frame, + RelayToClientMsg::Datagrams { + remote_endpoint_id: endpoint_id, + datagrams: data.to_vec().into() + } + ); + assert!( + !handle.is_finished(), + "an undeliverable packet must not end the session" + ); + assert_eq!( + metrics.send_packets_dropped.get(), + 1, + "the undeliverable packet should be counted as dropped" + ); + + done.cancel(); + drop(send_queue_s); + drop(message_s); + handle.await.std_context("join")?; + Ok(()) + } + #[tokio::test(start_paused = true)] #[traced_test] async fn client_actor_survives_production_keepalive_horizon_when_polled() -> Result { diff --git a/protocols/krikos-blobs/examples/limit.rs b/protocols/krikos-blobs/examples/limit.rs index 73925ea007e..165c6d8c964 100644 --- a/protocols/krikos-blobs/examples/limit.rs +++ b/protocols/krikos-blobs/examples/limit.rs @@ -25,8 +25,8 @@ use krikos::{EndpointAddr, EndpointId, SecretKey, endpoint::presets, protocol::R use krikos_blobs::{ BlobFormat, BlobsProtocol, Hash, provider::events::{ - AbortReason, ConnectMode, EventMask, EventSender, ProviderMessage, RequestMode, - ThrottleMode, + AbortReason, ConnectMode, EventMask, EventSender, ObserveMode, ProviderMessage, + RequestMode, ThrottleMode, }, store::mem::MemStore, ticket::BlobTicket, @@ -114,23 +114,45 @@ fn limit_by_hash(allowed_hashes: HashSet) -> EventSender { // with OK or not OK depending on the hash. We do not want detailed // events once it has been decided to handle a request. get: RequestMode::Intercept, + // Every mode is per request type, so an allowlist has to cover *all* the + // request types that can read the store, not just `get`. Leaving these at + // their `DEFAULT` of "no events" would mean the check below is skipped for + // them and a peer just asks for the blob by another name. + get_many: RequestMode::Intercept, + observe: ObserveMode::Intercept, ..EventMask::DEFAULT }; let (tx, mut rx) = EventSender::channel(32, mask); n0_future::task::spawn(async move { + let allowed = |hash: &Hash| { + if allowed_hashes.contains(hash) { + println!("Request for hash {hash} allowed"); + Ok(()) + } else { + println!("Request for hash {hash} not allowed"); + Err(AbortReason::Permission) + } + }; while let Some(msg) = rx.recv().await { - if let ProviderMessage::GetRequestReceived(msg) = msg { - let res = if !msg.request.ranges.is_blob() { - println!("HashSeq request not allowed"); - Err(AbortReason::Permission) - } else if !allowed_hashes.contains(&msg.request.hash) { - println!("Request for hash {} not allowed", msg.request.hash); - Err(AbortReason::Permission) - } else { - println!("Request for hash {} allowed", msg.request.hash); - Ok(()) - }; - msg.tx.send(res).await.ok(); + match msg { + ProviderMessage::GetRequestReceived(msg) => { + let res = if !msg.request.ranges.is_blob() { + println!("HashSeq request not allowed"); + Err(AbortReason::Permission) + } else { + allowed(&msg.request.hash) + }; + msg.tx.send(res).await.ok(); + } + ProviderMessage::GetManyRequestReceived(msg) => { + let res = msg.request.hashes.iter().try_for_each(&allowed); + msg.tx.send(res).await.ok(); + } + ProviderMessage::ObserveRequestReceived(msg) => { + let res = allowed(&msg.request.hash); + msg.tx.send(res).await.ok(); + } + _ => {} } } }); diff --git a/protocols/krikos-blobs/examples/random_store.rs b/protocols/krikos-blobs/examples/random_store.rs index 1fa131bcfa8..416d5584db3 100644 --- a/protocols/krikos-blobs/examples/random_store.rs +++ b/protocols/krikos-blobs/examples/random_store.rs @@ -7,7 +7,7 @@ use krikos::{EndpointId, SecretKey, endpoint::presets}; use krikos_blobs::{ HashAndFormat, api::downloader::Shuffled, - provider::events::{AbortReason, EventMask, EventSender, ProviderMessage}, + provider::events::{AbortReason, EventMask, EventSender, ProviderMessage, RequestMode}, store::fs::FsStore, test::{add_hash_sequences, create_random_blobs}, }; @@ -101,7 +101,14 @@ pub fn get_or_generate_secret_key() -> Result { } pub fn dump_provider_events(allow_push: bool) -> (tokio::task::JoinHandle<()>, EventSender) { - let (tx, mut rx) = EventSender::channel(100, EventMask::ALL_READONLY); + let mask = EventMask { + // `ALL_READONLY` leaves `push` disabled, which would reject every push before + // the handler below ever sees it and make `--allow-push` a no-op. The handler + // is the thing that decides, so the mask has to hand the request to it. + push: RequestMode::InterceptLog, + ..EventMask::ALL_READONLY + }; + let (tx, mut rx) = EventSender::channel(100, mask); fn dump_updates(mut rx: irpc::channel::mpsc::Receiver) { tokio::spawn(async move { while let Ok(Some(update)) = rx.recv().await { diff --git a/protocols/krikos-blobs/src/provider.rs b/protocols/krikos-blobs/src/provider.rs index d443d5f16ec..e44bc5a333a 100644 --- a/protocols/krikos-blobs/src/provider.rs +++ b/protocols/krikos-blobs/src/provider.rs @@ -148,7 +148,7 @@ impl StreamPair { f: impl FnOnce() -> GetRequest, ) -> Result { self.events - .request(f, self.connection_id, self.reader.id()) + .get_request(f, self.connection_id, self.reader.id()) .await } @@ -157,7 +157,7 @@ impl StreamPair { f: impl FnOnce() -> GetManyRequest, ) -> Result { self.events - .request(f, self.connection_id, self.reader.id()) + .get_many_request(f, self.connection_id, self.reader.id()) .await } @@ -166,7 +166,7 @@ impl StreamPair { f: impl FnOnce() -> PushRequest, ) -> Result { self.events - .request(f, self.connection_id, self.reader.id()) + .push_request(f, self.connection_id, self.reader.id()) .await } @@ -175,7 +175,7 @@ impl StreamPair { f: impl FnOnce() -> ObserveRequest, ) -> Result { self.events - .request(f, self.connection_id, self.reader.id()) + .observe_request(f, self.connection_id, self.reader.id()) .await } diff --git a/protocols/krikos-blobs/src/provider/events.rs b/protocols/krikos-blobs/src/provider/events.rs index b36f7374c00..a5f22b58df9 100644 --- a/protocols/krikos-blobs/src/provider/events.rs +++ b/protocols/krikos-blobs/src/provider/events.rs @@ -35,13 +35,35 @@ pub enum ConnectMode { #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] #[repr(u8)] pub enum ObserveMode { - /// We don't get notification of connect events at all. + /// We don't get observe request events at all. #[default] None, - /// We get a notification for connect events. + /// We get a notification for each observe request. Notify, - /// We get a request for connect events and can reject incoming connections. + /// We get a request for each observe request, and can reject it. Intercept, + /// Observe requests are completely disabled. All of them will be rejected. + /// + /// An observe response streams the local bitfield for a hash, i.e. exactly which + /// byte ranges of it this node holds, so a locked-down provider needs a way to + /// refuse them outright rather than only being able to intercept them. + Disabled, +} + +impl From for RequestMode { + fn from(value: ObserveMode) -> Self { + // The `*Log` variants, not the plain ones: an observe request transfers no + // blobs, so it never produces per-blob transfer events, and the completion + // update is the only update it can ever emit. Mapping to `Notify`/`Intercept` + // would make the request tracker `Disabled`, and a handler waiting on + // `RequestUpdate::Completed` for an observe request would wait forever. + match value { + ObserveMode::None => RequestMode::None, + ObserveMode::Notify => RequestMode::NotifyLog, + ObserveMode::Intercept => RequestMode::InterceptLog, + ObserveMode::Disabled => RequestMode::Disabled, + } + } } /// Request mode for all data related requests. @@ -437,12 +459,65 @@ impl EventSender { Ok(()) } + /// A get request was received. + pub(crate) async fn get_request( + &self, + f: impl FnOnce() -> GetRequest, + connection_id: u64, + request_id: u64, + ) -> Result { + self.request(f, self.mask.get, connection_id, request_id) + .await + } + + /// A get_many request was received. + pub(crate) async fn get_many_request( + &self, + f: impl FnOnce() -> GetManyRequest, + connection_id: u64, + request_id: u64, + ) -> Result { + self.request(f, self.mask.get_many, connection_id, request_id) + .await + } + + /// A push request was received. + /// + /// Note that a push writes to the local store, which is why + /// [`EventMask::DEFAULT`] disables it. + pub(crate) async fn push_request( + &self, + f: impl FnOnce() -> PushRequest, + connection_id: u64, + request_id: u64, + ) -> Result { + self.request(f, self.mask.push, connection_id, request_id) + .await + } + + /// An observe request was received. + pub(crate) async fn observe_request( + &self, + f: impl FnOnce() -> ObserveRequest, + connection_id: u64, + request_id: u64, + ) -> Result { + self.request(f, self.mask.observe.into(), connection_id, request_id) + .await + } + /// Abstract request, to DRY the 3 to 4 request types. /// /// DRYing stuff with lots of bounds is no fun at all... - pub(crate) async fn request( + /// + /// `mode` is the [`RequestMode`] configured for this particular request type. + /// It must be passed in by the caller: reading it off the mask here would + /// have to pick one field, and so would silently apply the wrong policy to + /// the other three request types. + async fn request( &self, f: impl FnOnce() -> Req, + mode: RequestMode, connection_id: u64, request_id: u64, ) -> Result @@ -461,7 +536,7 @@ impl EventSender { { let client = self.inner.as_ref(); Ok(self.create_tracker(( - match self.mask.get { + match mode { RequestMode::None => RequestUpdates::None, RequestMode::Notify if client.is_some() => { let msg = RequestReceived { diff --git a/protocols/krikos-blobs/src/tests.rs b/protocols/krikos-blobs/src/tests.rs index c839f034260..1b99a1322f2 100644 --- a/protocols/krikos-blobs/src/tests.rs +++ b/protocols/krikos-blobs/src/tests.rs @@ -1,4 +1,4 @@ -use std::{collections::HashSet, io, ops::Range, path::PathBuf}; +use std::{collections::HashSet, io, ops::Range, path::PathBuf, time::Duration}; use bao_tree::ChunkRanges; use bytes::Bytes; @@ -15,12 +15,17 @@ use tracing::info; use crate::{ BlobFormat, Hash, HashAndFormat, - api::{Store, blobs::Bitfield}, + api::{ + Store, + blobs::{Bitfield, BlobStatus}, + }, get, hashseq::HashSeq, net_protocol::BlobsProtocol, protocol::{ChunkRangesSeq, GetManyRequest, ObserveRequest, PushRequest}, - provider::events::{AbortReason, EventMask, EventSender, ProviderMessage, RequestUpdate}, + provider::events::{ + AbortReason, EventMask, EventSender, ProviderMessage, RequestMode, RequestUpdate, + }, store::{ fs::{ FsStore, @@ -341,11 +346,21 @@ async fn two_nodes_get_many_mem() -> TestResult<()> { two_nodes_get_many(r1, &store1, r2, &store2).await } +/// [`EventMask::ALL_READONLY`] with push enabled. +/// +/// There is deliberately no such constant in the crate itself — push writes to the local +/// store, so enabling it has to be a conscious act by the operator. These tests are that +/// conscious act: they intercept `PushRequestReceived` and admit the pushing node. +const ALL_WITH_PUSH: EventMask = EventMask { + push: RequestMode::InterceptLog, + ..EventMask::ALL_READONLY +}; + fn event_handler( allowed_nodes: impl IntoIterator, ) -> (EventSender, watch::Receiver, AbortOnDropHandle<()>) { let (count_tx, count_rx) = tokio::sync::watch::channel(0usize); - let (events_tx, mut events_rx) = EventSender::channel(16, EventMask::ALL_READONLY); + let (events_tx, mut events_rx) = EventSender::channel(16, ALL_WITH_PUSH); let allowed_nodes = allowed_nodes.into_iter().collect::>(); let task = AbortOnDropHandle::new(n0_future::task::spawn(async move { while let Some(event) = events_rx.recv().await { @@ -433,6 +448,142 @@ async fn two_nodes_push_blobs_mem() -> TestResult<()> { two_nodes_push_blobs(r1, &store1, r2, &store2, count_rx).await } +/// A push must be refused when the event mask disables it. +/// +/// [`EventMask::DEFAULT`] sets `push: RequestMode::Disabled` precisely because a push +/// writes to the local store, so an unauthorized peer must not be able to place blobs of +/// its choosing into ours. +/// +/// The pusher cannot observe the refusal: `execute_push_sink` stops its receive stream +/// and returns `Stats::default()` unconditionally, and the provider handles each stream +/// in a detached task, so a rejection never surfaces as a connection- or call-level +/// error. The property under test is therefore the one that actually matters — the blob +/// must not land in the receiving store. `allow` is a control: pushing the same blob to +/// a node that permits pushes proves the push path works in this setup, so a passing +/// "denied" assertion cannot be an artifact of a push that never happened. +async fn push_is_rejected_when_disabled( + pusher: Router, + pusher_store: &Store, + deny: Router, + deny_store: &Store, + allow: Router, + allow_store: &Store, + mut allow_count_rx: watch::Receiver, +) -> TestResult<()> { + let size = 1024; + let tt = pusher_store.add_bytes(test_data(size)).await?; + let hash = tt.hash; + let request = PushRequest::new(hash, ChunkRangesSeq::root()); + + // The connections must outlive the pushes: `execute_push_sink` returns once it has + // called `finish()`, but dropping the `Connection` at that point tears down the + // still-in-flight stream and the receiver never finishes importing. + let mut conns = Vec::new(); + for (target, must_succeed) in [(&deny, false), (&allow, true)] { + let conn = pusher + .endpoint() + .connect(target.endpoint().addr(), crate::ALPN) + .await?; + let res = pusher_store + .remote() + .execute_push_sink(conn.clone(), request.clone(), Drain) + .await; + // Whether the refusal reaches the pusher at all is a race, which is exactly why + // the property under test is store contents rather than this result: the + // provider resets the stream, and `execute_push_sink` only sees that if the + // reset lands before it has finished writing. Both outcomes are correct for the + // denying node. The permitting node must succeed, or the control below — the + // thing that proves a passing deny assertion is not vacuous — proves nothing. + if must_succeed { + res?; + } + conns.push(conn); + } + + // The permissive node completing its push bounds the wait for the denying one: + // both were pushed over loopback, the denying one first. + allow_count_rx.changed().await?; + assert_eq!( + allow_store.get_bytes(hash).await?, + test_data(size), + "control: a node that permits pushes must receive the blob" + ); + // Nothing on the denying node is observable from here — a disabled request type + // emits no event, `execute_push_sink` does not wait for the receiver, and the + // provider imports on a detached task. Sampling `has` once could therefore run + // before an accepted push had landed and pass on a broken mask, so keep sampling + // for a while: an accepted push over loopback lands well inside this window. + let deadline = tokio::time::Instant::now() + Duration::from_secs(1); + loop { + // `status`, not `has`: `has` is only true for a *complete* blob, so a rejection + // that arrived after some chunks had already been imported would leave + // attacker-chosen bytes in the store and still satisfy the assertion. + assert_eq!( + deny_store.blobs().status(hash).await?, + BlobStatus::NotFound, + "a push rejected by the event mask must not write to the receiving store" + ); + if tokio::time::Instant::now() >= deadline { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + + tokio::try_join!(pusher.shutdown(), deny.shutdown(), allow.shutdown())?; + Ok(()) +} + +#[tokio::test] +async fn push_is_rejected_when_disabled_mem() -> TestResult<()> { + tracing_subscriber::fmt::try_init().ok(); + let (r1, store1, sp1) = node_test_setup_mem().await?; + // `EventSender::DEFAULT` carries `EventMask::DEFAULT`, i.e. push disabled. + let (r_deny, store_deny, sp_deny) = node_test_setup_mem().await?; + let (events_tx, count_rx, _task) = event_handler([r1.endpoint().id()]); + let (r_allow, store_allow, sp_allow) = node_test_setup_with_events_mem(events_tx).await?; + for sp in [&sp1, &sp_deny, &sp_allow] { + sp.add_endpoint_info(r1.endpoint().addr()); + sp.add_endpoint_info(r_deny.endpoint().addr()); + sp.add_endpoint_info(r_allow.endpoint().addr()); + } + push_is_rejected_when_disabled( + r1, + &store1, + r_deny, + &store_deny, + r_allow, + &store_allow, + count_rx, + ) + .await +} + +#[tokio::test] +async fn push_is_rejected_when_disabled_fs() -> TestResult<()> { + tracing_subscriber::fmt::try_init().ok(); + let testdir = tempfile::tempdir()?; + let (r1, store1, _, sp1) = node_test_setup_fs(testdir.path().join("a")).await?; + let (r_deny, store_deny, _, sp_deny) = node_test_setup_fs(testdir.path().join("deny")).await?; + let (events_tx, count_rx, _task) = event_handler([r1.endpoint().id()]); + let (r_allow, store_allow, _, sp_allow) = + node_test_setup_with_events_fs(testdir.path().join("allow"), events_tx).await?; + for sp in [&sp1, &sp_deny, &sp_allow] { + sp.add_endpoint_info(r1.endpoint().addr()); + sp.add_endpoint_info(r_deny.endpoint().addr()); + sp.add_endpoint_info(r_allow.endpoint().addr()); + } + push_is_rejected_when_disabled( + r1, + &store1, + r_deny, + &store_deny, + r_allow, + &store_allow, + count_rx, + ) + .await +} + pub async fn add_test_hash_seq( blobs: &Store, sizes: impl IntoIterator, diff --git a/protocols/krikos-docs/src/store/fs.rs b/protocols/krikos-docs/src/store/fs.rs index 62325c68878..63e6fd5731f 100644 --- a/protocols/krikos-docs/src/store/fs.rs +++ b/protocols/krikos-docs/src/store/fs.rs @@ -804,22 +804,31 @@ impl<'a> crate::ranger::Store for StoreInstance<'a> { // regular range: iter1 = x <= t < y, iter2 = none Ordering::Less => { // iterator for entries from range.x to range.y + // + // Both endpoints come from the remote peer, so the bounds have to be + // clamped to our namespace; otherwise a range naming another namespace + // would read that document's entries out of the shared records table. let start = Bound::Included(range.x().to_byte_tuple()); let end = Bound::Excluded(range.y().to_byte_tuple()); - let bounds = RecordsBounds::new(start, end); + let bounds = RecordsBounds::new(start, end).clamp_to_namespace(&self.namespace); let iter = RecordsRange::with_bounds(&tables.records, bounds)?; chain_none(iter) } // split range: iter1 = start <= t < y, iter2 = x <= t <= end Ordering::Greater => { // iterator for entries from start to range.y + // + // `from_start`/`to_end` pin only one side to our namespace, so the + // remote-supplied side still needs clamping. let end = Bound::Excluded(range.y().to_byte_tuple()); - let bounds = RecordsBounds::from_start(&self.namespace, end); + let bounds = RecordsBounds::from_start(&self.namespace, end) + .clamp_to_namespace(&self.namespace); let iter = RecordsRange::with_bounds(&tables.records, bounds)?; // iterator for entries from range.x to end let start = Bound::Included(range.x().to_byte_tuple()); - let bounds = RecordsBounds::to_end(&self.namespace, start); + let bounds = RecordsBounds::to_end(&self.namespace, start) + .clamp_to_namespace(&self.namespace); let iter2 = RecordsRange::with_bounds(&tables.records, bounds)?; iter.chain(Some(iter2).into_iter().flatten()) @@ -1049,6 +1058,90 @@ mod tests { Ok(()) } + /// `get_range` must never return entries outside the namespace it is pinned to. + /// + /// All namespaces share one records table, keyed `(namespace, author, key)`, so the + /// query bounds are the only thing keeping documents apart. The range in a + /// `RangeItem` is remote-supplied and unvalidated, and the resulting diff is echoed + /// straight back to the peer — so a range naming a foreign namespace must not widen + /// what the session can see, in any of the three orderings. + #[test] + fn get_range_stays_within_its_namespace() -> Result<()> { + let dbfile = tempfile::NamedTempFile::new()?; + let mut store = Store::persistent(dbfile.path())?; + let author = store.new_author(&mut rand::rng())?; + + let ours = NamespaceSecret::new(&mut rand::rng()); + // One foreign namespace on each side of ours. Namespace ids are random, so a + // single neighbour lands below or above by chance and only covers the clamp + // facing it: a broken start clamp leaks a lower neighbour, a broken end clamp a + // higher one. With one of each, both are caught on every run. + let (mut lower, mut higher) = (None, None); + while lower.is_none() || higher.is_none() { + let candidate = NamespaceSecret::new(&mut rand::rng()); + match candidate.id().cmp(&ours.id()) { + Ordering::Less => drop(lower.get_or_insert(candidate)), + Ordering::Greater => drop(higher.get_or_insert(candidate)), + Ordering::Equal => {} + } + } + let theirs = [ + lower.expect("lower namespace"), + higher.expect("higher namespace"), + ]; + store.new_replica(ours.clone())?; + for ns in &theirs { + store.new_replica(ns.clone())?; + } + for ns in [&ours, &theirs[0], &theirs[1]] { + let mut wrapper = StoreInstance::new(ns.id(), &mut store); + let id = RecordIdentifier::new(ns.id(), author.id(), b"key"); + let entry = Entry::new(id, Record::current_from_data(b"value")); + wrapper.entry_put(SignedEntry::from_entry(entry, ns, &author))?; + } + + // Endpoints spanning the whole table, i.e. naming neither namespace in + // particular. `min < max`, so swapping them walks each of the two branches that + // build bounds out of remote input. + let min = RecordIdentifier::new(NamespaceId::from(&[0u8; 32]), author.id(), b""); + let max = RecordIdentifier::new(NamespaceId::from(&[255u8; 32]), author.id(), b""); + + let mut wrapper = StoreInstance::new(ours.id(), &mut store); + // How many of our own entries each range must still return. Clamping that + // over-shoots — an inverted comparison, a fallback arm substituting the wrong + // bound — yields an empty range, which leaks nothing and would satisfy the + // negative check below on its own. `get_fingerprint` delegates to `get_range`, + // so an always-empty range silently stops sync from ever converging. + for (label, range, ours_expected) in [ + // Spans the whole table; clamped, that is all of our namespace. + ("less", Range::new(min.clone(), max.clone()), 1), + // Wrapping range: everything *outside* `[min, max)`, which within our + // namespace is nothing. + ("greater", Range::new(max.clone(), min.clone()), 0), + // Identity range: the whole replica, no remote bounds involved. + ("equal", Range::new(min.clone(), min.clone()), 1), + ] { + let namespaces = wrapper + .get_range(range)? + .map(|entry| entry.map(|e| e.namespace())) + .collect::>>()?; + let leaked = namespaces + .iter() + .filter(|ns| **ns != ours.id()) + .collect::>(); + assert!( + leaked.is_empty(), + "{label} range leaked entries from another namespace: {leaked:?}" + ); + assert_eq!( + namespaces.len(), + ours_expected, + "{label} range returned the wrong number of entries from our own namespace" + ); + } + Ok(()) + } + #[test] fn test_basics() -> Result<()> { let dbfile = tempfile::NamedTempFile::new()?; diff --git a/protocols/krikos-docs/src/store/fs/bounds.rs b/protocols/krikos-docs/src/store/fs/bounds.rs index 3565a605e3a..5e75ab2fe45 100644 --- a/protocols/krikos-docs/src/store/fs/bounds.rs +++ b/protocols/krikos-docs/src/store/fs/bounds.rs @@ -61,6 +61,60 @@ impl RecordsBounds { Self::new(start, Self::namespace_end(ns)) } + /// Intersect these bounds with `ns`. + /// + /// Every namespace shares the same records table, so the bounds are the only thing + /// keeping documents apart. Sync range endpoints arrive from the remote peer + /// unvalidated and may name any namespace, so they have to be intersected with the + /// namespace the session is pinned to: a range reaching into another document then + /// selects nothing instead of reading it. + pub fn clamp_to_namespace(self, ns: &NamespaceId) -> Self { + let Self(start, end) = self; + // `namespace_start` is always `Included`; the tighter of two lower bounds is the + // greater one, and at equal keys `Excluded` is the tighter shape. Only an + // `Unbounded` caller start has nothing to keep, and there the namespace bound is + // the answer. + let start = match (start, Self::namespace_start(ns)) { + (Bound::Included(remote), Bound::Included(ns_start)) => { + Bound::Included(remote.max(ns_start)) + } + (Bound::Excluded(remote), Bound::Included(ns_start)) if remote >= ns_start => { + Bound::Excluded(remote) + } + (_, ns_start) => ns_start, + }; + // `namespace_end` is `Excluded`, or `Unbounded` for the last namespace, in which + // case the caller's own end is already within it. An `Included` caller end below + // the namespace end is the tighter bound and has to be kept: replacing it would + // return the whole namespace for what was asked as a single key. + let end = match (end, Self::namespace_end(ns)) { + (Bound::Excluded(remote), Bound::Excluded(ns_end)) => { + Bound::Excluded(remote.min(ns_end)) + } + (Bound::Included(remote), Bound::Excluded(ns_end)) if remote < ns_end => { + Bound::Included(remote) + } + (Bound::Excluded(remote), Bound::Unbounded) => Bound::Excluded(remote), + (Bound::Included(remote), Bound::Unbounded) => Bound::Included(remote), + (_, ns_end) => ns_end, + }; + // The intersection is empty whenever the range covers only foreign namespaces. + // Normalize that to a range selecting nothing, as inverted bounds are not a + // valid query. + let is_empty = match (&start, &end) { + (Bound::Included(s), Bound::Excluded(e)) + | (Bound::Excluded(s), Bound::Excluded(e)) + | (Bound::Excluded(s), Bound::Included(e)) => s >= e, + (Bound::Included(s), Bound::Included(e)) => s > e, + _ => false, + }; + if is_empty { + let empty = (ns.to_bytes(), [0u8; 32], Bytes::new()); + return Self(Bound::Included(empty.clone()), Bound::Excluded(empty)); + } + Self(start, end) + } + pub fn as_ref(&self) -> (Bound>, Bound>) { fn map(id: &RecordsIdOwned) -> RecordsId<'_> { (&id.0, &id.1, &id.2[..]) diff --git a/protocols/krikos-docs/tests/sync.rs b/protocols/krikos-docs/tests/sync.rs index 35ab8989ff5..15526afbf05 100644 --- a/protocols/krikos-docs/tests/sync.rs +++ b/protocols/krikos-docs/tests/sync.rs @@ -307,7 +307,7 @@ async fn sync_full_basic() -> testresult::TestResult<()> { .await; info!("peer0: wait for 2 events (join & accept sync finished from peer1)"); - assert_next( + assert_next_ignoring( &mut events0, TIMEOUT, vec![ @@ -315,6 +315,7 @@ async fn sync_full_basic() -> testresult::TestResult<()> { Box::new(move |e| match_sync_finished(e, peer1)), match_event!(LiveEvent::PendingContentReady), ], + redundant_live_events(peer1), ) .await; @@ -326,10 +327,11 @@ async fn sync_full_basic() -> testresult::TestResult<()> { .await?; assert_latest(blobs1, &doc1, key1, value1).await; info!("peer1: wait for 1 event (local insert, and pendingcontentready)"); - assert_next( + assert_next_ignoring( &mut events1, TIMEOUT, vec![match_event!(LiveEvent::InsertLocal { entry} if entry.content_hash() == hash1)], + redundant_live_events(peer0), ) .await; @@ -342,7 +344,7 @@ async fn sync_full_basic() -> testresult::TestResult<()> { // Missing/Incomplete split is pure timing. Match both. // peer0: assert events for entry received via gossip info!("peer0: wait for 2 events (gossip'ed entry from peer1)"); - assert_next( + assert_next_ignoring( &mut events0, TIMEOUT, vec![ @@ -351,6 +353,7 @@ async fn sync_full_basic() -> testresult::TestResult<()> { ), Box::new(move |e| matches!(e, LiveEvent::ContentReady { hash } if *hash == hash1)), ], + redundant_live_events(peer1), ).await; assert_latest(blobs0, &doc0, key1, value1).await; @@ -408,7 +411,7 @@ async fn sync_full_basic() -> testresult::TestResult<()> { assert_latest(blobs2, &doc2, b"k2", b"v2").await; info!("peer0: wait for 2 events (join & accept sync finished from peer2)"); - assert_next( + assert_next_ignoring( &mut events0, TIMEOUT, vec![ @@ -416,11 +419,12 @@ async fn sync_full_basic() -> testresult::TestResult<()> { Box::new(move |e| match_sync_finished(e, peer2)), match_event!(LiveEvent::PendingContentReady), ], + redundant_live_events(peer2), ) .await; info!("peer1: wait for 2 events (join & accept sync finished from peer2)"); - assert_next( + assert_next_ignoring( &mut events1, TIMEOUT, vec![ @@ -428,6 +432,7 @@ async fn sync_full_basic() -> testresult::TestResult<()> { Box::new(move |e| match_sync_finished(e, peer2)), match_event!(LiveEvent::PendingContentReady), ], + redundant_live_events(peer2), ) .await; @@ -553,7 +558,11 @@ async fn test_sync_via_relay() -> Result<()> { assert_next_unordered_with_optionals( &mut events, - Duration::from_secs(2), + // This is the join, over a relay: the peers still have to complete their relay + // handshake before any of these events can happen. Two seconds was under that on + // a loaded runner, and the failure looked like a missing event rather than a + // slow one. Use the same budget as every other wait in this file. + TIMEOUT, vec![ Box::new(move |e| matches!(e, LiveEvent::NeighborUp(n) if *n== node1_id)), Box::new(move |e| match_sync_finished(e, node1_id)), @@ -565,7 +574,12 @@ async fn test_sync_via_relay() -> Result<()> { ), match_event!(LiveEvent::PendingContentReady), ], - vec![Box::new(move |e| match_sync_finished(e, node1_id))], + vec![ + Box::new(move |e| match_sync_finished(e, node1_id)), + // A relayed connection can drop mid-sync; the engine retries, and the + // required matcher above still demands a successful sync and the entry. + match_event!(LiveEvent::SyncFinished(e) if e.peer == node1_id && e.result.is_err()), + ], ) .await; let actual = blobs2 @@ -596,6 +610,8 @@ async fn test_sync_via_relay() -> Result<()> { vec![ Box::new(move |e| match_sync_finished(e, node1_id)), Box::new(move |e| matches!(e, LiveEvent::PendingContentReady)), + // As above: the relayed connection may drop and the sync be retried. + match_event!(LiveEvent::SyncFinished(e) if e.peer == node1_id && e.result.is_err()), ], ) .await; @@ -712,6 +728,12 @@ async fn sync_restart_node() -> Result<()> { vec![ match_event!(LiveEvent::SyncFinished(e) if e.peer == id2 && e.result.is_ok()), match_event!(LiveEvent::PendingContentReady), + // node1 has just restarted, so the first attempt to reach node2 can still + // land on the connection node2 held to the old process and fail + // ("Failed to close connection1"). Sync retries, and the required matcher + // above still demands a successful one plus the entry itself, so tolerating + // the failed attempt does not weaken what this asserts. + match_event!(LiveEvent::SyncFinished(e) if e.peer == id2 && e.result.is_err()), ] ).await; assert_latest(blobs1, &doc1, b"n2/b", b"b").await; @@ -731,6 +753,8 @@ async fn sync_restart_node() -> Result<()> { match_event!(LiveEvent::PendingContentReady), match_event!(LiveEvent::SyncFinished(e) if e.peer == id2 && e.result.is_ok()), match_event!(LiveEvent::PendingContentReady), + // As above: a retried sync attempt against the restarted node may fail once. + match_event!(LiveEvent::SyncFinished(e) if e.peer == id2 && e.result.is_err()), ] ).await; @@ -855,8 +879,23 @@ async fn test_download_policies() -> Result<()> { let mut synced_a = 0usize; let mut synced_b = 0usize; loop { + // Bind the whole `Result>`, not just the `Ok(Some(..))` shape. A + // refutable pattern that fails to match disables that branch for the rest of + // the `select!`, so a stream that ended or errored used to leave the loop + // waiting on the other one until the 120s timeout, reported only as + // "timeout elapsed" with nothing about how far it got. Fail where it breaks + // instead, and say what had been seen. tokio::select! { - Ok(Some(ev)) = events_a.try_next() => { + ev = events_a.try_next() => { + let ev = match ev { + Ok(Some(ev)) => ev, + Ok(None) => bail!( + "node a's event stream ended early \ + (synced_a={synced_a}, downloaded_a={downloaded_a:?}, \ + synced_b={synced_b}, downloaded_b={downloaded_b:?})" + ), + Err(err) => return Err(err).context("node a's event stream errored"), + }; match ev { InsertRemote { content_status, entry, .. } => { synced_a += 1; @@ -870,7 +909,16 @@ async fn test_download_policies() -> Result<()> { _ => {} } } - Ok(Some(ev)) = events_b.try_next() => { + ev = events_b.try_next() => { + let ev = match ev { + Ok(Some(ev)) => ev, + Ok(None) => bail!( + "node b's event stream ended early \ + (synced_a={synced_a}, downloaded_a={downloaded_a:?}, \ + synced_b={synced_b}, downloaded_b={downloaded_b:?})" + ), + Err(err) => return Err(err).context("node b's event stream errored"), + }; match ev { InsertRemote { content_status, entry, .. } => { synced_b += 1; @@ -899,12 +947,12 @@ async fn test_download_policies() -> Result<()> { break; } } - (downloaded_a, downloaded_b) + Ok((downloaded_a, downloaded_b)) }; let (downloaded_a, mut downloaded_b) = n0_future::time::timeout(TIMEOUT, fut) .await - .context("timeout elapsed")?; + .context("timeout elapsed")??; downloaded_b.sort(); assert_eq!(downloaded_a, vec!["lotr/fellowship_of_the_ring"]); @@ -1321,28 +1369,50 @@ fn apply_matchers(item: &T, matchers: &mut Vec bool + Send> false } -/// Receive the next `matchers.len()` elements from a stream and matches them against the functions -/// in `matchers`, in order. +/// Receive elements from a stream and match them against `matchers`, in order, skipping +/// any event that matches one of `ignored` without consuming a matcher. /// -/// Returns all received events. +/// A live document emits events these ordered assertions do not care about and cannot +/// keep from interleaving. The two seen in practice are a second `SyncFinished` — when +/// the ticket-driven sync and the `NeighborUp`-driven sync both run, as the optional +/// matchers in `sync_full_basic` already document — and a `PendingContentReady` +/// whenever a download queue drains. Either one landing between two asserted events +/// used to fail the whole test. +/// +/// Skipping exactly the tolerated events keeps the order of the events under test +/// asserted, which is what dropping to [`assert_next_unordered_with_optionals`] would +/// give up. `matchers` takes precedence: an event that satisfies the next expected +/// matcher is consumed by it even if it would also be ignorable, so a required +/// `SyncFinished` in a sequence is still matched rather than skipped. +/// +/// Returns the events that matched `matchers`. #[allow(clippy::type_complexity)] -async fn assert_next( +async fn assert_next_ignoring( mut stream: impl Stream> + Unpin + Send, timeout: Duration, matchers: Vec bool + Send>>, + ignored: Vec bool + Send>>, ) -> Vec { let fut = async { let mut items = vec![]; + let mut skipped = vec![]; for (i, f) in matchers.iter().enumerate() { - let item = stream - .try_next() - .await - .expect("event stream ended prematurely") - .expect("event stream errored"); - if !(f)(&item) { - panic!("assertion failed for event {i} {item:?}"); + loop { + let item = stream + .try_next() + .await + .expect("event stream ended prematurely") + .expect("event stream errored"); + if (f)(&item) { + items.push(item); + break; + } + if ignored.iter().any(|g| g(&item)) { + skipped.push(item); + continue; + } + panic!("assertion failed for event {i} {item:?} (skipped: {skipped:?})"); } - items.push(item); } items }; @@ -1430,6 +1500,25 @@ async fn assert_next_unordered_with_optionals( events } +/// The live events that may interleave with any asserted sequence involving `peer`. +/// +/// A `SyncFinished` for `peer` because sync can legitimately run more than once — the +/// ticket-driven run and the `NeighborUp`-driven run race, and either may also be +/// retried after a transient connection failure, so the outcome is not constrained +/// either. A `PendingContentReady` because it is emitted whenever a download queue +/// drains, which is not tied to the entry a given assertion is waiting for. +/// +/// Only for use as the `ignored` set of [`assert_next_ignoring`], where the expected +/// matchers take precedence: this makes the events tolerated between the asserted ones, +/// not accepted in their place. +#[allow(clippy::type_complexity)] +fn redundant_live_events(peer: PublicKey) -> Vec bool + Send>> { + vec![ + Box::new(move |e| matches!(e, LiveEvent::SyncFinished(ev) if ev.peer == peer)), + match_event!(LiveEvent::PendingContentReady), + ] +} + /// Asserts that the event is a [`LiveEvent::SyncFinished`] and that the contained [`SyncEvent`] /// has no error and matches `peer` and `namespace`. fn match_sync_finished(event: &LiveEvent, peer: PublicKey) -> bool { diff --git a/scripts/determinism-boundaries.semantic.txt b/scripts/determinism-boundaries.semantic.txt index e868eb20887..6a8f16b8556 100644 --- a/scripts/determinism-boundaries.semantic.txt +++ b/scripts/determinism-boundaries.semantic.txt @@ -541,6 +541,7 @@ spawn-task krikos-relay/src/quic.rs tests::test_qad_client_closes_unresponsive_f spawn-task krikos-relay/src/quic.rs tests::test_qad_connect_delayed tokio::spawn 1 spawn-task krikos-relay/src/server/client.rs tests::client_actor_survives_production_keepalive_horizon_when_polled tokio::task::spawn 1 spawn-task krikos-relay/src/server/client.rs tests::test_client_actor_basic tokio::task::spawn 1 +spawn-task krikos-relay/src/server/client.rs tests::undeliverable_packet_does_not_end_the_session tokio::task::spawn 1 spawn-task krikos-relay/src/server/http_server/listener.rs impl::spawn tokio::task::spawn 1 spawn-task krikos-relay/src/server/http_server/tests.rs test_server_basic tokio::spawn 1 spawn-task krikos-relay/src/server/http_server/tests.rs test_server_basic tokio::spawn 2 diff --git a/scripts/determinism-boundaries.txt b/scripts/determinism-boundaries.txt index d65b7dd3a34..8bad716dd8e 100644 --- a/scripts/determinism-boundaries.txt +++ b/scripts/determinism-boundaries.txt @@ -19,8 +19,8 @@ clock-timer krikos-dns-server/src/server.rs:161 let dns_result = tokio::time::ti clock-timer krikos-dns-server/src/server.rs:165 let http_result = tokio::time::timeout_at(deadline, self.http_server.shutdown()) clock-timer krikos-dns-server/src/server.rs:169 let store_result = tokio::time::timeout_at(deadline, self.store.shutdown()) clock-timer krikos-dns-server/src/server.rs:174 tokio::time::timeout_at(deadline, server.shutdown()) -clock-timer krikos-dns-server/src/store.rs:393 tokio::time::sleep(Duration::from_secs(1)).await; -clock-timer krikos-dns-server/src/store.rs:400 tokio::time::sleep(Duration::from_secs(1)).await; +clock-timer krikos-dns-server/src/store.rs:416 tokio::time::sleep(Duration::from_secs(1)).await; +clock-timer krikos-dns-server/src/store.rs:423 tokio::time::sleep(Duration::from_secs(1)).await; clock-timer krikos-dns-server/src/store/signed_packets.rs:1007 tokio::time::timeout(SHUTDOWN_DEADLINE, store.shutdown()) clock-timer krikos-dns-server/src/store/signed_packets.rs:268 let timeout = tokio::time::sleep(self.options.max_batch_time.get()); clock-timer krikos-dns-server/src/store/signed_packets.rs:361 Timestamp::now().as_micros().saturating_sub(expiry_us), @@ -30,10 +30,10 @@ clock-timer krikos-dns-server/src/store/signed_packets.rs:683 tokio::time::sleep clock-timer krikos-dns-server/src/store/signed_packets.rs:961 let drop_started = Instant::now(); clock-timer krikos-dns-server/src/store/signed_packets.rs:985 tokio::time::timeout(SHUTDOWN_DEADLINE, store.shutdown()) clock-timer krikos-dns-server/tests/publish_resolve.rs:77 std::time::SystemTime::now() -clock-timer krikos-dns/src/pkarr.rs:328 /// [`Timestamp::now`] is guaranteed to be strictly monotonic: it will never -clock-timer krikos-dns/src/pkarr.rs:336 /// Tracks the last timestamp returned by [`Timestamp::now`] to ensure monotonicity. -clock-timer krikos-dns/src/pkarr.rs:349 use n0_future::time::SystemTime; -clock-timer krikos-dns/src/pkarr.rs:350 let micros = SystemTime::now() +clock-timer krikos-dns/src/pkarr.rs:355 /// [`Timestamp::now`] is guaranteed to be strictly monotonic: it will never +clock-timer krikos-dns/src/pkarr.rs:363 /// Tracks the last timestamp returned by [`Timestamp::now`] to ensure monotonicity. +clock-timer krikos-dns/src/pkarr.rs:376 use n0_future::time::SystemTime; +clock-timer krikos-dns/src/pkarr.rs:377 let micros = SystemTime::now() clock-timer krikos-dns/src/pkarr.rs:81 let timestamp = Timestamp::now(); clock-timer krikos-relay/src/defaults.rs:28 use n0_future::time::Duration; clock-timer krikos-relay/src/ping_tracker.rs:5 use n0_future::time::Instant; @@ -49,12 +49,12 @@ clock-timer krikos-relay/src/quic.rs:678 let start = Instant::now(); clock-timer krikos-relay/src/server/admission.rs:107 last_refill: Instant::now(), clock-timer krikos-relay/src/server/admission.rs:112 let now = Instant::now(); clock-timer krikos-relay/src/server/admission.rs:150 tokio::time::advance(std::time::Duration::from_millis(500)).await; -clock-timer krikos-relay/src/server/client.rs:1007 let res = tokio::time::timeout(Duration::from_millis(100), stream.next()).await; -clock-timer krikos-relay/src/server/client.rs:1013 tokio::time::sleep(Duration::from_secs(1)).await; -clock-timer krikos-relay/src/server/client.rs:1016 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) -clock-timer krikos-relay/src/server/client.rs:816 tokio::time::advance(Duration::from_secs(21)).await; -clock-timer krikos-relay/src/server/client.rs:823 tokio::time::advance(Duration::from_secs(6)).await; -clock-timer krikos-relay/src/server/client.rs:996 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) +clock-timer krikos-relay/src/server/client.rs:1097 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) +clock-timer krikos-relay/src/server/client.rs:1108 let res = tokio::time::timeout(Duration::from_millis(100), stream.next()).await; +clock-timer krikos-relay/src/server/client.rs:1114 tokio::time::sleep(Duration::from_secs(1)).await; +clock-timer krikos-relay/src/server/client.rs:1117 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) +clock-timer krikos-relay/src/server/client.rs:917 tokio::time::advance(Duration::from_secs(21)).await; +clock-timer krikos-relay/src/server/client.rs:924 tokio::time::advance(Duration::from_secs(6)).await; clock-timer krikos-relay/src/server/clients.rs:619 tokio::time::timeout(Duration::from_secs(1), async move { clock-timer krikos-relay/src/server/clients.rs:624 tokio::time::sleep(Duration::from_millis(100)).await; clock-timer krikos-relay/src/server/clients.rs:696 tokio::time::timeout(Duration::from_secs(1), { @@ -841,8 +841,9 @@ spawn-task krikos-relay/src/quic.rs:30 use tokio::{sync::Semaphore, task::JoinSe spawn-task krikos-relay/src/quic.rs:619 let task = AbortOnDropHandle::new(tokio::spawn({ spawn-task krikos-relay/src/quic.rs:679 let server_task = tokio::spawn( spawn-task krikos-relay/src/server.rs:51 task::{JoinError, JoinSet}, -spawn-task krikos-relay/src/server/client.rs:719 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); -spawn-task krikos-relay/src/server/client.rs:814 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); +spawn-task krikos-relay/src/server/client.rs:739 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); +spawn-task krikos-relay/src/server/client.rs:841 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); +spawn-task krikos-relay/src/server/client.rs:915 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); spawn-task krikos-relay/src/server/http_server/listener.rs:323 let task = tokio::task::spawn( spawn-task krikos-relay/src/server/http_server/listener.rs:327 let mut set = tokio::task::JoinSet::new(); spawn-task krikos-relay/src/server/http_server/tests.rs:305 let handler_task = tokio::spawn(async move { @@ -1002,8 +1003,8 @@ unordered-collection krikos-dns/src/endpoint_info.rs:45 collections::{BTreeSet, unordered-collection krikos-dns/src/endpoint_info.rs:78 fn dedup(items: &mut Vec) -> HashSet { unordered-collection krikos-dns/src/endpoint_info.rs:80 let mut seen = HashSet::new(); unordered-collection krikos-relay/src/server/client.rs:4 collections::HashSet, -unordered-collection krikos-relay/src/server/client.rs:600 clients: HashSet, -unordered-collection krikos-relay/src/server/client.rs:609 clients: HashSet::new(), +unordered-collection krikos-relay/src/server/client.rs:620 clients: HashSet, +unordered-collection krikos-relay/src/server/client.rs:629 clients: HashSet::new(), unordered-collection krikos-relay/src/server/clients.rs:12 use dashmap::DashMap; unordered-collection krikos-relay/src/server/clients.rs:180 clients: DashMap::new(), unordered-collection krikos-relay/src/server/clients.rs:181 sent_to: DashMap::new(),