RUST-BUG-110: bind peer service to HELLO admission

crates/emulebb-ed2k/src/ed2k_tcp/listener/session.rs

@@ -105,6 +105,37 @@ pub(in crate::ed2k_tcp) enum Ed2kSessionSource {
     },
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum PeerSessionPhase {
+    AwaitingInboundHello,
+    AwaitingOutboundHello,
+    Established,
+}
+
+const MAX_PRE_ESTABLISHMENT_FRAMES: usize = 8;
+const MAX_PRE_ESTABLISHMENT_BYTES: usize = 64 * 1024;
+
+impl PeerSessionPhase {
+    const fn allows(self, protocol: u8, opcode: u8) -> bool {
+        match self {
+            Self::AwaitingInboundHello => matches!(
+                (protocol, opcode),
+                (OP_EDONKEYPROT, OP_HELLO) | (OP_EMULEPROT, OP_PORTTEST)
+            ),
+            Self::AwaitingOutboundHello => matches!(
+                (protocol, opcode),
+                (OP_EDONKEYPROT, OP_HELLO | OP_HELLOANSWER)
+                    | (OP_EMULEPROT, OP_EMULEINFO | OP_EMULEINFOANSWER)
+            ),
+            Self::Established => true,
+        }
+    }
+
+    const fn sends_outbound_hello(self) -> bool {
+        matches!(self, Self::AwaitingOutboundHello)
+    }
+}
+
 pub(in crate::ed2k_tcp) struct Ed2kConnectionContext<'a> {
     pub(in crate::ed2k_tcp) dht: &'a DhtNode,
     pub(in crate::ed2k_tcp) server_state: &'a Arc<RwLock<Ed2kServerState>>,
@@ -155,8 +186,8 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
     };
     let response_identity =
         enrich_hello_identity(response_identity, server_state, kad_firewall).await;
-    let mut callback_expected_user_hash = None;
-    let (mut transport, promoted_grant, send_outbound_hello) = match source {
+    let mut outbound_expected_user_hash = None;
+    let (mut transport, mut pending_promoted_grant, mut session_phase) = match source {
         Ed2kSessionSource::Inbound(stream) => {
             let local_addr = stream.local_addr().with_context(|| {
                 format!("failed to resolve local eD2k listener address for {peer_addr}")
@@ -189,7 +220,7 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
                     anyhow::bail!("timed out waiting for initial eD2k peer bytes");
                 }
             };
-            (transport, None, false)
+            (transport, None, PeerSessionPhase::AwaitingInboundHello)
         }
         Ed2kSessionSource::CallbackUpload {
             transport,
@@ -201,8 +232,8 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
                 "callback_connect",
                 || "adopting outbound callback transport".to_string(),
             );
-            callback_expected_user_hash = expected_user_hash;
-            (*transport, None, true)
+            outbound_expected_user_hash = expected_user_hash;
+            (*transport, None, PeerSessionPhase::AwaitingOutboundHello)
         }
         // A slot grant for a disconnected waiter arrives on a connection WE
         // dialed (master AddUpNextClient US_CONNECTING connect-out); the
@@ -211,19 +242,27 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
             dump_ed2k_tcp_listener_meta(peer_addr, Some(transport.mode), "promote_connect", || {
                 format!("file_hash={}", grant.file_hash)
             });
-            (*transport, Some(*grant), true)
+            outbound_expected_user_hash = grant.peer.user_hash;
+            (
+                *transport,
+                Some(*grant),
+                PeerSessionPhase::AwaitingOutboundHello,
+            )
         }
     };
     let _peer_connection_guard = transfer_runtime.register_peer_connection(peer_addr.ip());
-    transport
-        .stream
-        .set_nodelay(true)
-        .with_context(|| format!("failed to enable TCP_NODELAY for peer {peer_addr}"))?;
+    if let Err(error) = transport.stream.set_nodelay(true) {
+        if let Some(grant) = pending_promoted_grant.take() {
+            transfer_runtime.release_upload_session(&grant.handle).await;
+        }
+        return Err(error)
+            .with_context(|| format!("failed to enable TCP_NODELAY for peer {peer_addr}"));
+    }
     debug!(
         "eD2k TCP peer session with {peer_addr} transport={}",
         transport.mode.as_str()
     );
