fix(aich): require accepted authority before hashset activation

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

@@ -879,6 +879,12 @@ pub(in crate::ed2k_tcp) async fn drive_download_session(
                 }
                 (OP_EMULEPROT, OP_HASHSETANSWER2) => {
                     let hashset_answer = decode_hashset_answer2(&packet.payload)?;
+                    let aich_generation = if session_state.aich_hashset_answer_pending {
+                        session_state.aich_hashset_answer_pending = false;
+                        session_state.aich_hashset_requested_generation
+                    } else {
+                        None
+                    };
                     if !request_file_identifier
                         .matches_relaxed(&hashset_answer.file_identifier)
                     {
@@ -904,9 +910,13 @@ pub(in crate::ed2k_tcp) async fn drive_download_session(
                             .store_md4_hashset(file_hash_hex, hashset)
                             .await?;
                     }
-                    if let Some(hashset) = hashset_answer.aich_hashset {
+                    // Unrequested AICH sections cannot establish identity.
+                    // Preserve usable MD4 data while ignoring an untrusted tree.
+                    if let Some(hashset) = hashset_answer.aich_hashset
+                        && let Some(generation) = aich_generation
+                    {
                         manifest = transfer_runtime
-                            .store_aich_hashset(file_hash_hex, hashset)
+                            .store_aich_hashset_for_generation(file_hash_hex, hashset, generation)
                             .await?;
                         request_file_identifier = Ed2kFileIdentifier::from_manifest(&manifest)?;
                     }

crates/emulebb-ed2k/src/ed2k_tcp/download/session/startup.rs

@@ -178,11 +178,17 @@ pub(super) async fn advance_download_startup(step: DownloadStartupStep<'_>) -> R
         session_state.aich_file_hash_requested = true;
     }
 
+    let request_md4 = !manifest.md4_hashset_acquired && !session_state.hashset_requested;
+    let request_aich = session_state.remote_supports_file_identifiers
+        && manifest.file_size > ED2K_PART_SIZE
+        && manifest.aich_root.is_some()
+        && !manifest.aich_hashset_acquired
+        && session_state.peer_advertised_aich_root
+        && session_state.aich_hashset_requested_generation.is_none();
     if send_initial_requests
         && session_state.hello_complete
         && manifest.file_size != 0
-        && !manifest.md4_hashset_acquired
-        && !session_state.hashset_requested
+        && (request_md4 || request_aich)
         && session_state.startup_file_response_received
     {
         if manifest.file_size <= ED2K_PART_SIZE {
@@ -194,16 +200,10 @@ pub(super) async fn advance_download_startup(step: DownloadStartupStep<'_>) -> R
                 encode_hashset_request2(
                     request_file_identifier,
                     Ed2kHashsetRequestOptions {
-                        request_md4: true,
-                        // Request the AICH hashset only when a peer has advertised
-                        // an AICH root for this file. We can no longer key this off
-                        // a trusted `manifest.aich_root` (network-learned roots are
-                        // no longer promoted on first sight -- they need IP
-                        // corroboration), so track the peer-advertised signal
-                        // separately. This preserves the prior behaviour without
-                        // making the request depend on a trust decision.
-                        request_aich: manifest.file_size > ED2K_PART_SIZE
-                            && session_state.peer_advertised_aich_root,
+                        request_md4,
+                        // Stock requests part hashes for an already accepted
+                        // master. A peer advertisement is evidence, not trust.
+                        request_aich,
                     },
                 )?
             } else {
@@ -225,6 +225,11 @@ pub(super) async fn advance_download_startup(step: DownloadStartupStep<'_>) -> R
                     }
                 })?;
             session_state.hashset_requested = true;
+            if request_aich {
+                session_state.aich_hashset_requested_generation =
+                    Some(manifest.aich_trust_generation);
+                session_state.aich_hashset_answer_pending = true;
+            }
             session_state.hashset_requested_at = Some(tokio::time::Instant::now());
         }
     }

crates/emulebb-ed2k/src/ed2k_tcp/download/session/state.rs

@@ -89,6 +89,11 @@ pub(super) struct DownloadSessionState {
     pub(super) peer_advertised_aich_root: bool,
     pub(super) aich_file_hash_requested: bool,
     pub(super) hashset_requested: bool,
+    /// One AICH-set attempt per connection, bound to the accepted generation.
+    /// Keep the attempted generation after consuming the reply: the wire has
+    /// no request ID with which to distinguish a delayed ABA response.
+    pub(super) aich_hashset_requested_generation: Option<u64>,
+    pub(super) aich_hashset_answer_pending: bool,
     pub(super) hashset_requested_at: Option<Instant>,
     pub(super) upload_requested: bool,
     /// The next queue-rank packet is solicited by OP_STARTUPLOADREQ (or by an
@@ -186,6 +191,8 @@ impl DownloadSessionState {
             peer_advertised_aich_root: false,
             aich_file_hash_requested: false,
             hashset_requested: false,
+            aich_hashset_requested_generation: None,
+            aich_hashset_answer_pending: false,
             hashset_requested_at: None,
             upload_requested: false,
             queue_rank_pending: false,

crates/emulebb-ed2k/src/ed2k_tcp/tests/download/hashset_startup.rs

@@ -2,8 +2,58 @@ use super::*;
 
 #[tokio::test]
 async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_arrives() {
+    run_hashset_startup_case(StartupAichCase::Advertised).await;
+}
+
+#[tokio::test]
+async fn consistent_unrelated_peer_aich_tree_cannot_become_download_identity() {
+    run_hashset_startup_case(StartupAichCase::Unrelated).await;
+}
+
+#[tokio::test]
+async fn trusted_operator_master_allows_a_matching_modern_part_set() {
+    run_hashset_startup_case(StartupAichCase::Operator).await;
+}
+
+#[tokio::test]
+async fn trusted_master_allows_aich_only_acquisition_after_md4_is_known() {
+    run_hashset_startup_case(StartupAichCase::OperatorWithMd4).await;
+}
+
+#[tokio::test]
+async fn current_consensus_master_allows_a_matching_modern_part_set() {
+    run_hashset_startup_case(StartupAichCase::NetworkConsensus).await;
+}
+
+#[tokio::test]
+async fn late_modern_aich_answer_cannot_cross_a_consensus_aba_generation() {
+    run_hashset_startup_case(StartupAichCase::ConsensusAba).await;
+}
+
+enum StartupAichCase {
+    Advertised,
+    Unrelated,
+    Operator,
+    OperatorWithMd4,
+    NetworkConsensus,
+    ConsensusAba,
+}
+
+async fn run_hashset_startup_case(case: StartupAichCase) {
+    let unrelated_aich_tree = matches!(case, StartupAichCase::Unrelated);
+    let operator_root = matches!(
+        case,
+        StartupAichCase::Operator | StartupAichCase::OperatorWithMd4
+    );
+    let consensus_root = matches!(
+        case,
+        StartupAichCase::NetworkConsensus | StartupAichCase::ConsensusAba
+    );
+    let late_aich_answer = matches!(case, StartupAichCase::ConsensusAba);
+    let trusted_root = operator_root || consensus_root;
+    let md4_preseeded = matches!(case, StartupAichCase::OperatorWithMd4);
     let root = unique_test_dir("ed2k-large-file-secure-ident-order");
-    let transfer_runtime = Ed2kTransferRuntime::load_or_create(&root).unwrap();
+    let transfer_runtime = Arc::new(Ed2kTransferRuntime::load_or_create(&root).unwrap());
     let payload = vec![0x5A; (ED2K_PART_SIZE as usize) + 32_768];
     let md4_hashset = payload
         .chunks(ED2K_PART_SIZE as usize)
@@ -43,11 +93,51 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar
         .store_piece_data(&file_hash_hex, 1, &payload[ED2K_PART_SIZE as usize..])
         .await
         .unwrap();
-    let source_aich = source_runtime
+    let mut source_aich = source_runtime
         .aich_hashset(&file_hash)
         .await
         .unwrap()
         .expect("missing source AICH hashset");
+    if unrelated_aich_tree {
+        // A mathematically valid two-part tree unrelated to the MD4-identified
+        // payload. It must not authorize salvage or survive local completion.
+        source_aich.part_hashes = vec![[0x21; 20], [0x43; 20]];
+        source_aich.master_hash = sha1::Sha1::digest(
+            source_aich
+                .part_hashes
+                .iter()
+                .flatten()
+                .copied()
+                .collect::<Vec<_>>(),
+        )
+        .into();
+    }
+    let advertised_root = source_aich.master_hash;
+    if operator_root {
+        transfer_runtime
+            .reconcile_aich_root(&file_hash_hex, Some(advertised_root))
+            .await
+            .unwrap();
+    } else if consensus_root {
+        let observations = (0..10)
+            .map(|signer| {
+                (
+                    advertised_root,
+                    IpAddr::V4(std::net::Ipv4Addr::new(203, signer, 0, 1)),
+                )
+            })
+            .collect::<Vec<_>>();
+        transfer_runtime
+            .record_network_aich_observations(&file_hash_hex, &observations)
+            .await
+            .unwrap();
+    }
+    if md4_preseeded {
+        transfer_runtime
+            .store_md4_hashset(&file_hash_hex, md4_hashset.clone())
+            .await
+            .unwrap();
+    }
     let source_identifier = super::Ed2kFileIdentifier {
         file_hash,
         file_size: Some(payload.len() as u64),
@@ -57,6 +147,7 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar
     let listener = TcpListener::bind((test_bind_ip(), 0)).await.unwrap();
     let peer_addr = listener.local_addr().unwrap();
     let payload_for_server = payload.clone();
+    let observed_runtime = Arc::clone(&transfer_runtime);
     let server = tokio::spawn(async move {
         let (mut stream, _) = listener.accept().await.unwrap();
 
@@ -115,21 +206,107 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar
             requested_identifier.file_size,
             Some(payload_for_server.len() as u64)
         );
-        assert!(request_options.request_md4);
-        assert!(request_options.request_aich);
+        assert_eq!(request_options.request_md4, !md4_preseeded);
+        assert_eq!(request_options.request_aich, trusted_root);
+
+        let mut expected_generation = None;
+        if consensus_root {
+            let requested = observed_runtime
+                .manifest(&file_hash.to_string())
+                .await
+                .unwrap();
+            assert_eq!(requested.aich_root, Some(hex::encode(advertised_root)));
+            assert_eq!(
+                requested.aich_root_origin,
+                Some(crate::ed2k_transfer::Ed2kAichRootOrigin::NetworkConsensus)
+            );
+            expected_generation = Some(requested.aich_trust_generation);
+            if late_aich_answer {
+                // WHY: receiving the real request is the causal barrier. Retire
+                // its generation before sending its otherwise matching answer;
+                // returning to the same master prevents a root mismatch from
+                // accidentally testing the wrong response-admission boundary.
+                let mut dissenting_root = advertised_root;
+                dissenting_root[0] ^= 0x80;
+                let demoted = observed_runtime
+                    .record_network_aich_root(
+                        &file_hash.to_string(),
+                        Some(dissenting_root),
+                        IpAddr::V4(std::net::Ipv4Addr::new(198, 0, 0, 1)),
+                    )
+                    .await
+                    .unwrap();
+                assert!(demoted.aich_root.is_none());
+                assert!(demoted.aich_trust_generation > requested.aich_trust_generation);
+                let renewed = observed_runtime
+                    .record_network_aich_observations(
+                        &file_hash.to_string(),
+                        &[10, 11].map(|signer| {
+                            (
+                                advertised_root,
+                                IpAddr::V4(std::net::Ipv4Addr::new(203, signer, 0, 1)),
+                            )
+                        }),
+                    )
+                    .await
+                    .unwrap();
+                assert_eq!(renewed.aich_root, requested.aich_root);
+                assert_eq!(renewed.aich_root_origin, requested.aich_root_origin);
+                assert!(renewed.aich_trust_generation > demoted.aich_trust_generation);
+                assert!(!renewed.aich_hashset_acquired);
+                expected_generation = Some(renewed.aich_trust_generation);
+            }
+        }
 
         let hashset_answer = super::encode_hashset_answer2(
             &source_identifier,
-            Some(&md4_hashset),
+            (!md4_preseeded).then_some(md4_hashset.as_slice()),
             Some(&source_aich),
         )
         .unwrap();
         stream.write_all(&hashset_answer).await.unwrap();
+        // With known MD4, STARTUPLOADREQ can precede processing this AICH-only
+        // answer. A reply to the following frame joins the actual receive path.
+        stream
+            .write_all(&encode_packet(OP_EMULEPROT, OP_PUBLICIP_REQ, &[]))
+            .await
+            .unwrap();
 
         let start_upload = read_packet(&mut stream).await;
         assert_eq!(start_upload[0], OP_EDONKEYPROT);
         assert_eq!(start_upload[5], super::OP_STARTUPLOADREQ);
         assert_eq!(&start_upload[6..22], &file_hash.0);
+        let public_ip_answer = read_packet(&mut stream).await;
+        assert_eq!(public_ip_answer[0], OP_EMULEPROT);
+        assert_eq!(public_ip_answer[5], OP_PUBLICIP_ANSWER);
+        assert_eq!(
+            decode_public_ip_answer_payload(&public_ip_answer[6..]).unwrap(),
+            test_bind_ip()
+        );
+
+        // Observe before supplying any payload: a later local completion pass
+        // must not hide the single-peer acquisition bypass.
+        let before_payload = observed_runtime
+            .manifest(&file_hash.to_string())
+            .await
+            .unwrap();
+        assert!(before_payload.md4_hashset_acquired);
+        if trusted_root {
+            assert_eq!(before_payload.aich_root, Some(hex::encode(advertised_root)));
+            assert_eq!(before_payload.aich_hashset_acquired, !late_aich_answer);
+            assert_eq!(before_payload.aich_hashset.is_empty(), late_aich_answer);
+            if let Some(generation) = expected_generation {
+                assert_eq!(before_payload.aich_trust_generation, generation);
+                assert_eq!(
+                    before_payload.aich_root_origin,
+                    Some(crate::ed2k_transfer::Ed2kAichRootOrigin::NetworkConsensus)
+                );
+            }
+        } else {
+            assert!(before_payload.aich_root.is_none());
+            assert!(!before_payload.aich_hashset_acquired);
+            assert!(before_payload.aich_hashset.is_empty());
+        }
 
         let accept = encode_accept_upload_req();
         stream.write_all(&accept).await.unwrap();
@@ -243,6 +420,13 @@ async fn large_file_download_starts_hashset_flow_before_peer_secure_ident_key_ar
     let manifest = transfer_runtime.manifest(&file_hash_hex).await.unwrap();
     assert!(manifest.completed);
     assert!(manifest.aich_hashset_acquired);
+    assert_eq!(
+        manifest.aich_root_origin,
+        Some(crate::ed2k_transfer::Ed2kAichRootOrigin::LocalComputed)
+    );
     assert_eq!(manifest.aich_hashset.len(), 2);
+    if unrelated_aich_tree {
+        assert_ne!(manifest.aich_root, Some(hex::encode(advertised_root)));
+    }
     server.await.unwrap();
 }

crates/emulebb-ed2k/src/ed2k_tcp/tests/listener/startup.rs

@@ -298,10 +298,17 @@ async fn listener_established_session_tolerates_unknown_opcodes_and_continues()
         .await
         .unwrap();
     let aich_root = [0x7B; 20];
-    transfer_runtime
-        .reconcile_aich_root(&file_hash_hex, Some(aich_root))
+    let proposed = transfer_runtime
+        .record_network_aich_root(
+            &file_hash_hex,
+            Some(aich_root),
+            std::net::IpAddr::from([203, 0, 113, 7]),
+        )
         .await
         .unwrap();
+    assert!(proposed.aich_root.is_none());
+    assert!(proposed.aich_root_origin.is_none());
+    assert!(!proposed.aich_hashset_acquired);
     transfer_runtime
         .store_md4_hashset(&no_sources_hash_hex, Vec::new())
         .await