fix: close the two findings the security review left open

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. 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. |
+| 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` and `::tests::undecodable_frame_does_not_end_the_session`, which each drive 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. |
 

krikos-relay/src/server/client.rs

@@ -542,7 +542,23 @@ where
     ) -> Result<(), HandleFrameError> {
         trace!(?maybe_frame, "handle incoming frame");
         let frame = match maybe_frame {
-            Some(frame) => frame?,
+            Some(Ok(frame)) => frame,
+            // A frame we cannot decode costs this client one datagram, not its session.
+            // It is relaying for every peer it is talking to, so tearing the connection
+            // down over one unparsable frame is far out of proportion — and it is the
+            // mirror of `send_packet`, which already drops rather than disconnects when
+            // the *encoder* refuses. Only a broken stream is fatal, because then there
+            // is nothing left to keep.
+            //
+            // Nothing bounds how many of these a client may send, deliberately: the
+            // read side is already rate limited, so a client sending malformed frames
+            // gets no more of this server than one sending valid ones.
+            Some(Err(err @ RelayRecvError::Proto { .. })) => {
+                self.metrics.recv_frames_invalid.inc();
+                warn!("dropping undecodable frame: {err:#}");
+                return Ok(());
+            }
+            Some(Err(err)) => return Err(err.into()),
             None => return Err(e!(HandleFrameError::StreamTerminated)),
         };
 
@@ -885,6 +901,84 @@ mod tests {
         Ok(())
     }
 
+    /// A frame the server cannot decode must not end the sending client's session.
+    ///
+    /// The client is relaying for every peer it is talking to, so disconnecting it over
+    /// one unparsable frame loses all of that instead of one datagram. This is the
+    /// mirror of `undeliverable_packet_does_not_end_the_session`, which covers the same
+    /// property on the encode side.
+    ///
+    /// The frame is written straight onto the wire, because a conforming client cannot
+    /// produce one: `Conn::start_send` refuses to encode an empty datagram, which is
+    /// exactly why only a buggy or hostile peer sends it.
+    #[tokio::test]
+    #[traced_test]
+    async fn undecodable_frame_does_not_end_the_session() -> Result {
+        let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(23);
+        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 dst = 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 });
+
+        // A datagram frame with empty contents: well formed enough to reach the parser,
+        // rejected by it because the relay could not forward it on.
+        let undecodable = ClientToRelayMsg::Datagrams {
+            dst_endpoint_id: dst,
+            datagrams: Datagrams::from(&[][..]),
+        }
+        .write_to(Vec::new());
+        // Straight onto the byte sink, under `Conn`'s own encoder.
+        io_rw
+            .conn
+            .send(bytes::Bytes::from(undecodable))
+            .await
+            .anyerr()?;
+
+        // Queued behind it: if the session survived, this still gets answered.
+        let payload = [3u8; 8];
+        io_rw.send(ClientToRelayMsg::Ping(payload)).await?;
+        let frame = recv_frame(FrameType::Pong, &mut io_rw).await.anyerr()?;
+        assert_eq!(frame, RelayToClientMsg::Pong(payload));
+
+        assert!(
+            !handle.is_finished(),
+            "an undecodable frame must not end the session"
+        );
+        assert_eq!(
+            metrics.recv_frames_invalid.get(),
+            1,
+            "the undecodable frame should be counted"
+        );
+
+        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 {

krikos-relay/src/server/metrics.rs

@@ -46,6 +46,15 @@ pub struct Metrics {
     #[metrics(help = "Number of 'send' packets dropped.")]
     pub send_packets_dropped: Counter,
 
+    /// Frames received from a client that could not be decoded, and were dropped.
+    ///
+    /// The client stays connected. A frame this server cannot parse is one datagram lost,
+    /// not a broken connection, and the sending client is usually relaying for several
+    /// peers at once. Sustained growth here means a client is sending frames this server
+    /// does not accept — a version mismatch or a bug on its side.
+    #[metrics(help = "Number of frames received from a client that could not be decoded.")]
+    pub recv_frames_invalid: Counter,
+
     /// Intended to count sent frames other than datagrams, but currently unused (never incremented).
     #[metrics(help = "Number of packets sent that were not 'send' packets")]
     pub other_packets_sent: Counter,

protocols/krikos-blobs/src/api/remote.rs

@@ -37,8 +37,8 @@ use crate::{
     hashseq::{HashSeq, HashSeqIter},
     limits::PROGRESS_QUEUE_CAPACITY,
     protocol::{
-        ChunkRangesSeq, GetManyRequest, GetRequest, MAX_MESSAGE_SIZE, ObserveItem, ObserveRequest,
-        PushRequest, Request, RequestType,
+        ChunkRangesSeq, ERR_LIMIT, ERR_PERMISSION, GetManyRequest, GetRequest, MAX_MESSAGE_SIZE,
+        ObserveItem, ObserveRequest, PushRequest, Request, RequestType,
     },
     provider::events::{ClientResult, ProgressError},
     store::KRIKOS_BLOCK_SIZE,
@@ -607,8 +607,11 @@ impl Remote {
             payload_bytes_sent: 0,
             sender: progress,
         };
-        // we are not going to need this!
-        recv.stop(0u32.into()).anyerr()?;
+        // The receive half is kept open, and read at the end. It carries no data, but a
+        // provider that refuses the push resets it with the reason code
+        // (`handle_read_request_result` -> `writer.reset(e.code())`), and that reset is
+        // the only signal a refusal produces. Stopping this stream here -- before the
+        // request had even been written -- is what made a refused push report success.
         // write the request. Unlike for reading, we can just serialize it sync using postcard.
         let request = write_push_request(request, &mut send).await?;
         let mut request_ranges = request.ranges.iter_infinite();
@@ -623,6 +626,7 @@ impl Remote {
         if request.ranges.is_blob() {
             // we are done
             send.finish().anyerr()?;
+            push_accepted(&mut recv).await?;
             return Ok(Default::default());
         }
         let hash_seq = self.store().get_bytes(root).await?;
@@ -638,6 +642,7 @@ impl Remote {
             }
         }
         send.finish().anyerr()?;
+        push_accepted(&mut recv).await?;
         Ok(Default::default())
     }
 
@@ -817,6 +822,70 @@ impl Remote {
     }
 }
 
+/// Wait for the provider's verdict on a push we have finished sending.
+///
+/// The push protocol carries no response payload, so the provider's half of the stream
+/// is used purely as an acknowledgement: it is closed cleanly when the push was accepted
+/// and handled, and reset with a [`ProgressError`] code when it was refused — by the
+/// event mask, by an interceptor, or by a rate limit.
+///
+/// This is why the caller must not stop that stream. Waiting on it also makes a push
+/// synchronous with respect to the receiver: `execute_push` now returns once the blob
+/// has actually landed, rather than once the last byte has been handed to the socket.
+///
+/// A provider from before this acknowledgement existed simply drops its half without
+/// writing, which arrives as a clean end of stream and is read as acceptance — the same
+/// answer such a provider gave before.
+async fn push_accepted(recv: &mut krikos::endpoint::RecvStream) -> Result<()> {
+    // Scoped, not a module import: `RecvStreamExt` also carries a
+    // `read_length_prefixed` that would become ambiguous with the one the observe path
+    // above resolves today.
+    use krikos::endpoint::ReadError;
+
+    use crate::util::RecvStreamExt;
+
+    let err = match recv.expect_eof().await {
+        Ok(()) => return Ok(()),
+        Err(err) => err,
+    };
+    let reset = err
+        .get_ref()
+        .and_then(|e| e.downcast_ref::<ReadError>())
+        .and_then(|e| match e {
+            ReadError::Reset(code) => Some(code),
+            _ => None,
+        });
+    let Some(code) = reset else {
+        // Not a verdict at all: the connection died, or the stream produced data no
+        // push provider is supposed to send.
+        return Err(e!(PushError::Io, err).into());
+    };
+    let reason = if *code == ERR_PERMISSION {
+        "the receiver does not accept pushes from us".to_string()
+    } else if *code == ERR_LIMIT {
+        "the receiver is rate limiting us".to_string()
+    } else {
+        format!("the receiver rejected the push, code {code}")
+    };
+    Err(e!(PushError::Refused, AnyError::from_string(reason)).into())
+}
+
+/// Why a push did not complete.
+#[allow(missing_docs)]
+#[non_exhaustive]
+#[stack_error(derive, add_meta)]
+pub enum PushError {
+    /// The receiver refused the push. It has stored nothing.
+    #[error("push refused by the receiver")]
+    Refused { source: AnyError },
+    /// The acknowledgement could not be read.
+    #[error("failed to read the push acknowledgement")]
+    Io {
+        #[error(std_err)]
+        source: io::Error,
+    },
+}
+
 /// Failures for a get operation
 #[allow(missing_docs)]
 #[non_exhaustive]

protocols/krikos-blobs/src/tests.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashSet, io, ops::Range, path::PathBuf, time::Duration};
+use std::{collections::HashSet, io, ops::Range, path::PathBuf};
 
 use bao_tree::ChunkRanges;
 use bytes::Bytes;
@@ -454,13 +454,11 @@ async fn two_nodes_push_blobs_mem() -> TestResult<()> {
 /// 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.
+/// Two things are asserted, because either alone can pass vacuously: the push must fail
+/// for the pusher, and the blob must not be in the receiving store. `allow` is the
+/// control — pushing the same blob to a node that permits pushes proves the push path
+/// works in this setup, so a failing "denied" push cannot be an artifact of a push that
+/// never happened.
 async fn push_is_rejected_when_disabled(
     pusher: Router,
     pusher_store: &Store,
@@ -479,7 +477,7 @@ 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, must_succeed) in [(&deny, false), (&allow, true)] {
+    for (label, target, should_succeed) in [("deny", &deny, false), ("allow", &allow, true)] {
         let conn = pusher
             .endpoint()
             .connect(target.endpoint().addr(), crate::ALPN)
@@ -488,46 +486,31 @@ async fn push_is_rejected_when_disabled(
             .remote()
             .execute_push_sink(conn.clone(), request.clone(), Drain)
             .await;
-        // Whether the refusal reaches the pusher at all is a race, which is exactly why
-        // the property under test is store contents rather than this result: the
-        // provider resets the stream, and `execute_push_sink` only sees that if the
-        // reset lands before it has finished writing. Both outcomes are correct for the
-        // denying node. The permitting node must succeed, or the control below — the
-        // thing that proves a passing deny assertion is not vacuous — proves nothing.
-        if must_succeed {
-            res?;
-        }
+        assert_eq!(
+            res.is_ok(),
+            should_succeed,
+            "{label}: push result was {res:?}"
+        );
         conns.push(conn);
     }
 
-    // The permissive node completing its push bounds the wait for the denying one:
-    // both were pushed over loopback, the denying one first.
+    // No polling window is needed for either store any more: a push is acknowledged, so
+    // by the time the permitting node's push returned `Ok` it had finished importing,
+    // and by the time the denying node's push returned `Err` it had already decided.
     allow_count_rx.changed().await?;
     assert_eq!(
         allow_store.get_bytes(hash).await?,
         test_data(size),
         "control: a node that permits pushes must receive the blob"
     );
-    // Nothing on the denying node is observable from here — a disabled request type
-    // emits no event, `execute_push_sink` does not wait for the receiver, and the
-    // provider imports on a detached task. Sampling `has` once could therefore run
-    // before an accepted push had landed and pass on a broken mask, so keep sampling
-    // for a while: an accepted push over loopback lands well inside this window.
-    let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
-    loop {
-        // `status`, not `has`: `has` is only true for a *complete* blob, so a rejection
-        // that arrived after some chunks had already been imported would leave
-        // attacker-chosen bytes in the store and still satisfy the assertion.
-        assert_eq!(
-            deny_store.blobs().status(hash).await?,
-            BlobStatus::NotFound,
-            "a push rejected by the event mask must not write to the receiving store"
-        );
-        if tokio::time::Instant::now() >= deadline {
-            break;
-        }
-        tokio::time::sleep(Duration::from_millis(50)).await;
-    }
+    // `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"
+    );
 
     tokio::try_join!(pusher.shutdown(), deny.shutdown(), allow.shutdown())?;
     Ok(())

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::undecodable_frame_does_not_end_the_session	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<ServerBuilder>::spawn	tokio::task::spawn	1
 spawn-task	krikos-relay/src/server/http_server/tests.rs	test_server_basic	tokio::spawn	1