-    if promoted_grant.is_none() {
+    if pending_promoted_grant.is_none() {
         dump_ed2k_tcp_listener_meta(peer_addr, Some(transport.mode), "accept", || {
             format!("udp_port={kad_udp_port}")
         });
@@ -259,39 +298,27 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
     // a flood of these is a share-probe and the peer is banned (MFC file_request_flood).
     let mut failed_file_req_count: u32 = 0;
 
+    // Seed locally-known scheduling facts before HELLO binding. The queued hash
+    // is only an expected initial identity, not a bound live identity: a stale
+    // endpoint mismatch closes without the impersonation ban reserved for a
+    // later changed HELLO on an already-bound session.
+    if let Some(grant) = pending_promoted_grant.as_ref() {
+        peer_upload_identity = grant.peer.clone();
+        peer_upload_identity.should_crypt = transport.mode.is_obfuscated();
+    }
+
     // Every outbound peer session announces itself before entering the shared
-    // dispatcher. Callback sessions then wait for the requester's normal file
-    // requests; promoted-upload sessions additionally attach and push their
-    // already-granted upload slot.
-    if send_outbound_hello {
+    // dispatcher. Promoted uploads keep their grant pending until a valid live
+    // HELLO identity is bound; only then may OP_ACCEPTUPLOADREQ be sent.
+    if session_phase.sends_outbound_hello() {
         let hello_packet = encode_hello_request(response_identity);
         dump_ed2k_tcp_listener_send(peer_addr, transport.mode, "hello_request", &hello_packet);
-        transport
-            .write_all(&hello_packet)
-            .await
-            .with_context(|| format!("failed to send OP_HELLO to outbound peer {peer_addr}"))?;
-    }
-    if let Some(grant) = promoted_grant {
-        peer_upload_identity = grant.peer.clone();
-        peer_upload_identity.should_crypt = transport.mode.is_obfuscated();
-        bound_user_hash = grant.peer.user_hash;
-        let file_hash = Ed2kHash::from_str(&grant.file_hash)
-            .with_context(|| format!("invalid promoted upload file hash {}", grant.file_hash))?;
-        requested_file_hash = Some(file_hash);
-        let attached = upload_queue
-            .attach_granted_session(
-                transfer_runtime,
-                &grant.peer,
-                grant.handle,
-                file_hash,
-                &mut transport,
-                peer_addr,
-            )
-            .await?;
-        if !attached {
-            // The grant went stale while connecting (aged out or re-owned by an
-            // inbound reconnect): nothing to serve on this connection.
-            return Ok(());
+        if let Err(error) = transport.write_all(&hello_packet).await {
+            if let Some(grant) = pending_promoted_grant.take() {
+                transfer_runtime.release_upload_session(&grant.handle).await;
+            }
+            return Err(error)
+                .with_context(|| format!("failed to send OP_HELLO to outbound peer {peer_addr}"));
         }
     }
 
@@ -308,9 +335,17 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
     // Last time the peer actually sent us a packet: feeds the waiting-connection
     // idle close (oracle socket timeout) inside poll_on_timeout.
     let mut last_packet_at = tokio::time::Instant::now();
+    let handshake_deadline = last_packet_at + ED2K_CONNECTION_IDLE_TIMEOUT;
+    let mut pre_establishment_frames = 0_usize;
+    let mut pre_establishment_bytes = 0_usize;
     let result: Result<()> = async {
         loop {
-        let read_timeout = upload_queue.read_timeout();
+        let mut read_timeout = upload_queue.read_timeout();
+        if session_phase != PeerSessionPhase::Established {
+            read_timeout = read_timeout.min(
+                handshake_deadline.saturating_duration_since(tokio::time::Instant::now()),
+            );
+        }
         // While serving as an inbound buddy, also drain relay frames (OP_CALLBACK)
         // that handle_kad_callback_req pushes down this held socket.
         let packet = if let Some(hold) = buddy_hold.as_mut() {
@@ -406,9 +441,41 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
         last_packet_at = tokio::time::Instant::now();
         dump_ed2k_tcp_listener_recv(peer_addr, transport.mode, "session", &packet);
 
+        if session_phase != PeerSessionPhase::Established {
+            pre_establishment_frames = pre_establishment_frames.saturating_add(1);
+            pre_establishment_bytes = pre_establishment_bytes
+                .saturating_add(super::super::TCP_PACKET_HEADER_LEN)
+                .saturating_add(packet.payload.len());
+            if pre_establishment_frames > MAX_PRE_ESTABLISHMENT_FRAMES
+                || pre_establishment_bytes > MAX_PRE_ESTABLISHMENT_BYTES
+            {
+                debug!(
+                    "closing eD2k peer {peer_addr}: pre-establishment budget exceeded frames={pre_establishment_frames} bytes={pre_establishment_bytes}"
+                );
+                break Ok(());
+            }
+        }
+
+        if !session_phase.allows(packet.protocol, packet.opcode) {
+            debug!(
+                "closing eD2k peer {peer_addr}: packet protocol=0x{:02X} opcode=0x{:02X} is invalid during {session_phase:?}",
+                packet.protocol,
+                packet.opcode,
+            );
+            break Ok(());
+        }
+
         match (packet.protocol, packet.opcode) {
             (OP_EDONKEYPROT, OP_HELLO) => {
                 let hello_profile = decode_hello_profile(&packet.payload)?;
+                if !outbound_hello_identity_matches(
+                    session_phase,
+                    outbound_expected_user_hash,
+                    hello_profile.identity.user_hash,
+                    peer_addr,
+                ) {
+                    break Ok(());
+                }
                 if !apply_peer_hello_profile(
                     PeerHelloStateRefs {
                         bound_user_hash: &mut bound_user_hash,
@@ -425,6 +492,7 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
                 ) {
                     break Ok(());
                 }
+                session_phase = PeerSessionPhase::Established;
                 debug!(
                     "received eD2k OP_HELLO from {peer_addr} transport={} mule_hello={}",
                     transport.mode.as_str(),
@@ -463,6 +531,19 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
                         .await
                         .with_context(|| format!("failed to reply to OP_HELLO from {peer_addr}"))?;
                 }
+                if !attach_pending_promoted_grant(
+                    &mut pending_promoted_grant,
+                    &mut upload_queue,
+                    transfer_runtime,
+                    &peer_upload_identity,
+                    &mut requested_file_hash,
+                    &mut transport,
+                    peer_addr,
+                )
+                .await?
+                {
+                    break Ok(());
+                }
                 // A callback we originated for a detached LowID upload waiter
                 // owns a missed-turn entitlement, not a speculative active slot.
                 // Claim it only after the peer's HELLO has bound its identity,
@@ -600,12 +681,13 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
             }
             (OP_EDONKEYPROT, OP_HELLOANSWER) => {
                 let hello_profile = decode_hello_answer_profile(&packet.payload)?;
-                if callback_expected_user_hash.is_some_and(|expected| {
-                    expected != hello_profile.identity.user_hash
-                }) {
-                    debug!(
-                        "callback peer {peer_addr} answered with a different user hash than the route supplied"
-                    );
+                if !outbound_hello_identity_matches(
+                    session_phase,
+                    outbound_expected_user_hash,
+                    hello_profile.identity.user_hash,
+                    peer_addr,
+                ) {
+                    break Ok(());
                 }
                 if !apply_peer_hello_profile(
                     PeerHelloStateRefs {
@@ -623,10 +705,24 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
                 ) {
                     break Ok(());
                 }
+                session_phase = PeerSessionPhase::Established;
                 debug!(
                     "received eD2k OP_HELLOANSWER from {peer_addr} transport={}",
                     transport.mode.as_str()
                 );
+                if !attach_pending_promoted_grant(
+                    &mut pending_promoted_grant,
+                    &mut upload_queue,
+                    transfer_runtime,
+                    &peer_upload_identity,
+                    &mut requested_file_hash,
+                    &mut transport,
+                    peer_addr,
+                )
+                .await?
+                {
+                    break Ok(());
+                }
                 peer_secure_ident.peer_sec_ident = hello_profile.misc_options1.secure_ident;
                 peer_secure_ident.our_external_ip = reachability.get();
                 if hello_profile.supports_secure_ident && !peer_secure_ident.requested_peer_key {
@@ -1224,6 +1320,9 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
     }
     .await;
 
+    if let Some(grant) = pending_promoted_grant.take() {
+        transfer_runtime.release_upload_session(&grant.handle).await;
+    }
     upload_queue.release(transfer_runtime).await;
     // Release the inbound buddy slot when the held session ends so the
     // buddy-management loop can re-establish on a reconnect (oracle buddy-loss).
@@ -1234,6 +1333,54 @@ pub(in crate::ed2k_tcp) async fn handle_connection(
     result
 }
 
+fn outbound_hello_identity_matches(
+    session_phase: PeerSessionPhase,
+    expected_user_hash: Option<[u8; 16]>,
+    observed_user_hash: [u8; 16],
+    peer_addr: SocketAddr,
+) -> bool {
+    if session_phase != PeerSessionPhase::AwaitingOutboundHello
+        || expected_user_hash.is_none_or(|expected| expected == observed_user_hash)
+    {
+        return true;
+    }
+    debug!("closing outbound eD2k peer {peer_addr}: HELLO identity differs from the routed peer");
+    false
+}
+
+async fn attach_pending_promoted_grant(
+    pending_grant: &mut Option<Ed2kUploadGrantedPromotion>,
+    upload_queue: &mut ListenerUploadQueue,
+    transfer_runtime: &Ed2kTransferRuntime,
+    peer_identity: &Ed2kUploadPeerIdentity,
+    requested_file_hash: &mut Option<Ed2kHash>,
+    transport: &mut Ed2kTransport,
+    peer_addr: SocketAddr,
+) -> Result<bool> {
+    let Some(grant) = pending_grant.take() else {
+        return Ok(true);
+    };
+    let file_hash = match Ed2kHash::from_str(&grant.file_hash) {
+        Ok(file_hash) => file_hash,
+        Err(error) => {
+            transfer_runtime.release_upload_session(&grant.handle).await;
+            return Err(error)
+                .with_context(|| format!("invalid promoted upload file hash {}", grant.file_hash));
+        }
+    };
+    *requested_file_hash = Some(file_hash);
+    upload_queue
+        .attach_granted_session(
+            transfer_runtime,
+            peer_identity,
+            grant.handle,
+            file_hash,
+            transport,
+            peer_addr,
+        )
+        .await
+}
+
 async fn handle_udp_port_test_event(
     event: Result<u64, tokio::sync::broadcast::error::RecvError>,
     registration: &mut Option<PortTestRegistration>,
@@ -1455,6 +1602,10 @@ fn apply_peer_hello_profile(
     };
     state.upload_identity.banned =
         transfer_runtime.is_client_banned(peer_ban_ip, Some(&profile.identity.user_hash));
+    if state.upload_identity.banned {
+        debug!("disconnecting banned eD2k peer {peer_addr} after HELLO identity binding");
+        return false;
+    }
     true
 }
 

crates/emulebb-ed2k/src/ed2k_tcp/listener/session/upload_queue.rs

@@ -487,7 +487,7 @@ impl ListenerUploadQueue {
         &mut self,
         transfer_runtime: &Ed2kTransferRuntime,
         peer_identity: &Ed2kUploadPeerIdentity,
-        handle: Ed2kUploadSessionHandle,
+        mut handle: Ed2kUploadSessionHandle,
         file_hash: Ed2kHash,
         transport: &mut Ed2kTransport,
         peer_addr: SocketAddr,
@@ -497,6 +497,13 @@ impl ListenerUploadQueue {
         {
             return Ok(false);
         }
+        if transfer_runtime
+            .refresh_upload_session_peer(&mut handle, peer_identity.clone())
+            .await
+            != Ed2kUploadSessionStatus::Granted
+        {
+            return Ok(false);
+        }
         self.record_diag_peer(peer_identity);
         self.session = Some(handle);
         self.file_hash = Some(file_hash);