fix: three access-control bugs found by a security review, plus a relay teardown bug

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. |
 

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

krikos-dns-server/src/store.rs

@@ -163,17 +163,30 @@ impl ZoneStore {
                 .as_async()
                 .get_mutable_most_recent(pubkey.as_bytes(), None)
                 .await;
-            if let Some(item) = maybe_item
-                && let Ok(packet) = mutable_item_to_signed_packet(&item)
-            {
-                debug!("DHT resolve successful {:?}", packet);
-                return self
-                    .cache
-                    .lock()
-                    .await
-                    .insert_and_resolve_dht(&packet, name, record_type);
+            match maybe_item {
+                // Keep the rejection distinguishable from a plain miss: a packet that
+                // was returned but does not verify against the key we asked for is the
+                // observable trace of an attempt to poison this resolver, and it is the
+                // only signal the verification produces.
+                Some(item) => match mutable_item_to_signed_packet(&item, pubkey) {
+                    Ok(packet) => {
+                        debug!("DHT resolve successful {:?}", packet);
+                        return self.cache.lock().await.insert_and_resolve_dht(
+                            &packet,
+                            name,
+                            record_type,
+                        );
+                    }
+                    Err(err) => {
+                        self.metrics.dht_packets_rejected.inc();
+                        warn!(
+                            pubkey = %pubkey.to_z32(),
+                            "DHT returned a packet that does not verify against the requested key: {err:#}"
+                        );
+                    }
+                },
+                None => debug!("DHT resolve failed"),
             }
-            debug!("DHT resolve failed");
         }
         Ok(None)
     }
