From 8289223d1fa13ab9eda69a58fe47be81ab061065 Mon Sep 17 00:00:00 2001 From: jonaswre Date: Sun, 2 Aug 2026 13:34:59 +0000 Subject: [PATCH 01/11] fix(blobs): enforce the per-request-type event mask, not always mask.get EventSender::request hardcoded `match self.mask.get` for all four request types. `mask.push`, `mask.get_many` and `mask.observe` were never read anywhere in the crate -- the only mask reads were `connected`, `get` and `throttle`. This is a security fix: push writes to the local store, and the mask is the documented control that gates it. The contract says so in three places: EventMask docs ("push requests are disabled by default, as they can write to the local store"), DEFAULT and ALL_READONLY both set `push: RequestMode::Disabled`, and ALL_READONLY explicitly declines to ship a push-enabled constant because it "would risk misuse". None of it took effect. `RequestMode::None` -- what BlobsProtocol::new(&store, None) installs -- means "allow, don't notify", so the Disabled arm returning ProgressError::Permission was unreachable for push under every configuration in the tree, including the public ALPN registration in framework/app. An unauthenticated peer that knows an EndpointId could therefore write blobs of its choosing into the store, which the node then serves to anyone presenting the hash. Content is BLAKE3-verified against the pusher's own hash, so existing data cannot be forged or corrupted; the impact is unauthorized write and content injection under the operator's identity. Thread the applicable RequestMode into `request` as a parameter, supplied by four typed wrappers next to the mask itself. Observe maps through a new From for RequestMode; it has no Disabled variant, so its behaviour is unchanged beyond reading the right field. The existing push tests were passing *because* of the bug: event_handler used ALL_READONLY, which disables push. They now use an explicitly push-enabled mask -- the conscious act the ALL_READONLY doc asks for. New tests push the same blob to a denying node and to a permitting one. The pusher cannot observe a refusal (execute_push_sink stops its receive stream and returns Stats::default(), and the provider handles streams in detached tasks), so the assertion is that the blob does not land. The permitting node is the control: it proves the push path works in that setup, so the deny assertion cannot pass vacuously. Both fail without the fix, for the right reason, and pass with it. krikos-blobs 108/108. Inherited from upstream iroh-blobs v0.103.0. Co-Authored-By: Claude Opus 5 --- protocols/krikos-blobs/src/provider.rs | 8 +- protocols/krikos-blobs/src/provider/events.rs | 67 ++++++++- protocols/krikos-blobs/src/tests.rs | 127 +++++++++++++++++- 3 files changed, 194 insertions(+), 8 deletions(-) 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..672b250c681 100644 --- a/protocols/krikos-blobs/src/provider/events.rs +++ b/protocols/krikos-blobs/src/provider/events.rs @@ -44,6 +44,16 @@ pub enum ObserveMode { Intercept, } +impl From for RequestMode { + fn from(value: ObserveMode) -> Self { + match value { + ObserveMode::None => RequestMode::None, + ObserveMode::Notify => RequestMode::Notify, + ObserveMode::Intercept => RequestMode::Intercept, + } + } +} + /// Request mode for all data related requests. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] #[repr(u8)] @@ -437,12 +447,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 +524,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..4108a77dc3f 100644 --- a/protocols/krikos-blobs/src/tests.rs +++ b/protocols/krikos-blobs/src/tests.rs @@ -20,7 +20,9 @@ use crate::{ 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 +343,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 +445,117 @@ 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 in [&deny, &allow] { + let conn = pusher + .endpoint() + .connect(target.endpoint().addr(), crate::ALPN) + .await?; + pusher_store + .remote() + .execute_push_sink(conn.clone(), request.clone(), Drain) + .await?; + 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" + ); + assert!( + !deny_store.blobs().has(hash).await?, + "a push rejected by the event mask must not write to the receiving store" + ); + + 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, From c401cf93bd8ea12458570963411d92dcf5034269 Mon Sep 17 00:00:00 2001 From: jonaswre Date: Sun, 2 Aug 2026 13:35:13 +0000 Subject: [PATCH 02/11] fix(docs): clamp remote sync ranges to the session namespace get_range built its redb bounds straight from the range endpoints in an incoming RangeItem. Every namespace shares one records table keyed (namespace, author, key), so those bounds are the only thing keeping documents apart, and the endpoints arrive off the wire unvalidated: RecordIdentifier is a raw Bytes with a derived Deserialize, and validate_limits counts parts and entries without ever inspecting `range`. The Ordering::Equal branch scoped correctly to the session namespace. The other two did not. Less used both remote endpoints verbatim; Greater used from_start/to_end, which pin only *one* side of each sub-range and leave the other remote. RecordsBounds::new is a passthrough and RecordsRange::with_bounds applies no post-filter, so nothing downstream caught it. get_fingerprint delegates to get_range and inherited the flaw. This is a security fix. The resulting diff is echoed back to the peer (ranger.rs, `have_local: false` -> diff -> RangeItem), and namespace and signature validation apply only to *incoming* entries, never to the outgoing diff. Since accept_request admits any authenticated peer for any namespace in our sync set, a peer holding one valid ticket could read every other document on the node: namespace ids, author keys, record keys, timestamps, content hashes and signatures. A leaked NamespaceId is itself the read capability, and leaked content hashes are fetchable from the blobs store, so it chains past metadata to content. Add RecordsBounds::clamp_to_namespace and apply it to every branch that consumes remote input, including both sub-ranges of Greater. An empty intersection -- what a range naming only foreign namespaces clamps to -- normalizes to a range selecting nothing, since inverted bounds are not a valid query. Normal traffic never exercised the gap: sessions start at Range::new(x,x) (the Equal branch) and recursion only produces in-namespace split points. Reconciliation is unaffected -- sync_big, sync_full_basic, sync_gossip_bulk and sync_restart_node all pass, krikos-docs 81 lib + 13 integration green. New test drives all three orderings with endpoints spanning the whole table and asserts nothing outside the session namespace comes back. It fails on the Less branch without the fix. Inherited from upstream iroh-docs v0.101.0. Co-Authored-By: Claude Opus 5 --- protocols/krikos-docs/src/store/fs.rs | 66 +++++++++++++++++++- protocols/krikos-docs/src/store/fs/bounds.rs | 39 ++++++++++++ 2 files changed, 102 insertions(+), 3 deletions(-) diff --git a/protocols/krikos-docs/src/store/fs.rs b/protocols/krikos-docs/src/store/fs.rs index 62325c68878..cb5e2cde198 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,57 @@ 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()); + let theirs = NamespaceSecret::new(&mut rand::rng()); + store.new_replica(ours.clone())?; + store.new_replica(theirs.clone())?; + for ns in [&ours, &theirs] { + 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); + for (label, range) in [ + ("less", Range::new(min.clone(), max.clone())), + ("greater", Range::new(max.clone(), min.clone())), + ("equal", Range::new(min.clone(), min.clone())), + ] { + let leaked = wrapper + .get_range(range)? + .map(|entry| entry.map(|e| e.namespace())) + .collect::>>()? + .into_iter() + .filter(|ns| *ns != ours.id()) + .collect::>(); + assert!( + leaked.is_empty(), + "{label} range leaked entries from another namespace: {leaked:?}" + ); + } + 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..f06ff04f18e 100644 --- a/protocols/krikos-docs/src/store/fs/bounds.rs +++ b/protocols/krikos-docs/src/store/fs/bounds.rs @@ -61,6 +61,45 @@ 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; + // Both `namespace_start` and every caller's start are `Included`; the tighter of + // two lower bounds is the greater one. Unexpected shapes fall back to the + // namespace bound, which is never wider than what was asked for. + let start = match (start, Self::namespace_start(ns)) { + (Bound::Included(remote), Bound::Included(ns_start)) => { + Bound::Included(remote.max(ns_start)) + } + (_, ns_start) => ns_start, + }; + // `namespace_end` is `Excluded`, or `Unbounded` for the last namespace, in which + // case the remote's own end is already within it. + let end = match (end, Self::namespace_end(ns)) { + (Bound::Excluded(remote), Bound::Excluded(ns_end)) => { + Bound::Excluded(remote.min(ns_end)) + } + (Bound::Excluded(remote), Bound::Unbounded) => Bound::Excluded(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. + if let (Bound::Included(s), Bound::Excluded(e)) = (&start, &end) + && s >= e + { + 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[..]) From 9dcb8cb13a45e5ce32667adbf759e31ebabcf97a Mon Sep 17 00:00:00 2001 From: jonaswre Date: Sun, 2 Aug 2026 13:35:28 +0000 Subject: [PATCH 03/11] fix(dns-server): verify DHT packets against the key that was queried The mainline fallback in ZoneStore::resolve passed whatever the DHT returned to mutable_item_to_signed_packet, which assembled it with SignedPacket::from_parts_unchecked -- no signature verification -- and never compared item.key() to the pubkey being resolved. That check cannot be delegated to mainline. MutableItem::from_dht_message verifies the signature against the key carried in the *response* and then stores query.target() verbatim; target_from_key is only ever used to build a request. The contrast is a few lines up in the same match, where immutable items *are* validated against the target. So mainline guarantees "signed by somebody", while we were reading it as "signed by the key I asked for". This is a security fix. Any DHT node answering the lookup could reply with a packet correctly signed under a key it controls and a high seq (get_mutable_most_recent folds on seq, so it wins deterministically). The zone cache then strips the attacker's z32 root and node_zone_handler re-appends the queried one, so the records are served authoritatively under the victim's name. Clients cannot detect it -- from_txt_lookup derives the EndpointId from the queried name and plain DNS carries no signature to re-check. QUIC/TLS still prevents impersonating the node identity, but the attacker forges user-data attributed to that EndpointId and serves arbitrary records under its public-key domain, which is precisely the binding pkarr exists to provide. Add SignedPacket::from_parts, which verifies rather than trusting, and have mutable_item_to_signed_packet take the queried key and build from it. A packet signed by any other key now fails signature verification instead of being silently re-attributed. Exposure was limited: mainline defaults to None and config.prod.toml disables it, but config.dev.toml enables it and lib.rs documents it as a supported production mode. New test accepts an item under the key that signed it and refuses the same item under another; it fails without the fix. krikos-dns-server 38/38. Co-Authored-By: Claude Opus 5 --- krikos-dns-server/src/store.rs | 62 ++++++++++++++++++++++++++++++---- krikos-dns/src/pkarr.rs | 19 +++++++++++ 2 files changed, 74 insertions(+), 7 deletions(-) diff --git a/krikos-dns-server/src/store.rs b/krikos-dns-server/src/store.rs index fbb76f39f73..b99b23461b1 100644 --- a/krikos-dns-server/src/store.rs +++ b/krikos-dns-server/src/store.rs @@ -164,7 +164,7 @@ impl ZoneStore { .get_mutable_most_recent(pubkey.as_bytes(), None) .await; if let Some(item) = maybe_item - && let Ok(packet) = mutable_item_to_signed_packet(&item) + && let Ok(packet) = mutable_item_to_signed_packet(&item, pubkey) { debug!("DHT resolve successful {:?}", packet); return self @@ -215,17 +215,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 +414,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 +464,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..f30cc013013 100644 --- a/krikos-dns/src/pkarr.rs +++ b/krikos-dns/src/pkarr.rs @@ -130,6 +130,25 @@ 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. + pub fn from_parts( + public_key: &[u8; 32], + signature: &[u8], + 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. From 20da6678c252827977edac7ef09f390346ea3c00 Mon Sep 17 00:00:00 2001 From: jonaswre Date: Sun, 2 Aug 2026 13:35:42 +0000 Subject: [PATCH 04/11] fix(relay): stop one client's datagram ending another client's session The ingress parser accepted client datagrams the egress encoder then refused, and that refusal landed on the receiving client rather than the sender. Two shapes got through. An empty payload: Datagrams::from_bytes needs only the ECN byte, so a 34-byte frame decodes to empty contents, which RelayedStream::start_send rejects with SendError::EmptyPacket. And a one-byte overflow: ingress bounds the frame *after* the type varint (payload <= 65503), while the forwarded RelayToClientMsg carries its own type byte on top of the same payload, for 65537 > MAX_PACKET_SIZE. Either way send_packet returned Err, run_inner propagated it, and Actor::run logged "actor errored, exiting" and dropped the actor, whose Drop unregisters the client and closes its stream. The honest client cannot produce either frame -- client/conn.rs blocks empty sends locally -- so the checks read as unreachable, but a hand-rolled client reaches both trivially, aimed at any EndpointId it knows. Reject both at ingress, bounding by the *forwarded* length rather than the received one. And make the egress refusals non-fatal: a packet on that queue was put there by another client, so drop it (it is already counted in send_packets_dropped) and keep the session, matching how every other forwarding failure is handled -- handle_frame logs and continues, a full send queue drops. Only genuine stream errors, which mean this connection is already broken, stay fatal. Not a security finding: the impact is availability only. Verified there is no misrouting, src-spoofing or data-leak variant -- src always comes from the authenticated handshake identity, dst is an exact 32-byte map lookup, and start_send returns before touching the inner sink, so no partial frame is ever emitted. Two tests, both failing first: one asserts everything the parser accepts survives re-encoding, the other that an undeliverable packet is dropped while the session stays live and later packets still arrive. The proptest generator no longer emits empty datagrams, which are not valid relayed packets in either direction. krikos-relay 97/97 with --all-features. Co-Authored-By: Claude Opus 5 --- krikos-relay/src/protos/relay.rs | 86 +++++++++++++++++++++++-- krikos-relay/src/server/client.rs | 100 +++++++++++++++++++++++++++++- 2 files changed, 180 insertions(+), 6 deletions(-) diff --git a/krikos-relay/src/protos/relay.rs b/krikos-relay/src/protos/relay.rs index 3dbebcf89b1..9e08d3a533e 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,59 @@ 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(); + // 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. + let max_payload = MAX_PACKET_SIZE - EndpointId::LENGTH - 1 /* ECN */; + for len in [0, 1, max_payload - 1, max_payload] { + for is_batch in [false, true] { + 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; + }; + 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() + ); + } + } + } + /// 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. @@ -932,7 +1005,10 @@ mod proptests { ( ecn(), prop::option::of(MAX_PAYLOAD_SIZE / 20..MAX_PAYLOAD_SIZE), - vec(any::(), 0..MAX_PAYLOAD_SIZE), + // 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. + vec(any::(), 1..MAX_PAYLOAD_SIZE), ) .prop_map(|(ecn, segment_size, data)| Datagrams { ecn, diff --git a/krikos-relay/src/server/client.rs b/krikos-relay/src/server/client.rs index ff1eeb3bab4..a78c39a4cbb 100644 --- a/krikos-relay/src/server/client.rs +++ b/krikos-relay/src/server/client.rs @@ -510,7 +510,24 @@ 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. + match &err { + WriteFrameError::Stream { + source: + RelaySendError::EmptyPacket { .. } + | RelaySendError::ExceedsMaxPacketSize { .. }, + .. + } => { + warn!("dropping undeliverable packet: {err:#}"); + Ok(()) + } + _ => Err(err), + } } } } @@ -784,6 +801,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 { From b0a6dfd364996d8f87e52a00fe0d424644719fee Mon Sep 17 00:00:00 2001 From: jonaswre Date: Tue, 4 Aug 2026 14:29:05 +0000 Subject: [PATCH 05/11] fix(blobs): close the gaps the per-request-type event mask opened MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Routing each request type through its own mask field is right, but it un-gated every config that had only ever set `mask.get` and relied on the old shared dispatch: - `examples/limit.rs::limit_by_hash` now answers get_many and observe with the same allowlist instead of leaving them at `DEFAULT`, where a peer could read any blob by asking for it as a `GetManyRequest`. - `examples/random_store.rs` hands push to its handler rather than inheriting `ALL_READONLY`'s `Disabled`, which made `--allow-push` a no-op. `ObserveMode` gains `Disabled` — an observe response streams the local bitfield for a hash, so a locked-down provider needs to refuse it outright, which no mask value could express — and its conversion now maps to the `*Log` variants: an observe request transfers no blobs, so completion is the only update it can emit, and the plain variants dropped it. The push-denial test asserts `BlobStatus::NotFound` rather than `!has()`, which is false for a partial import, and keeps sampling until a deadline instead of once, since nothing on the denying node is observable from the test and a single early sample would pass on a broken mask. Co-Authored-By: Claude Opus 5 --- protocols/krikos-blobs/examples/limit.rs | 50 +++++++++++++------ .../krikos-blobs/examples/random_store.rs | 11 +++- protocols/krikos-blobs/src/provider/events.rs | 22 ++++++-- protocols/krikos-blobs/src/tests.rs | 31 +++++++++--- 4 files changed, 87 insertions(+), 27 deletions(-) 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/events.rs b/protocols/krikos-blobs/src/provider/events.rs index 672b250c681..a5f22b58df9 100644 --- a/protocols/krikos-blobs/src/provider/events.rs +++ b/protocols/krikos-blobs/src/provider/events.rs @@ -35,21 +35,33 @@ 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::Notify, - ObserveMode::Intercept => RequestMode::Intercept, + ObserveMode::Notify => RequestMode::NotifyLog, + ObserveMode::Intercept => RequestMode::InterceptLog, + ObserveMode::Disabled => RequestMode::Disabled, } } } diff --git a/protocols/krikos-blobs/src/tests.rs b/protocols/krikos-blobs/src/tests.rs index 4108a77dc3f..4c9e9403f8b 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,7 +15,10 @@ use tracing::info; use crate::{ BlobFormat, Hash, HashAndFormat, - api::{Store, blobs::Bitfield}, + api::{ + Store, + blobs::{Bitfield, BlobStatus}, + }, get, hashseq::HashSeq, net_protocol::BlobsProtocol, @@ -496,10 +499,26 @@ async fn push_is_rejected_when_disabled( test_data(size), "control: a node that permits pushes must receive the blob" ); - assert!( - !deny_store.blobs().has(hash).await?, - "a push rejected by the event mask must not write to the receiving store" - ); + // 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(()) From c65ede537e44d7307f8e5fd2d464a11f946ba264 Mon Sep 17 00:00:00 2001 From: jonaswre Date: Tue, 4 Aug 2026 14:29:14 +0000 Subject: [PATCH 06/11] fix(docs): keep clamp_to_namespace narrowing for every bound shape MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The catch-all arms replaced the caller's bound with the namespace bound, which widens rather than narrows: an `Included` end below the namespace end (what `RecordsBounds::author_key` with `KeyFilter::Exact` produces) became the whole namespace, and an `Excluded` start was re-admitted as `Included(namespace_start)`. No current caller hits either arm, but the function is `pub` and its doc promises the fallback is never wider. Handle the tighter shapes explicitly and extend the empty-range normalization to the bound combinations those arms can now produce. The regression test picked both namespaces at random, so a foreign namespace landed on one side of ours and only exercised the clamp facing it — each defect was caught half the time, unreproducibly. Pin one neighbour below and one above. It also only asserted that nothing leaked, which an over-clamp returning nothing satisfies; assert the expected count of our own entries per range as well. Co-Authored-By: Claude Opus 5 --- protocols/krikos-docs/src/store/fs.rs | 55 ++++++++++++++++---- protocols/krikos-docs/src/store/fs/bounds.rs | 29 ++++++++--- 2 files changed, 66 insertions(+), 18 deletions(-) diff --git a/protocols/krikos-docs/src/store/fs.rs b/protocols/krikos-docs/src/store/fs.rs index cb5e2cde198..63e6fd5731f 100644 --- a/protocols/krikos-docs/src/store/fs.rs +++ b/protocols/krikos-docs/src/store/fs.rs @@ -1072,10 +1072,28 @@ mod tests { let author = store.new_author(&mut rand::rng())?; let ours = NamespaceSecret::new(&mut rand::rng()); - let theirs = 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())?; - store.new_replica(theirs.clone())?; - for ns in [&ours, &theirs] { + 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")); @@ -1089,22 +1107,37 @@ mod tests { let max = RecordIdentifier::new(NamespaceId::from(&[255u8; 32]), author.id(), b""); let mut wrapper = StoreInstance::new(ours.id(), &mut store); - for (label, range) in [ - ("less", Range::new(min.clone(), max.clone())), - ("greater", Range::new(max.clone(), min.clone())), - ("equal", Range::new(min.clone(), min.clone())), + // 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 leaked = wrapper + let namespaces = wrapper .get_range(range)? .map(|entry| entry.map(|e| e.namespace())) - .collect::>>()? - .into_iter() - .filter(|ns| *ns != ours.id()) + .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(()) } diff --git a/protocols/krikos-docs/src/store/fs/bounds.rs b/protocols/krikos-docs/src/store/fs/bounds.rs index f06ff04f18e..5e75ab2fe45 100644 --- a/protocols/krikos-docs/src/store/fs/bounds.rs +++ b/protocols/krikos-docs/src/store/fs/bounds.rs @@ -70,30 +70,45 @@ impl RecordsBounds { /// selects nothing instead of reading it. pub fn clamp_to_namespace(self, ns: &NamespaceId) -> Self { let Self(start, end) = self; - // Both `namespace_start` and every caller's start are `Included`; the tighter of - // two lower bounds is the greater one. Unexpected shapes fall back to the - // namespace bound, which is never wider than what was asked for. + // `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 remote's own end is already within it. + // 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. - if let (Bound::Included(s), Bound::Excluded(e)) = (&start, &end) - && s >= e - { + 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)); } From 519a645edf43f69a7d4690836adee9d7d96a49cf Mon Sep 17 00:00:00 2001 From: jonaswre Date: Tue, 4 Aug 2026 14:29:23 +0000 Subject: [PATCH 07/11] fix(relay): make non-stream send errors non-fatal by construction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `send_packet` listed the two droppable `SendError` variants and sent everything else to `Err`, the inverse of the policy the comment states. Both that enum and `WriteFrameError` are `#[non_exhaustive]`, so a future encode-validation variant would land in the fatal arm and let any peer end an unrelated client's session again, with no compile error. Match on the fatal set instead: a new `WriteFrameError` defaults to fatal, a new encode-side `SendError` defaults to a drop. The re-encodability test computed its boundary payload lengths for the packed layout only, so both batch boundary cases were rejected by the older ingress length check and never reached the new `forwarded_len` guard — deleting that guard's batch branch left the test green. Compute the boundary per layout and assert how many frames survived ingress, so the cases cannot go back to being silently skipped. Restoring empty-datagram coverage for the relay-to-client direction: narrowing the shared generator was only required for client-to-relay, but `RelayToClientMsg::from_bytes` still accepts zero-length contents and lost its only coverage. Split the strategy into `datagrams` and `forwardable_datagrams`. Co-Authored-By: Claude Opus 5 --- krikos-relay/src/protos/relay.rs | 64 +++++++++++++++++++++++-------- krikos-relay/src/server/client.rs | 21 +++++----- 2 files changed, 61 insertions(+), 24 deletions(-) diff --git a/krikos-relay/src/protos/relay.rs b/krikos-relay/src/protos/relay.rs index 9e08d3a533e..72476a2f75a 100644 --- a/krikos-relay/src/protos/relay.rs +++ b/krikos-relay/src/protos/relay.rs @@ -913,11 +913,20 @@ mod tests { fn accepted_client_datagrams_are_re_encodable() { let dst = SecretKey::from_bytes(&[7u8; 32]).public(); let src = SecretKey::from_bytes(&[8u8; 32]).public(); - // 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. - let max_payload = MAX_PACKET_SIZE - EndpointId::LENGTH - 1 /* ECN */; - for len in [0, 1, max_payload - 1, max_payload] { - for is_batch in [false, true] { + // 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()), @@ -934,6 +943,7 @@ mod tests { // 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}); \ @@ -951,6 +961,14 @@ mod tests { ); } } + // 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 @@ -999,16 +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), - // 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. - vec(any::(), 1..MAX_PAYLOAD_SIZE), + vec(any::(), min_len..MAX_PAYLOAD_SIZE), ) .prop_map(|(ecn, segment_size, data)| Datagrams { ecn, @@ -1058,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 a78c39a4cbb..084c376383d 100644 --- a/krikos-relay/src/server/client.rs +++ b/krikos-relay/src/server/client.rs @@ -516,16 +516,19 @@ where // 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: - RelaySendError::EmptyPacket { .. } - | RelaySendError::ExceedsMaxPacketSize { .. }, - .. - } => { - warn!("dropping undeliverable packet: {err:#}"); - Ok(()) - } + WriteFrameError::Stream { source, .. } => match source { + RelaySendError::StreamError { .. } => Err(err), + _ => { + warn!("dropping undeliverable packet: {err:#}"); + Ok(()) + } + }, _ => Err(err), } } From 1b7514cab2c2032bf6dbe0e69492b727af4b744e Mon Sep 17 00:00:00 2001 From: jonaswre Date: Tue, 4 Aug 2026 14:29:31 +0000 Subject: [PATCH 08/11] fix(dns): pin the signature length and surface DHT verification failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `from_parts` took the public key as `&[u8; 32]` but the signature as `&[u8]`. The wire layout is fixed-offset, so any other length carves the timestamp and packet out of the wrong bytes and reports a signature failure instead of a length error. Take `&[u8; 64]`. `from_parts_unchecked` still recommended itself for "DHT mutable items", the exact unbound use the checked constructor was added to replace — a consumer following that doc reassembles an attacker-signed packet under the queried name. Say what it is actually safe for. A forged DHT item is the one thing the new check produces, and it was swallowed by `let Ok(packet)`: a poisoned item outranking the victim's record made the name unresolvable, logged identically to a plain miss. Log the rejection and count it. Co-Authored-By: Claude Opus 5 --- krikos-dns-server/src/metrics.rs | 5 +++++ krikos-dns-server/src/store.rs | 33 ++++++++++++++++++++++---------- krikos-dns/src/pkarr.rs | 14 +++++++++++--- 3 files changed, 39 insertions(+), 13 deletions(-) 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 b99b23461b1..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, pubkey) - { - 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) } diff --git a/krikos-dns/src/pkarr.rs b/krikos-dns/src/pkarr.rs index f30cc013013..4a03feb7411 100644 --- a/krikos-dns/src/pkarr.rs +++ b/krikos-dns/src/pkarr.rs @@ -135,9 +135,13 @@ impl SignedPacket { /// 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], + signature: &[u8; 64], timestamp: Timestamp, encoded_packet: &[u8], ) -> Result { @@ -272,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], From ef8bd5f64d9e0f5acdaaf6b6f4439f29f681ffdc Mon Sep 17 00:00:00 2001 From: jonaswre Date: Tue, 4 Aug 2026 14:58:14 +0000 Subject: [PATCH 09/11] chore(determinism): re-baseline the boundary inventory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both inventories are line-sensitive, so the security fixes on this branch moved sixteen recorded occurrences and CI has been red on the boundary check since the relay commit landed. The only content change is one addition: the `tokio::task::spawn` in `server/client.rs::tests::undeliverable_packet_does_not_end_the_session`. It falls under the existing "direct spawns in `#[cfg(test)]` regions" row — it owns only the actor under test, the test joins it before returning, and no simulator backend executes that module — so it is classified as acceptable nondeterminism and the row now names it. Co-Authored-By: Claude Opus 5 --- docs/testing/determinism-audit.md | 2 +- scripts/determinism-boundaries.semantic.txt | 1 + scripts/determinism-boundaries.txt | 33 +++++++++++---------- 3 files changed, 19 insertions(+), 17 deletions(-) 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/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(), From cae5306b5d525f464357e00fa3912981be1bdbf9 Mon Sep 17 00:00:00 2001 From: jonaswre Date: Tue, 4 Aug 2026 15:01:56 +0000 Subject: [PATCH 10/11] fix(test): stop asserting on the pusher-side result of a refused push MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `push_is_rejected_when_disabled` propagated the `execute_push_sink` result for both targets, including the node that refuses. Whether a refusal reaches the pusher is a race: the provider resets the stream, and `execute_push_sink` sees that only if the reset lands before it has finished writing. On the Windows runner it did, and the test failed at the push rather than at any assertion — "sending stopped by peer: error 0". That result was never the property under test; the doc comment on the function already says the refusal is unobservable from the pusher and that the blob not landing is what matters. Tolerate either outcome for the denying node, and keep requiring success from the permitting one, since that control is what stops a passing deny assertion from being vacuous. Co-Authored-By: Claude Opus 5 --- protocols/krikos-blobs/src/tests.rs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/protocols/krikos-blobs/src/tests.rs b/protocols/krikos-blobs/src/tests.rs index 4c9e9403f8b..1b99a1322f2 100644 --- a/protocols/krikos-blobs/src/tests.rs +++ b/protocols/krikos-blobs/src/tests.rs @@ -479,15 +479,24 @@ async fn push_is_rejected_when_disabled( // 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 in [&deny, &allow] { + for (target, must_succeed) in [(&deny, false), (&allow, true)] { let conn = pusher .endpoint() .connect(target.endpoint().addr(), crate::ALPN) .await?; - pusher_store + let res = pusher_store .remote() .execute_push_sink(conn.clone(), request.clone(), Drain) - .await?; + .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); } From da3acba367126879f7db485a0f0847981192524f Mon Sep 17 00:00:00 2001 From: jonaswre Date: Tue, 4 Aug 2026 17:20:52 +0000 Subject: [PATCH 11/11] fix(test): stop the docs sync tests failing on interleaved live events These four have been failing on the macOS-arm runner, a different one each run, since before this branch: main's own CI shows test_download_policies timing out on 3a9a5b3a (2026-07-31). None of them is a sync defect. The ordered assertions in sync_full_basic required an exact event sequence from a live document that emits more than the sequence under test. A second SyncFinished -- the ticket-driven sync and the NeighborUp-driven sync both run, which the optional matchers in that same test already document -- or a PendingContentReady from a draining download queue landing between two asserted events failed the whole test. Add assert_next_ignoring, which skips a tolerated event without consuming a matcher, and give those sites the two events that interleave. Ordering stays asserted, which switching to assert_next_unordered_with_optionals would have given up; expected matchers take precedence, so a required SyncFinished is still matched, not skipped. assert_next has no callers left, so it goes. sync_restart_node and test_sync_via_relay accepted only `result.is_ok()` sync events. node1 has just restarted, so the first attempt can still land on the connection node2 held to the dead process ("Failed to close connection1"), and a relayed connection can drop mid-sync. Sync retries in both cases. Tolerate one failed attempt; the required matchers still demand a successful sync plus the entry and its content. test_sync_via_relay's join also had a 2s budget to finish a relay handshake, sync, insert and download -- it now uses the same TIMEOUT as every other wait here. test_download_policies is left doing what it did, but able to say why it stopped. Its select! arms used the refutable pattern `Ok(Some(ev)) = try_next()`; a stream that ends or errors fails to match, which disables that arm and leaves the loop waiting on the other one until the 120s timeout, reported as "timeout elapsed" and nothing else. Bind the whole Result and fail where it breaks, with the counters and downloaded keys. That is a diagnosis, not a fix: the root cause of the hang is not established, and this is what makes the next occurrence readable. 20 iterations of all four under 6-way CPU saturation: 0/80 failures. Co-Authored-By: Claude Opus 5 --- protocols/krikos-docs/tests/sync.rs | 135 +++++++++++++++++++++++----- 1 file changed, 112 insertions(+), 23 deletions(-) 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 {