numinnex commented on code in PR #3425:
URL: https://github.com/apache/iggy/pull/3425#discussion_r3371687278
##########
core/message_bus/src/replica/listener.rs:
##########
@@ -210,30 +238,215 @@ pub async fn run(
}
}
-/// Read the 256 B `Ping` frame, enforce command + cluster match and the
-/// directional rule, return the announced replica id. No transport-level
-/// authentication.
-#[allow(clippy::future_not_send)]
+/// Decrements the shared in-flight handshake counter when the spawned task
+/// ends, on any exit path (success, error, or grace timeout). Pairs with the
+/// pre-spawn increment in [`run`] to keep [`MAX_INFLIGHT_HANDSHAKES`]
accurate.
+struct InflightGuard(Rc<Cell<usize>>);
+
+impl Drop for InflightGuard {
+ fn drop(&mut self) {
+ self.0.set(self.0.get() - 1);
+ }
+}
+
+/// Read the 256 B `ReplicaHello` frame, enforce command + cluster match and
the
+/// directional rule, then (when `auth` is configured) run the acceptor half
+/// of the mutual MAC handshake and return the cryptographically verified
+/// peer id. With `auth = None` the listener stays in legacy mode and returns
+/// the announced id unverified.
+///
+/// The dialer (the peer) holds the strictly lower id; this acceptor holds the
+/// higher id (see the directional rule). The transcript therefore binds
+/// `dialer_id = peer_id`, `acceptor_id = self_id`.
+///
+/// On a rejection an authenticated, still-waiting dialer is answered with a
+/// nonzero-status [`build_challenge_message`] (see [`reject`]) so it learns
the
+/// cause from its own logs rather than seeing a bare connection close.
+#[allow(clippy::future_not_send, clippy::similar_names)]
async fn handshake_read(
stream: &mut TcpStream,
our_cluster: u128,
self_id: u8,
replica_count: u8,
max_message_size: usize,
+ auth: Option<&ReplicaAuth>,
) -> Result<u8, IggyError> {
let msg = framing::read_message(stream, max_message_size).await?;
let header = msg.header();
- if header.command != Command2::Ping {
- return Err(IggyError::InvalidCommand);
+ let peer_id = header.replica;
+ let has_nonce = auth::has_nonce(&header.reserved_command);
+ // Only a peer that sent a nonce speaks the authenticated protocol and is
+ // waiting to read our response; a legacy (no-nonce) dialer delegates its
+ // fd without reading, so a reject frame would land in its VSR reader
instead.
+ let nackable = auth.is_some() && has_nonce;
+
+ if header.command != Command2::ReplicaHello {
+ return reject(
+ stream,
+ our_cluster,
+ self_id,
+ peer_id,
+ HandshakeStatus::UnknownCommand,
+ nackable,
+ )
+ .await;
}
if header.cluster != our_cluster {
- return Err(IggyError::InvalidCommand);
+ return reject(
+ stream,
+ our_cluster,
+ self_id,
+ peer_id,
+ HandshakeStatus::ClusterMismatch,
+ nackable,
+ )
+ .await;
}
// Directional rule: a replica only accepts inbound from peers with
// strictly lower ids. The peer is responsible for not dialing us if
// it has the higher id; this is just defensive.
- if header.replica >= replica_count || header.replica >= self_id {
- return Err(IggyError::InvalidCommand);
+ if peer_id >= replica_count || peer_id >= self_id {
+ return reject(
+ stream,
+ our_cluster,
+ self_id,
+ peer_id,
+ HandshakeStatus::DirectionalRule,
+ nackable,
+ )
+ .await;
+ }
Review Comment:
This is true for initial cluster bootstrapping, but for already existing
clusters that expand their replica set enforcing this rule could be
problematic, as rather having the new node initiate the challenge request, all
of the other nodes in the cluster would have to do it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