@@ -215,17 +228,27 @@ fn finish_mainline_build(result: std::io::Result<Dht>) -> Result<Dht> {
     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<SignedPacket, SignedPacketVerifyError> {
     let timestamp = u64::try_from(item.seq()).map_err(|_| {
         n0_error::e!(SignedPacketVerifyError::InvalidTimestamp {
             timestamp: item.seq(),
         })
     })?;
-    SignedPacket::from_parts_unchecked(
-        item.key(),
+    // Built from `expected` rather than `item.key()`, so a packet signed by any other
+    // key fails signature verification instead of being silently re-attributed.
+    SignedPacket::from_parts(
+        expected.as_bytes(),
         item.signature(),
         Timestamp::from_micros(timestamp),
         item.value(),
@@ -404,6 +427,41 @@ mod tests {
         Ok(())
     }
 
+    /// A DHT response must be bound to the key we asked for.
+    ///
+    /// `mainline` verifies a mutable item's signature against the key carried in the
+    /// response, not against the queried target, and stores the target verbatim — so
+    /// "signed by somebody" is all it guarantees. Any DHT node can therefore answer a
+    /// lookup for one key with a packet correctly signed under a key it controls. Were
+    /// that accepted, its records would be re-served under the queried name, which is
+    /// exactly the binding pkarr exists to provide.
+    #[test]
+    fn dht_item_signed_by_another_key_is_rejected() -> Result {
+        let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(1);
+        // A perfectly valid packet — just not the one that was asked for.
+        let attacker_packet = random_signed_packet(&mut rng)?;
+        let attacker_key = PublicKeyBytes::from_signed_packet(&attacker_packet);
+        // pkarr carries the packet timestamp as the BEP-0044 sequence number.
+        let item = MutableItem::new_signed_unchecked(
+            *attacker_key.as_bytes(),
+            attacker_packet.signature().to_bytes(),
+            attacker_packet.encoded_packet(),
+            i64::try_from(attacker_packet.timestamp().as_micros())
+                .expect("pkarr timestamp fits in a mainline sequence number"),
+            None,
+        );
+
+        // Resolved under the key that actually signed it: accepted.
+        mutable_item_to_signed_packet(&item, &attacker_key)
+            .expect("a packet signed by the key we asked for must be accepted");
+
+        // Resolved under someone else's key: refused.
+        let victim = PublicKeyBytes::from_signed_packet(&random_signed_packet(&mut rng)?);
+        mutable_item_to_signed_packet(&item, &victim)
+            .expect_err("a packet signed by an unrelated key must be refused");
+        Ok(())
+    }
+
     #[test]
     fn mainline_construction_failure_is_reported() {
         let error = finish_mainline_build(Err(std::io::Error::other(
@@ -419,16 +477,19 @@ mod tests {
     }
 
     #[test]
-    fn negative_mainline_sequence_is_rejected() {
-        let item = MutableItem::new_signed_unchecked([0; 32], [0; 64], &[], -1, None);
+    fn negative_mainline_sequence_is_rejected() -> Result {
+        let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(2);
+        let key = PublicKeyBytes::from_signed_packet(&random_signed_packet(&mut rng)?);
+        let item = MutableItem::new_signed_unchecked(*key.as_bytes(), [0; 64], &[], -1, None);
 
-        let error = mutable_item_to_signed_packet(&item)
+        let error = mutable_item_to_signed_packet(&item, &key)
             .expect_err("negative mainline sequence must not wrap into a timestamp");
 
         assert!(matches!(
             error,
             SignedPacketVerifyError::InvalidTimestamp { timestamp: -1, .. }
         ));
+        Ok(())
     }
 
     fn random_signed_packet<R: CryptoRng + ?Sized>(rng: &mut R) -> Result<SignedPacket> {

krikos-dns/src/pkarr.rs

@@ -130,6 +130,29 @@ impl SignedPacket {
         Self::from_bytes(&bytes)
     }
 
+    /// Assemble a signed packet from its parts, verifying it against `public_key`.
+    ///
+    /// Use this over [`SignedPacket::from_parts_unchecked`] whenever the parts come from
+    /// somewhere that does not itself bind them to the key being resolved, such as a DHT
+    /// response.
+    /// Both fixed-size arrays: the wire layout is `public_key || signature || timestamp
+    /// || packet` at fixed offsets, so a signature of any other length would silently
+    /// carve the timestamp and the packet out of the wrong bytes and report a
+    /// signature-verification failure instead of a length error.
+    pub fn from_parts(
+        public_key: &[u8; 32],
+        signature: &[u8; 64],
+        timestamp: Timestamp,
+        encoded_packet: &[u8],
+    ) -> Result<Self, SignedPacketVerifyError> {
+        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(&timestamp.to_be_bytes());
+        bytes.extend_from_slice(encoded_packet);
+        Self::from_bytes(&bytes)
+    }
+
     /// Parse a signed packet without verifying the signature.
     ///
     /// Still validates minimum length and DNS parsing.
@@ -253,8 +276,12 @@ impl SignedPacket {
 
     /// Reconstruct a signed packet from its raw parts without verifying the signature.
     ///
-    /// This is useful for reconstructing a packet from storage or DHT mutable items
-    /// where the components are stored separately.
+    /// Only for parts that are already bound to the key they will be served under, such
+    /// as rows of a store this node wrote itself. Never for anything a peer supplied:
+    /// the parts of a DHT mutable item carry the *publisher's* key and signature, not
+    /// the key that was queried, so reassembling one here re-serves an attacker-signed
+    /// packet under the name that was looked up. Use [`SignedPacket::from_parts`] for
+    /// those, which verifies against the key you pass in.
     pub fn from_parts_unchecked(
         public_key: &[u8],
         signature: &[u8],

krikos-relay/src/protos/relay.rs

@@ -549,10 +549,30 @@ impl ClientToRelayMsg {
                 ensure!(content.len() >= EndpointId::LENGTH, Error::InvalidFrame);
 
                 let dst_endpoint_id = cache.key_from_slice(&content[..EndpointId::LENGTH])?;
-                let datagrams = Datagrams::from_bytes(
-                    content.slice(EndpointId::LENGTH..),
-                    frame_type == FrameType::ClientToRelayDatagramBatch,
-                )?;
+                let is_batch = frame_type == FrameType::ClientToRelayDatagramBatch;
+                let datagrams =
+                    Datagrams::from_bytes(content.slice(EndpointId::LENGTH..), is_batch)?;
+
+                // Only accept what can be forwarded. The relay re-encodes this as a
+                // `RelayToClientMsg::Datagrams`, which carries its own frame type byte on
+                // top of the same payload, so the egress frame is larger than the one
+                // checked above. The egress encoder also refuses empty contents. Either
+                // refusal would surface on the *receiving* client's stream, i.e. on a
+                // peer that did nothing, so both have to be caught here at ingress.
+                ensure!(!datagrams.contents.is_empty(), Error::InvalidFrame);
+                let forwarded_len = if is_batch {
+                    FrameType::RelayToClientDatagramBatch.encoded_len()
+                } else {
+                    FrameType::RelayToClientDatagram.encoded_len()
+                } + EndpointId::LENGTH
+                    + datagrams.encoded_len();
+                ensure!(
+                    forwarded_len <= MAX_PACKET_SIZE,
+                    Error::FrameTooLarge {
+                        frame_len: forwarded_len
+                    }
+                );
+
                 Self::Datagrams {
                     dst_endpoint_id,
                     datagrams,
@@ -880,6 +900,77 @@ mod tests {
         assert_eq!(&encoded[5..9], &u32::MAX.to_be_bytes());
     }
 
+    /// Anything the ingress parser accepts must be re-encodable towards its destination.
+    ///
+    /// The relay forwards a `ClientToRelayMsg::Datagrams` on as a
+    /// `RelayToClientMsg::Datagrams`, which is one frame-type byte larger because it
+    /// carries the sender's id in place of the destination's. `RelayedStream::start_send`
+    /// refuses to encode an empty or oversized frame — and that refusal lands on the
+    /// *receiving* client, whose actor treats a send failure as fatal. So any frame the
+    /// ingress parser lets through but the egress encoder rejects is a frame one client
+    /// can aim at another. Accept only what can be forwarded.
+    #[test]
+    fn accepted_client_datagrams_are_re_encodable() {
+        let dst = SecretKey::from_bytes(&[7u8; 32]).public();
+        let src = SecretKey::from_bytes(&[8u8; 32]).public();
+        // Number of frames that made it past the ingress parser, to prove the boundary
+        // lengths below were actually exercised rather than skipped by the `continue`.
+        let mut accepted = 0;
+        for is_batch in [false, true] {
+            // The largest payload whose *ingress* frame still fits, and its neighbours:
+            // the egress frame is one byte longer, so the boundary cannot be the same. A
+            // batch also carries a two-byte segment size, so its boundary sits two bytes
+            // lower; using the packed layout's boundary for both would push the batch
+            // cases past the plain `frame_len` check and never reach the new one.
+            let max_payload = MAX_PACKET_SIZE
+                - EndpointId::LENGTH
+                - 1 /* ECN */
+                - if is_batch { 2 /* segment size */ } else { 0 };
+            for len in [0, 1, max_payload - 1, max_payload] {
+                let datagrams = Datagrams {
+                    ecn: None,
+                    segment_size: is_batch.then(|| NonZeroU16::new(1).unwrap()),
+                    contents: vec![0u8; len].into(),
+                };
+                let encoded = ClientToRelayMsg::Datagrams {
+                    dst_endpoint_id: dst,
+                    datagrams,
+                }
+                .write_to(Vec::new());
+                let Ok(ClientToRelayMsg::Datagrams { datagrams, .. }) =
+                    ClientToRelayMsg::from_bytes(encoded.into(), &KeyCache::test())
+                else {
+                    // Refused at ingress: nothing to forward, which is the safe outcome.
+                    continue;
+                };
+                accepted += 1;
+                assert!(
+                    !datagrams.contents.is_empty(),
+                    "accepted an empty datagram (len {len}, batch {is_batch}); \
+                     forwarding it fails on the receiver's stream"
+                );
+                let forwarded = RelayToClientMsg::Datagrams {
+                    remote_endpoint_id: src,
+                    datagrams,
+                };
+                assert!(
+                    forwarded.encoded_len() <= MAX_PACKET_SIZE,
+                    "accepted a datagram that does not fit once forwarded: \
+                     {} bytes (len {len}, batch {is_batch})",
+                    forwarded.encoded_len()
+                );
+            }
+        }
+        // Per layout: `1` and `max_payload - 1` are forwardable, `0` is empty and
+        // `max_payload` overflows by the egress frame-type byte. If this count drops,
+        // the boundary arithmetic above has drifted and the cases that matter are being
+        // skipped instead of checked.
+        assert_eq!(
+            accepted, 4,
+            "expected both forwardable boundary lengths of both layouts to be accepted"
+        );
+    }
+
     /// A datagram frame must contain at least an EndpointId (32 bytes) after
     /// the frame type. A frame consisting only of the frame type byte used to
     /// panic when slicing the destination endpoint id.
@@ -926,13 +1017,31 @@ mod proptests {
         })
     }
 
+    /// Any datagram, including an empty one.
+    ///
+    /// The relay-to-client direction has no non-empty rule — a relay can send a
+    /// zero-length datagram and the client will decode it — so that direction has to
+    /// keep generating them.
     fn datagrams() -> impl Strategy<Value = Datagrams> {
+        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<Value = Datagrams> {
+        datagrams_of_len(1)
+    }
+
+    fn datagrams_of_len(min_len: usize) -> impl Strategy<Value = Datagrams> {
         // The max payload size (conservatively, since with segment_size = 0 we'd have slightly more space)
         const MAX_PAYLOAD_SIZE: usize = MAX_PACKET_SIZE - EndpointId::LENGTH - 1 /* ECN bytes */ - 2 /* segment size */;
         (
             ecn(),
             prop::option::of(MAX_PAYLOAD_SIZE / 20..MAX_PAYLOAD_SIZE),
-            vec(any::<u8>(), 0..MAX_PAYLOAD_SIZE),
+            vec(any::<u8>(), min_len..MAX_PAYLOAD_SIZE),
         )
             .prop_map(|(ecn, segment_size, data)| Datagrams {
                 ecn,
@@ -982,12 +1091,13 @@ mod proptests {
     }
 
     fn client_relay_frame() -> impl Strategy<Value = ClientToRelayMsg> {
-        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::<u8>()).prop_map(ClientToRelayMsg::Ping);
         let pong = prop::array::uniform8(any::<u8>()).prop_map(ClientToRelayMsg::Pong);
         prop_oneof![send_packet, ping, pong]