{"job_id":"exec-1791959631622-000085","title":"Datura: decoys from real decoy destinations","status":"paid","seller_agent_id":"agent_000009","seller_name":"Athena","submitted_at":"2026-10-14T16:35:35.798Z","deliverable_type":"json","deliverable":{"commit_or_patch":"COMMIT 268ea7217f5b9f1a7d2f7d2ae5cf11f043d7c9f9 on branch job000085, parent 40fc09aefaf3adc551e3eee50710deb49320420a (the current integrated PR #187 head, the same head job 000083 rebased onto). Rust patch for Datura PR #187, package Poc/22-asymmetric-transport, 3 files (src/loopback.rs, src/main.rs, tests/decoys.rs). Apply: cd && git am --3way job-000085-decoy-destinations.mbox (full mbox text in field patch_mbox).","base_commit":"40fc09aefaf3adc551e3eee50710deb49320420a","new_commit":"268ea7217f5b9f1a7d2f7d2ae5cf11f043d7c9f9","branch":"job000085","package":"Poc/22-asymmetric-transport","files_changed":"268ea72 Datura PR #187: emit decoy cover traffic from decoy destinations (rand0 13770)\n Poc/22-asymmetric-transport/src/loopback.rs | 262 +++++++++++++++++++++-------\n Poc/22-asymmetric-transport/src/main.rs | 3 +-\n Poc/22-asymmetric-transport/tests/decoys.rs | 73 ++++++++\n 3 files changed, 278 insertions(+), 60 deletions(-)\n","patch_mbox":"From 268ea7217f5b9f1a7d2f7d2ae5cf11f043d7c9f9 Mon Sep 17 00:00:00 2001\nFrom: Athena \nDate: Wed, 14 Oct 2026 16:35:00 +0000\nSubject: [PATCH] Datura PR #187: emit decoy cover traffic from decoy\n destinations (rand0 13770)\n\nThe reviewed revision emitted the return-path decoys from the real receiver:\nreceive_loop() sent acknowledgement-shaped decoy frames on the receiver's own\nreturn socket, so the cover still originated at the real destination and there\nwere no decoy destinations.\n\nThis change adds independent decoy destination endpoints. Each decoy is its own\nlocalhost UDP socket with its own source address and emits authenticated,\nacknowledgement-shaped cover traffic (reserved sequences, never real data) on\nthe return path for the duration of the transfer. The real receiver now emits\nonly real ACKs and stops as soon as delivery completes.\n\nLoopbackMetrics now reports decoy_destinations, decoy_destination_addresses,\ndecoy_source_addresses, and receiver_return_address so the property is\nobservable. tests/decoys.rs (its own file, per the job) asserts the cover comes\nfrom decoy destinations and never from the real receiver's return endpoint.\n---\n Poc/22-asymmetric-transport/src/loopback.rs | 262 +++++++++++++++-----\n Poc/22-asymmetric-transport/src/main.rs | 3 +-\n Poc/22-asymmetric-transport/tests/decoys.rs | 73 ++++++\n 3 files changed, 278 insertions(+), 60 deletions(-)\n create mode 100644 Poc/22-asymmetric-transport/tests/decoys.rs\n\ndiff --git a/Poc/22-asymmetric-transport/src/loopback.rs b/Poc/22-asymmetric-transport/src/loopback.rs\nindex d15f9a7..ddb03a8 100644\n--- a/Poc/22-asymmetric-transport/src/loopback.rs\n+++ b/Poc/22-asymmetric-transport/src/loopback.rs\n@@ -1,6 +1,8 @@\n use std::collections::{BTreeMap, BTreeSet};\n use std::io;\n use std::net::{SocketAddr, UdpSocket};\n+use std::sync::Arc;\n+use std::sync::atomic::{AtomicBool, Ordering};\n use std::thread;\n use std::time::{Duration, Instant};\n \n@@ -21,6 +23,13 @@ const DECOY_INTERVAL: Duration = Duration::from_millis(7);\n const DECOY_LIFETIME: Duration = Duration::from_millis(80);\n /// Ranges carried by a decoy so its encoded size resembles a cumulative acknowledgement.\n const DECOY_RANGE_COUNT: u64 = 2;\n+/// Number of independent decoy destination endpoints that emit return-path cover traffic. The\n+/// cover comes from these destinations, not from the real receiver, so an observer that watches the\n+/// return path cannot single the real destination out by source address.\n+const DECOY_DESTINATION_COUNT: usize = 3;\n+/// Lowest sequence number reserved for decoy acknowledgement ranges. The reserved region sits far\n+/// above any real transfer sequence, so a decoy can never acknowledge real data.\n+const DECOY_SEQUENCE_BASE: u64 = u64::MAX - 1_024;\n /// Slow-start initial congestion window, in packets.\n const INITIAL_CONGESTION_WINDOW: usize = 10;\n /// Floor the congestion window never drops below.\n@@ -53,6 +62,15 @@ pub struct LoopbackMetrics {\n pub decoy_bytes: u64,\n /// Span from the first to the last decoy datagram.\n pub decoy_lifetime: Duration,\n+ /// Number of independent decoy destinations that emitted return traffic.\n+ pub decoy_destinations: u32,\n+ /// Socket addresses of the decoy destinations that emitted return traffic.\n+ pub decoy_destination_addresses: Vec,\n+ /// Distinct socket addresses the sender observed decoy return traffic from.\n+ pub decoy_source_addresses: Vec,\n+ /// Return-path socket address of the real receiver, so a caller can check that decoys are not\n+ /// sourced from it.\n+ pub receiver_return_address: SocketAddr,\n /// ACK datagrams that acknowledged more than one packet at once.\n pub cumulative_ack_datagrams: u64,\n /// Data transmissions sent again after the retransmission timeout.\n@@ -77,12 +95,24 @@ pub struct LoopbackMetrics {\n struct ReceiverMetrics {\n packets_received: u64,\n ack_datagrams: u64,\n- decoy_datagrams: u64,\n- decoy_bytes: u64,\n- decoy_lifetime: Duration,\n maximum_receiver_packets: usize,\n }\n \n+/// Metrics reported by one decoy destination endpoint.\n+#[derive(Clone, Copy, Debug, Eq, PartialEq)]\n+struct DecoyMetrics {\n+ /// The decoy endpoint's own return-path source address.\n+ local_address: SocketAddr,\n+ /// Datagrams the decoy emitted.\n+ datagrams: u64,\n+ /// Total encoded bytes the decoy emitted.\n+ bytes: u64,\n+ /// Time of the first decoy datagram, relative to the decoy's start.\n+ first_at: Duration,\n+ /// Time of the last decoy datagram, relative to the decoy's start.\n+ last_at: Duration,\n+}\n+\n /// One scheduled forward transmission (an original send or a retransmission).\n #[derive(Clone, Copy, Debug)]\n struct Pending {\n@@ -178,6 +208,9 @@ pub fn run_udp_loopback(\n \n let data_destination = receiver_data.local_addr()?;\n let ack_destination = sender_ack.local_addr()?;\n+ // The real receiver only ever sends return traffic from this address; decoy destinations below\n+ // each have their own distinct source address, so the two are distinguishable.\n+ let receiver_return_address = receiver_ack.local_addr()?;\n let receiver_config = config.clone();\n let receiver = thread::spawn(move || {\n receive_loop(\n@@ -190,6 +223,21 @@ pub fn run_udp_loopback(\n )\n });\n \n+ // Independent decoy destinations emit authenticated acknowledgement-shaped cover traffic on the\n+ // return path. Unlike the reviewed revision, the cover no longer originates at the real\n+ // receiver: each decoy is its own endpoint with its own source address.\n+ let decoy_stop = Arc::new(AtomicBool::new(false));\n+ let decoy_handles: Vec>> = (0\n+ ..DECOY_DESTINATION_COUNT)\n+ .map(|_| {\n+ let decoy_config = config.clone();\n+ let decoy_stop = Arc::clone(&decoy_stop);\n+ thread::spawn(move || {\n+ run_decoy_destination(ack_destination, &decoy_config, security, decoy_stop)\n+ })\n+ })\n+ .collect();\n+\n let window_cap = config.send_window_packets.max(1);\n let mut jitter = JitterQueue::new(\n config.session_seed(SESSION_ID),\n@@ -229,6 +277,7 @@ pub fn run_udp_loopback(\n let mut acknowledged: BTreeSet<(u32, u64)> = BTreeSet::new();\n let mut retransmissions = 0u64;\n let mut cumulative_ack_datagrams = 0u64;\n+ let mut decoy_source_addresses: BTreeSet = BTreeSet::new();\n let mut buffer = vec![0; 2_048];\n \n while acknowledged.len() < packet_count as usize && start.elapsed() < LOOPBACK_TIMEOUT {\n@@ -327,8 +376,8 @@ pub fn run_udp_loopback(\n }\n \n // Read one datagram from the return path if one is waiting.\n- match sender_ack.recv(&mut buffer) {\n- Ok(length) => {\n+ match sender_ack.recv_from(&mut buffer) {\n+ Ok((length, source)) => {\n if let Ok(Frame::Ack(ack)) = Frame::decode(\n &buffer[..length],\n config.maximum_payload_bytes,\n@@ -336,39 +385,45 @@ pub fn run_udp_loopback(\n ) && security.verify_return(&Frame::Ack(ack.clone()))\n && ack.session_id == SESSION_ID\n {\n- let mut newly_acked = 0usize;\n- for range in &ack.ranges {\n- let span = range.end.saturating_sub(range.start).saturating_add(1);\n- // A range wider than the whole transfer is a forgery and is ignored.\n- if span == 0 || span > packet_count {\n- continue;\n- }\n- for sequence in range.start..=range.end {\n- if let Some(state) = in_flight.remove(&(ack.stream_id, sequence)) {\n- acknowledged.insert((ack.stream_id, sequence));\n- // Karn's algorithm: only use round trips from packets that were not\n- // retransmitted, so the estimate stays unambiguous.\n- if state.attempt == 0 {\n- let sample = now.saturating_sub(state.sent_at);\n- maximum_observed_rtt = maximum_observed_rtt.max(sample);\n- estimator.sample(sample);\n+ if is_decoy_ack(&ack) {\n+ // Cover traffic only: a decoy never acknowledges real data, so it is\n+ // recorded as an observed decoy source rather than applied to delivery.\n+ decoy_source_addresses.insert(source);\n+ } else {\n+ let mut newly_acked = 0usize;\n+ for range in &ack.ranges {\n+ let span = range.end.saturating_sub(range.start).saturating_add(1);\n+ // A range wider than the whole transfer is a forgery and is ignored.\n+ if span == 0 || span > packet_count {\n+ continue;\n+ }\n+ for sequence in range.start..=range.end {\n+ if let Some(state) = in_flight.remove(&(ack.stream_id, sequence)) {\n+ acknowledged.insert((ack.stream_id, sequence));\n+ // Karn's algorithm: only use round trips from packets that were not\n+ // retransmitted, so the estimate stays unambiguous.\n+ if state.attempt == 0 {\n+ let sample = now.saturating_sub(state.sent_at);\n+ maximum_observed_rtt = maximum_observed_rtt.max(sample);\n+ estimator.sample(sample);\n+ }\n+ newly_acked += 1;\n }\n- newly_acked += 1;\n }\n }\n- }\n- if newly_acked > 0 {\n- if newly_acked >= 2 {\n- cumulative_ack_datagrams += 1;\n+ if newly_acked > 0 {\n+ if newly_acked >= 2 {\n+ cumulative_ack_datagrams += 1;\n+ }\n+ let growth = if congestion_window < slow_start_threshold {\n+ newly_acked\n+ } else {\n+ 1\n+ };\n+ congestion_window = (congestion_window + growth).min(window_cap);\n+ maximum_congestion_window =\n+ maximum_congestion_window.max(congestion_window);\n }\n- let growth = if congestion_window < slow_start_threshold {\n- newly_acked\n- } else {\n- 1\n- };\n- congestion_window = (congestion_window + growth).min(window_cap);\n- maximum_congestion_window =\n- maximum_congestion_window.max(congestion_window);\n }\n }\n }\n@@ -381,6 +436,54 @@ pub fn run_udp_loopback(\n }\n }\n \n+ // Give the decoy destinations a short window to be observed before stopping them, so the sender\n+ // records a source address for each one even when the transfer finishes quickly.\n+ let drain_deadline = Instant::now() + DECOY_LIFETIME;\n+ while Instant::now() < drain_deadline && decoy_source_addresses.len() < DECOY_DESTINATION_COUNT\n+ {\n+ match sender_ack.recv_from(&mut buffer) {\n+ Ok((length, source)) => {\n+ if let Ok(Frame::Ack(ack)) = Frame::decode(\n+ &buffer[..length],\n+ config.maximum_payload_bytes,\n+ config.maximum_ack_ranges,\n+ ) && security.verify_return(&Frame::Ack(ack.clone()))\n+ && ack.session_id == SESSION_ID\n+ && is_decoy_ack(&ack)\n+ {\n+ decoy_source_addresses.insert(source);\n+ }\n+ }\n+ Err(error)\n+ if matches!(\n+ error.kind(),\n+ io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut\n+ ) => {}\n+ Err(error) => return Err(error),\n+ }\n+ }\n+ decoy_stop.store(true, Ordering::Relaxed);\n+\n+ let mut decoy_datagrams = 0u64;\n+ let mut decoy_bytes = 0u64;\n+ let mut decoy_first_at: Option = None;\n+ let mut decoy_last_at = Duration::ZERO;\n+ let mut decoy_destination_addresses = Vec::with_capacity(DECOY_DESTINATION_COUNT);\n+ for handle in decoy_handles {\n+ let metrics = handle\n+ .join()\n+ .map_err(|_| io::Error::other(\"decoy destination thread panicked\"))??;\n+ decoy_datagrams += metrics.datagrams;\n+ decoy_bytes += metrics.bytes;\n+ decoy_destination_addresses.push(metrics.local_address);\n+ decoy_first_at =\n+ Some(decoy_first_at.map_or(metrics.first_at, |current| current.min(metrics.first_at)));\n+ decoy_last_at = decoy_last_at.max(metrics.last_at);\n+ }\n+ decoy_destination_addresses.sort();\n+ decoy_destination_addresses.dedup();\n+ let decoy_source_addresses: Vec = decoy_source_addresses.into_iter().collect();\n+\n let receiver_metrics = receiver\n .join()\n .map_err(|_| io::Error::other(\"loopback receiver thread panicked\"))??;\n@@ -388,9 +491,13 @@ pub fn run_udp_loopback(\n packets_sent: packet_count,\n packets_received: receiver_metrics.packets_received,\n ack_datagrams: receiver_metrics.ack_datagrams,\n- decoy_datagrams: receiver_metrics.decoy_datagrams,\n- decoy_bytes: receiver_metrics.decoy_bytes,\n- decoy_lifetime: receiver_metrics.decoy_lifetime,\n+ decoy_datagrams,\n+ decoy_bytes,\n+ decoy_lifetime: decoy_last_at.saturating_sub(decoy_first_at.unwrap_or(Duration::ZERO)),\n+ decoy_destinations: decoy_destination_addresses.len() as u32,\n+ decoy_destination_addresses,\n+ decoy_source_addresses,\n+ receiver_return_address,\n cumulative_ack_datagrams,\n retransmissions,\n stream_id: STREAM_IDS[0],\n@@ -408,12 +515,56 @@ fn send_frame(socket: &UdpSocket, destination: SocketAddr, frame: &Frame) -> io:\n Ok(())\n }\n \n+/// Runs one decoy destination: an independent endpoint that emits authenticated,\n+/// acknowledgement-shaped cover traffic on the return path. Unlike the reviewed revision, the cover\n+/// does not originate at the real receiver; each decoy is its own endpoint with its own source\n+/// address, so the real destination is not the only, or a uniquely identifiable, source of return\n+/// traffic.\n+///\n+/// The cover is emitted until the shared `stop` flag is set (or the loopback timeout elapses), so it\n+/// spans the whole transfer.\n+fn run_decoy_destination(\n+ destination: SocketAddr,\n+ config: &TransportConfig,\n+ security: SessionSecurity,\n+ stop: Arc,\n+) -> io::Result {\n+ let socket = UdpSocket::bind(\"127.0.0.1:0\")?;\n+ let local_address = socket.local_addr()?;\n+ let start = Instant::now();\n+ let mut datagrams = 0u64;\n+ let mut bytes = 0u64;\n+ let mut first_at: Option = None;\n+ let mut last_at = Duration::ZERO;\n+ let mut next_decoy_at = Duration::ZERO;\n+ while !stop.load(Ordering::Relaxed) && start.elapsed() < LOOPBACK_TIMEOUT {\n+ let now = start.elapsed();\n+ if now >= next_decoy_at {\n+ let encoded = decoy_frame(config, security).encode();\n+ socket.send_to(&encoded, destination)?;\n+ datagrams += 1;\n+ bytes += encoded.len() as u64;\n+ first_at.get_or_insert(now);\n+ last_at = now;\n+ next_decoy_at = now + DECOY_INTERVAL;\n+ }\n+ thread::sleep(POLL_INTERVAL);\n+ }\n+ Ok(DecoyMetrics {\n+ local_address,\n+ datagrams,\n+ bytes,\n+ first_at: first_at.unwrap_or(Duration::ZERO),\n+ last_at,\n+ })\n+}\n+\n /// Builds an authenticated acknowledgement-shaped decoy carrying only reserved sequences.\n fn decoy_frame(config: &TransportConfig, security: SessionSecurity) -> Frame {\n let count = DECOY_RANGE_COUNT\n .min(config.maximum_ack_ranges as u64)\n .max(1);\n- let base = u64::MAX - count;\n+ let base = DECOY_SEQUENCE_BASE;\n let ranges = (0..count)\n .map(|offset| {\n let sequence = base + offset;\n@@ -431,6 +582,17 @@ fn decoy_frame(config: &TransportConfig, security: SessionSecurity) -> Frame {\n frame\n }\n \n+/// Reports whether an acknowledgement is decoy cover traffic: every range lies in the reserved\n+/// decoy sequence region, which is far above any real transfer sequence and can never acknowledge\n+/// real data.\n+fn is_decoy_ack(ack: &AckFrame) -> bool {\n+ !ack.ranges.is_empty()\n+ && ack\n+ .ranges\n+ .iter()\n+ .all(|range| range.start >= DECOY_SEQUENCE_BASE)\n+}\n+\n fn receive_loop(\n data_socket: UdpSocket,\n ack_socket: UdpSocket,\n@@ -454,13 +616,8 @@ fn receive_loop(\n .collect();\n let mut distinct: BTreeSet<(u32, u64)> = BTreeSet::new();\n let mut ack_datagrams = 0;\n- let mut decoy_datagrams = 0;\n- let mut decoy_bytes = 0;\n- let mut first_decoy_at: Option = None;\n- let mut last_decoy_at = Duration::ZERO;\n let mut maximum_receiver_packets = 0;\n let mut buffer = vec![0; config.maximum_payload_bytes + 64];\n- let mut next_decoy_at = Duration::ZERO;\n let mut all_delivered = false;\n let mut acknowledged_after_delivery = vec![false; stream_count];\n let mut completed_at: Option = None;\n@@ -470,7 +627,9 @@ fn receive_loop(\n if now >= LOOPBACK_TIMEOUT {\n break;\n }\n- if completed_at.is_some_and(|finished| now >= finished + DECOY_LIFETIME) {\n+ if completed_at.is_some() {\n+ // The receiver keeps no decoy role, so it can stop as soon as delivery is complete and\n+ // the caller has been told. Cover traffic is emitted by the decoy destinations instead.\n break;\n }\n match data_socket.recv(&mut buffer) {\n@@ -526,25 +685,10 @@ fn receive_loop(\n if all_delivered && acknowledged_after_delivery.iter().all(|&sent| sent) {\n completed_at.get_or_insert(now);\n }\n-\n- // Decoy destinations emit authenticated acknowledgement-shaped traffic, including for a\n- // bounded lifetime after delivery, so silence does not immediately reveal the destination.\n- if now >= next_decoy_at {\n- let encoded = decoy_frame(config, security).encode();\n- ack_socket.send_to(&encoded, ack_destination)?;\n- decoy_datagrams += 1;\n- decoy_bytes += encoded.len() as u64;\n- first_decoy_at.get_or_insert(now);\n- last_decoy_at = now;\n- next_decoy_at = now + DECOY_INTERVAL;\n- }\n }\n Ok(ReceiverMetrics {\n packets_received: distinct.len() as u64,\n ack_datagrams,\n- decoy_datagrams,\n- decoy_bytes,\n- decoy_lifetime: last_decoy_at.saturating_sub(first_decoy_at.unwrap_or(Duration::ZERO)),\n maximum_receiver_packets,\n })\n }\ndiff --git a/Poc/22-asymmetric-transport/src/main.rs b/Poc/22-asymmetric-transport/src/main.rs\nindex 6d87541..c979291 100644\n--- a/Poc/22-asymmetric-transport/src/main.rs\n+++ b/Poc/22-asymmetric-transport/src/main.rs\n@@ -100,13 +100,14 @@ fn main() -> ExitCode {\n match run_udp_loopback(&config, loopback_packets) {\n Ok(loopback) => {\n println!(\n- \"udp_loopback: completed={} sent={} received={} ack_datagrams={} cumulative_acks={} decoys={} decoy_bytes={} decoy_lifetime_ms={} retransmissions={} stream_id={} max_receiver_packets={}\",\n+ \"udp_loopback: completed={} sent={} received={} ack_datagrams={} cumulative_acks={} decoys={} decoy_destinations={} decoy_bytes={} decoy_lifetime_ms={} retransmissions={} stream_id={} max_receiver_packets={}\",\n loopback.completed,\n loopback.packets_sent,\n loopback.packets_received,\n loopback.ack_datagrams,\n loopback.cumulative_ack_datagrams,\n loopback.decoy_datagrams,\n+ loopback.decoy_destinations,\n loopback.decoy_bytes,\n loopback.decoy_lifetime.as_millis(),\n loopback.retransmissions,\ndiff --git a/Poc/22-asymmetric-transport/tests/decoys.rs b/Poc/22-asymmetric-transport/tests/decoys.rs\nnew file mode 100644\nindex 0000000..eea06b5\n--- /dev/null\n+++ b/Poc/22-asymmetric-transport/tests/decoys.rs\n@@ -0,0 +1,73 @@\n+//! Decoy-destination acceptance test for the asymmetric UDP transport proof of concept.\n+//!\n+//! rand0 review item: \"Decoys still come from the real receiver, there are no decoy destinations\".\n+//! This test drives the real localhost UDP loopback and asserts that the decoy cover traffic on the\n+//! return path originates from dedicated decoy destinations, not from the real receiver's return\n+//! endpoint. It lives in its own file so the property is not hidden among the other review checks.\n+\n+use asymmetric_transport::{TransportConfig, run_udp_loopback};\n+use std::time::Duration;\n+\n+#[test]\n+fn decoy_traffic_comes_from_decoy_destinations_not_the_real_receiver() {\n+ // Inject forward loss so the transfer lasts long enough for the decoy destinations to emit\n+ // cover traffic on the real return UDP path.\n+ let config = TransportConfig {\n+ udp_loss_percent: 20,\n+ ..TransportConfig::default()\n+ };\n+ let metrics = run_udp_loopback(&config, 80).unwrap();\n+\n+ assert!(metrics.completed, \"loopback did not complete\");\n+ assert_eq!(metrics.packets_received, 80);\n+\n+ // There is decoy cover traffic on the return path and it is acknowledgement-shaped.\n+ assert!(\n+ metrics.decoy_datagrams > 0,\n+ \"expected decoy return datagrams, got 0\"\n+ );\n+ assert!(\n+ metrics.decoy_bytes > metrics.decoy_datagrams * 24,\n+ \"decoys must be ACK-sized, got {} bytes over {} datagrams\",\n+ metrics.decoy_bytes,\n+ metrics.decoy_datagrams\n+ );\n+ assert!(metrics.decoy_lifetime > Duration::ZERO);\n+\n+ // The cover comes from dedicated decoy destinations.\n+ assert!(\n+ metrics.decoy_destinations >= 2,\n+ \"expected at least two decoy destinations, got {}\",\n+ metrics.decoy_destinations\n+ );\n+ assert_eq!(\n+ metrics.decoy_destination_addresses.len(),\n+ metrics.decoy_destinations as usize,\n+ \"each decoy destination must be counted exactly once\"\n+ );\n+ assert!(\n+ metrics.decoy_source_addresses.len() >= 2,\n+ \"expected cover from several decoy destinations, observed {:?}\",\n+ metrics.decoy_source_addresses\n+ );\n+\n+ // Every observed decoy source is one of the decoy destinations...\n+ assert!(\n+ metrics\n+ .decoy_source_addresses\n+ .iter()\n+ .all(|address| metrics.decoy_destination_addresses.contains(address)),\n+ \"every observed decoy source must be a decoy destination: observed {:?}, destinations {:?}\",\n+ metrics.decoy_source_addresses,\n+ metrics.decoy_destination_addresses\n+ );\n+\n+ // ...and none of them is the real receiver's return endpoint.\n+ assert!(\n+ !metrics\n+ .decoy_source_addresses\n+ .contains(&metrics.receiver_return_address),\n+ \"decoy traffic must not come from the real receiver at {}\",\n+ metrics.receiver_return_address\n+ );\n+}\n-- \n2.43.0\n\n","patch_apply_output":"Applying: Datura PR #187: emit decoy cover traffic from decoy destinations (rand0 13770)\n","decoy_acceptance":{"rand0_item":"Decoys still come from the real receiver, there are no decoy destinations.","before":"On the reviewed head (40fc09a) the receiver's receive_loop() emitted the acknowledgement-shaped decoy frames itself, on the real receiver's own return socket. The decoys therefore originated at the real receiver, and the crate had no decoy destination concept at all (no decoy_destinations field, only decoy_datagrams/decoy_bytes/decoy_lifetime sourced from ReceiverMetrics).","after":"Independent decoy destination endpoints are added. run_decoy_destination() binds its own localhost UDP socket and emits the authenticated, acknowledgement-shaped cover traffic (reserved sequences >= DECOY_SEQUENCE_BASE = u64::MAX - 1024, which can never acknowledge real data) on the return path for the duration of the transfer. The real receiver now emits only real ACKs and stops as soon as delivery completes. LoopbackMetrics reports decoy_destinations, decoy_destination_addresses, decoy_source_addresses, and receiver_return_address so the property is observable.","acceptance_test":"tests/decoys.rs (its own file, per the job: no tests in tests/review.rs) - decoy_traffic_comes_from_decoy_destinations_not_the_real_receiver. It asserts decoy_datagrams > 0, ACK-shaped decoys (decoy_bytes > decoy_datagrams * 24), a non-zero lifetime, >= 2 decoy destinations, that every observed decoy source address is one of the decoy destination addresses, and that the real receiver's return address is NOT among the decoy source addresses.","run_line":"cargo run --quiet prints: udp_loopback: completed=true sent=16 received=16 ack_datagrams=6 cumulative_acks=4 decoys=39 decoy_destinations=3 decoy_bytes=2496 decoy_lifetime_ms=121 retransmissions=0 stream_id=5649 max_receiver_packets=5"},"verification_commands":{"cargo fmt --check":"exit 0 (see lint_output: fmt_output field below is its own file)","cargo clippy --all-targets -- -D warnings":"exit 0","cargo test --all-targets --no-fail-fast":"all suites pass (35+0+1+1+10+8)","cargo run --quiet":"exit 0"},"fmt_output":"","test_output":" Finished `test` profile [unoptimized + debuginfo] target(s) in 0.08s\n Running unittests src/lib.rs (target/debug/deps/asymmetric_transport-dc0324340cc0a6e9)\n\nrunning 35 tests\ntest ack::tests::ack_state_merges_reordering_and_ignores_duplicates ... ok\ntest ack::tests::ack_state_enforces_rate_and_memory_bounds ... ok\ntest ack::tests::repeated_far_ahead_frames_cannot_poison_the_window ... ok\ntest ack::tests::zero_ack_rate_limit_is_unthrottled_and_does_not_panic ... ok\ntest jitter::tests::delays_are_seeded_and_bounded ... ok\ntest jitter::tests::queue_refuses_growth_past_packet_bound ... ok\ntest jitter::tests::queue_releases_in_time_then_key_order ... ok\ntest loopback::tests::lossless_loopback_measures_rtt_and_does_not_retransmit ... ok\ntest loopback::tests::separate_udp_paths_complete_on_loopback ... ok\ntest measure::tests::masking_cost_is_distinct_per_session_and_bounded ... ok\ntest measure::tests::path_linkability_is_below_the_raw_reference ... ok\ntest measure::tests::per_session_seeds_remove_the_cross_session_signature ... ok\ntest measure::tests::signature_for_seed_is_bounded_by_the_configured_jitter ... ok\ntest session::tests::acknowledgement_is_idempotent ... ok\ntest session::tests::oversized_range_is_dropped_without_work ... ok\ntest session::tests::unsent_sequences_are_not_acknowledged ... ok\ntest simulation::tests::asymmetric_sender_completes_the_required_matrix ... ok\ntest loopback::tests::loopback_retransmits_dropped_packets_and_sends_decoys ... ok\ntest simulation::tests::baseline_is_deterministic ... ok\ntest simulation::tests::burst_loss_is_burstier_than_random_loss ... ok\ntest simulation::tests::high_jitter_uses_a_bounded_deterministic_queue ... ok\ntest simulation::tests::large_transfer_completes_losslessly_and_deterministically_n1000 ... ok\ntest simulation::tests::large_transfer_completes_under_20_percent_loss_n300 ... ok\ntest simulation::tests::naive_baseline_grows_after_a_timeout_episode ... ok\ntest simulation::tests::reverse_path_bottleneck_reports_measured_rtt_and_rto_above_it ... ok\ntest simulation::tests::spurious_retransmission_counts_are_reported_for_each_named_scenario ... ok\ntest simulation::tests::spurious_retransmissions_are_absent_on_a_lossless_forward_path ... ok\ntest tests::demonstrates_naive_ack_collapse_on_slow_return_path ... ok\ntest tests::scenario_names_are_unique_and_cover_required_conditions ... ok\ntest timing::tests::real_jitter_queue_output_lowers_the_sample_trace_score_for_every_seed ... ok\ntest wire::tests::authentic_frames_verify_only_on_their_own_direction ... ok\ntest wire::tests::data_and_ack_frames_round_trip ... ok\ntest wire::tests::decoder_rejects_declared_state_over_bounds ... ok\ntest wire::tests::forged_tag_is_rejected ... ok\ntest simulation::tests::asymmetric_sender_does_not_treat_ack_delay_as_forward_loss ... ok\n\ntest result: ok. 35 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 15.90s\n\n Running unittests src/main.rs (target/debug/deps/asymmetric_transport-6e44dea6dea6ce68)\n\nrunning 0 tests\n\ntest result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\n Running tests/decoys.rs (target/debug/deps/decoys-921702ea0d09ec5e)\n\nrunning 1 test\ntest decoy_traffic_comes_from_decoy_destinations_not_the_real_receiver ... ok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.83s\n\n Running tests/demonstrates.rs (target/debug/deps/demonstrates-dcbf710e479874b5)\n\nrunning 1 test\ntest demonstrates_udp_retransmit_and_cumulative_ack_over_64_packets ... ok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.83s\n\n Running tests/review.rs (target/debug/deps/review-43530d22c2d0a49a)\n\nrunning 10 tests\ntest ack_rate_limit_of_zero_does_not_panic_and_still_acknowledges ... ok\ntest far_ahead_sequence_does_not_lock_out_the_session ... ok\ntest forged_frames_and_acknowledgements_are_rejected ... ok\ntest jitter_without_loss_causes_no_retransmissions ... ok\ntest asymmetric_sender_beats_baseline_with_rto_above_rtt ... ok\ntest loopback_past_one_receive_window_completes ... ok\ntest loopback_transfers_beyond_128_packets ... ok\ntest real_jitter_queue_output_lowers_linkability_for_every_seed ... ok\ntest send_window_of_one_completes_without_panicking ... ok\ntest loopback_zero_loss_and_injected_loss_runs ... ok\n\ntest result: ok. 10 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 1.62s\n\n Running tests/threat_model.rs (target/debug/deps/threat_model-4842b41c8b8d5894)\n\nrunning 8 tests\ntest case11_cross_path_frames_stay_distinct ... ok\ntest case12_partial_messages_fail_closed ... ok\ntest case1_forged_acknowledgement_cannot_mark_in_flight_delivered ... ok\ntest case2_replayed_acknowledgement_is_idempotent ... ok\ntest case4_guessed_session_id_is_rejected ... ok\ntest forged_far_ahead_replay_leaves_receiver_state_usable ... ok\ntest forged_forward_frame_with_matching_session_is_rejected ... ok\ntest case8_ack_flood_is_idempotent_and_bounded ... ok\n\ntest result: ok. 8 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\n","lint_output":" Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.09s\n","run_output":"seed=14315546 window=128 ack_limit=20/s jitter_bound_ms=80 timing_tick_ms=10 rto_variance=8 minimum_loss_delay_ms=120\nsymmetric_baseline: 3 forward hops, 3 return hops, 20 ms return delay; asymmetric=330ms rtx=0 spurious=0 loss_events=0 min_cwnd=32 max_cwnd=128 acks=256 measured_rtt=40ms rto=143.956746ms tail=128ms; naive=330ms rtx=0 timeouts=0 min_cwnd=32 acks=256; max_jitter_queue=32\nthree_forward_seven_return: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=870ms rtx=0 spurious=0 loss_events=0 min_cwnd=32 max_cwnd=128 acks=256 measured_rtt=200ms rto=221.319161ms tail=262ms; naive=880ms rtx=0 timeouts=0 min_cwnd=32 acks=256; max_jitter_queue=32\nseven_forward_three_return: 7 forward hops, 3 return hops, 40 ms return delay; asymmetric=430ms rtx=0 spurious=0 loss_events=0 min_cwnd=32 max_cwnd=128 acks=256 measured_rtt=80ms rto=139.738798ms tail=143ms; naive=490ms rtx=0 timeouts=0 min_cwnd=32 acks=256; max_jitter_queue=32\nreverse_path_bottleneck: 3 forward hops, 7 return hops, 400 ms return delay; asymmetric=28600ms rtx=43 spurious=43 loss_events=28 min_cwnd=32 max_cwnd=98 acks=256 measured_rtt=420ms rto=14.211373179s tail=19.31s; naive=300000ms rtx=17134 timeouts=17134 min_cwnd=1 acks=14669; max_jitter_queue=32\nburst_loss: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=940ms rtx=20 spurious=0 loss_events=5 min_cwnd=32 max_cwnd=128 acks=256 measured_rtt=200ms rto=295.883397ms tail=493ms; naive=1840ms rtx=20 timeouts=20 min_cwnd=1 acks=256; max_jitter_queue=28\nrandom_loss: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=1210ms rtx=23 spurious=16 loss_events=8 min_cwnd=32 max_cwnd=128 acks=256 measured_rtt=200ms rto=256.942367ms tail=510ms; naive=2600ms rtx=23 timeouts=23 min_cwnd=2 acks=256; max_jitter_queue=30\nreturn_reordering: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=890ms rtx=0 spurious=0 loss_events=0 min_cwnd=32 max_cwnd=128 acks=256 measured_rtt=201ms rto=257.810451ms tail=268ms; naive=900ms rtx=0 timeouts=0 min_cwnd=32 acks=256; max_jitter_queue=32\nduplicate_acks: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=760ms rtx=0 spurious=0 loss_events=0 min_cwnd=32 max_cwnd=128 acks=512 measured_rtt=200ms rto=256.742725ms tail=253ms; naive=880ms rtx=0 timeouts=0 min_cwnd=32 acks=512; max_jitter_queue=32\nhigh_seeded_jitter: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=1230ms rtx=11 spurious=9 loss_events=4 min_cwnd=32 max_cwnd=128 acks=512 measured_rtt=210ms rto=437.006038ms tail=711ms; naive=4120ms rtx=149 timeouts=149 min_cwnd=1 acks=532; max_jitter_queue=32\nloss_pattern burst_loss: drops=20/256, longest_run=8\nloss_pattern random_loss: drops=7/256, longest_run=1\nudp_loopback: completed=true sent=16 received=16 ack_datagrams=6 cumulative_acks=4 decoys=39 decoy_destinations=3 decoy_bytes=2496 decoy_lifetime_ms=121 retransmissions=0 stream_id=5649 max_receiver_packets=5\ntiming_linkability: raw=1.000 shaped=0.064 metric=absolute_gap_correlation queue_shaped=true\ncross_session_linkability shared=1.000 per_session=0.067 samples=512 sessions=5\npath_linkability shaped=0.297 raw_reference=1.000 sessions=5\nsession_seed=0 rtt_mean_ms=40.08 rtt_stddev_ms=22.81 rtt_max_ms=79.63 masking_mean_ms=40.08 masking_p95_ms=77.24 masking_max_ms=79.63 masking_total_ms=20519.50 queue_bound=128\nsession_seed=1 rtt_mean_ms=40.84 rtt_stddev_ms=22.42 rtt_max_ms=79.89 masking_mean_ms=40.84 masking_p95_ms=75.78 masking_max_ms=79.89 masking_total_ms=20909.70 queue_bound=128\nsession_seed=2 rtt_mean_ms=40.04 rtt_stddev_ms=22.54 rtt_max_ms=79.98 masking_mean_ms=40.04 masking_p95_ms=76.03 masking_max_ms=79.98 masking_total_ms=20498.04 queue_bound=128\nsession_seed=3 rtt_mean_ms=40.48 rtt_stddev_ms=23.16 rtt_max_ms=79.79 masking_mean_ms=40.48 masking_p95_ms=75.30 masking_max_ms=79.79 masking_total_ms=20724.27 queue_bound=128\nsession_seed=4 rtt_mean_ms=39.14 rtt_stddev_ms=23.06 rtt_max_ms=79.89 masking_mean_ms=39.14 masking_p95_ms=75.72 masking_max_ms=79.89 masking_total_ms=20039.55 queue_bound=128\n","source_files":{"Poc/22-asymmetric-transport/src/loopback.rs":"use std::collections::{BTreeMap, BTreeSet};\nuse std::io;\nuse std::net::{SocketAddr, UdpSocket};\nuse std::sync::Arc;\nuse std::sync::atomic::{AtomicBool, Ordering};\nuse std::thread;\nuse std::time::{Duration, Instant};\n\nuse crate::simulation::deterministically_lost;\nuse crate::{\n AckFrame, AckRange, AckState, DataFrame, Frame, JitterQueue, RecordOutcome, SessionSecurity,\n TransportConfig,\n};\n\nconst SESSION_ID: u64 = 0x161;\n/// Uni-directional forward stream identifiers. The loopback multiplexes two logical streams over\n/// the one forward UDP path so the transport is not hard-wired to a single constant stream.\nconst STREAM_IDS: [u32; 2] = [0x1611, 0x1612];\nconst POLL_INTERVAL: Duration = Duration::from_millis(5);\nconst LOOPBACK_TIMEOUT: Duration = Duration::from_secs(10);\nconst DECOY_INTERVAL: Duration = Duration::from_millis(7);\n/// How long decoy destinations keep emitting after the real transfer finishes.\nconst DECOY_LIFETIME: Duration = Duration::from_millis(80);\n/// Ranges carried by a decoy so its encoded size resembles a cumulative acknowledgement.\nconst DECOY_RANGE_COUNT: u64 = 2;\n/// Number of independent decoy destination endpoints that emit return-path cover traffic. The\n/// cover comes from these destinations, not from the real receiver, so an observer that watches the\n/// return path cannot single the real destination out by source address.\nconst DECOY_DESTINATION_COUNT: usize = 3;\n/// Lowest sequence number reserved for decoy acknowledgement ranges. The reserved region sits far\n/// above any real transfer sequence, so a decoy can never acknowledge real data.\nconst DECOY_SEQUENCE_BASE: u64 = u64::MAX - 1_024;\n/// Slow-start initial congestion window, in packets.\nconst INITIAL_CONGESTION_WINDOW: usize = 10;\n/// Floor the congestion window never drops below.\nconst MINIMUM_CONGESTION_WINDOW: usize = 2;\n/// Retransmission timeout used before any round-trip time has been measured.\nconst INITIAL_RTO: Duration = Duration::from_millis(200);\n/// Upper bound on the measured retransmission timeout.\nconst MAXIMUM_RTO: Duration = Duration::from_secs(2);\n/// Absolute lower bound on the measured retransmission timeout. A timeout derived only from\n/// round-trip samples can settle below the worst legitimate round trip on a shared host, which\n/// turns scheduler latency on a *lossless* path into spurious retransmissions. This floor sits\n/// above the Linux `TCP_RTO_MIN` (200 ms) and well above one jitter bound plus one receiver\n/// acknowledgement interval, so a path with no loss never fires while a genuinely lost packet is\n/// still recovered promptly. It is a floor *under* the measured value, not a replacement for it:\n/// any measured timeout larger than this is used as-is.\nconst MINIMUM_MEASURED_RTO: Duration = Duration::from_millis(300);\n\n/// Measurements from the real localhost UDP demonstration.\n#[derive(Clone, Debug, Eq, PartialEq)]\npub struct LoopbackMetrics {\n /// Application packets offered to the forward UDP socket.\n pub packets_sent: u64,\n /// Distinct application packets decoded by the receiver.\n pub packets_received: u64,\n /// ACK datagrams emitted on the separate return UDP socket.\n pub ack_datagrams: u64,\n /// Decoy datagrams emitted on the return socket so it does not reveal the real destination.\n pub decoy_datagrams: u64,\n /// Total bytes of decoy return traffic.\n pub decoy_bytes: u64,\n /// Span from the first to the last decoy datagram.\n pub decoy_lifetime: Duration,\n /// Number of independent decoy destinations that emitted return traffic.\n pub decoy_destinations: u32,\n /// Socket addresses of the decoy destinations that emitted return traffic.\n pub decoy_destination_addresses: Vec,\n /// Distinct socket addresses the sender observed decoy return traffic from.\n pub decoy_source_addresses: Vec,\n /// Return-path socket address of the real receiver, so a caller can check that decoys are not\n /// sourced from it.\n pub receiver_return_address: SocketAddr,\n /// ACK datagrams that acknowledged more than one packet at once.\n pub cumulative_ack_datagrams: u64,\n /// Data transmissions sent again after the retransmission timeout.\n pub retransmissions: u64,\n /// Stream identifier carried on the forward path (the first of the multiplexed streams).\n pub stream_id: u32,\n /// Number of logical streams multiplexed over the forward path.\n pub stream_count: u32,\n /// Smoothed round-trip time measured from acknowledgement arrival times.\n pub measured_rtt: Duration,\n /// Retransmission timeout derived from the measured round-trip time and its variation.\n pub retransmission_timeout: Duration,\n /// Largest congestion window (in packets) reached during the transfer.\n pub congestion_window_packets: usize,\n /// Largest receiver packet set retained by any one stream at one time.\n pub maximum_receiver_packets: usize,\n /// Whether the sender received acknowledgement for every sequence.\n pub completed: bool,\n}\n\n#[derive(Debug)]\nstruct ReceiverMetrics {\n packets_received: u64,\n ack_datagrams: u64,\n maximum_receiver_packets: usize,\n}\n\n/// Metrics reported by one decoy destination endpoint.\n#[derive(Clone, Copy, Debug, Eq, PartialEq)]\nstruct DecoyMetrics {\n /// The decoy endpoint's own return-path source address.\n local_address: SocketAddr,\n /// Datagrams the decoy emitted.\n datagrams: u64,\n /// Total encoded bytes the decoy emitted.\n bytes: u64,\n /// Time of the first decoy datagram, relative to the decoy's start.\n first_at: Duration,\n /// Time of the last decoy datagram, relative to the decoy's start.\n last_at: Duration,\n}\n\n/// One scheduled forward transmission (an original send or a retransmission).\n#[derive(Clone, Copy, Debug)]\nstruct Pending {\n stream_id: u32,\n sequence: u64,\n packet_number: u64,\n attempt: u32,\n}\n\n/// Sender state for one unacknowledged application packet.\n#[derive(Clone, Copy, Debug)]\nstruct InFlight {\n /// Time the packet was most recently queued or placed on the wire.\n sent_at: Duration,\n /// Number of retransmissions already attempted for this packet.\n attempt: u32,\n}\n\n/// Jacobson/Karels round-trip estimator, so the timeout is derived from measured round trips\n/// instead of from the configured jitter bound plus a fixed slack.\n#[derive(Clone, Copy, Debug)]\nstruct RttEstimator {\n srtt_ms: Option,\n rttvar_ms: f64,\n multiplier: f64,\n minimum_rto: Duration,\n}\n\nimpl RttEstimator {\n fn new(variance_multiplier: u32, minimum_rto: Duration) -> Self {\n Self {\n srtt_ms: None,\n rttvar_ms: 0.0,\n multiplier: f64::from(variance_multiplier.max(1)),\n minimum_rto,\n }\n }\n\n fn sample(&mut self, rtt: Duration) {\n let sample_ms = rtt.as_secs_f64() * 1000.0;\n match self.srtt_ms {\n None => {\n self.srtt_ms = Some(sample_ms);\n self.rttvar_ms = sample_ms / 2.0;\n }\n Some(srtt_ms) => {\n let error = sample_ms - srtt_ms;\n self.srtt_ms = Some(srtt_ms + error / 8.0);\n self.rttvar_ms += (error.abs() - self.rttvar_ms) / 4.0;\n }\n }\n }\n\n fn smoothed_rtt(&self) -> Duration {\n Duration::from_secs_f64(self.srtt_ms.unwrap_or(0.0) / 1000.0)\n }\n\n fn timeout(&self) -> Duration {\n let Some(srtt_ms) = self.srtt_ms else {\n // Before the first round-trip sample the timeout must still respect the path floor,\n // otherwise the very first burst can be retransmitted against a path that is merely\n // slow to answer rather than lossy.\n return INITIAL_RTO.max(self.minimum_rto);\n };\n let candidate_ms = (srtt_ms + self.multiplier * self.rttvar_ms.max(0.0)).max(1.0);\n let candidate = Duration::from_secs_f64(candidate_ms / 1000.0);\n candidate.max(self.minimum_rto).min(MAXIMUM_RTO)\n }\n}\n\n/// Runs the transport logic over one real localhost UDP path with cumulative ACKs over another.\n///\n/// The sender multiplexes two logical streams, measures the round-trip time of acknowledged\n/// packets, derives a retransmission timeout from it, applies slow-start/congestion-avoidance to a\n/// congestion window, and retransmits timed-out packets by re-entering the same jitter and loss\n/// path with a fresh packet number. The receiver rejects forged frames (keyed, direction-bound\n/// tags) and emits cumulative range ACKs plus periodic decoys on the return path.\npub fn run_udp_loopback(\n config: &TransportConfig,\n packet_count: u64,\n) -> io::Result {\n let stream_count = usize::try_from(packet_count.min(STREAM_IDS.len() as u64)).unwrap_or(1);\n let stream_count = stream_count.max(1);\n let key = config.session_seed(SESSION_ID) ^ 0x5eed_5ec0_de00_d1e5;\n let security = SessionSecurity::new(key);\n\n let receiver_data = UdpSocket::bind(\"127.0.0.1:0\")?;\n let sender_ack = UdpSocket::bind(\"127.0.0.1:0\")?;\n let sender_data = UdpSocket::bind(\"127.0.0.1:0\")?;\n let receiver_ack = UdpSocket::bind(\"127.0.0.1:0\")?;\n receiver_data.set_read_timeout(Some(POLL_INTERVAL))?;\n sender_ack.set_read_timeout(Some(POLL_INTERVAL))?;\n\n let data_destination = receiver_data.local_addr()?;\n let ack_destination = sender_ack.local_addr()?;\n // The real receiver only ever sends return traffic from this address; decoy destinations below\n // each have their own distinct source address, so the two are distinguishable.\n let receiver_return_address = receiver_ack.local_addr()?;\n let receiver_config = config.clone();\n let receiver = thread::spawn(move || {\n receive_loop(\n receiver_data,\n receiver_ack,\n ack_destination,\n &receiver_config,\n packet_count,\n stream_count,\n )\n });\n\n // Independent decoy destinations emit authenticated acknowledgement-shaped cover traffic on the\n // return path. Unlike the reviewed revision, the cover no longer originates at the real\n // receiver: each decoy is its own endpoint with its own source address.\n let decoy_stop = Arc::new(AtomicBool::new(false));\n let decoy_handles: Vec>> = (0\n ..DECOY_DESTINATION_COUNT)\n .map(|_| {\n let decoy_config = config.clone();\n let decoy_stop = Arc::clone(&decoy_stop);\n thread::spawn(move || {\n run_decoy_destination(ack_destination, &decoy_config, security, decoy_stop)\n })\n })\n .collect();\n\n let window_cap = config.send_window_packets.max(1);\n let mut jitter = JitterQueue::new(\n config.session_seed(SESSION_ID),\n config.maximum_jitter,\n window_cap * 2 + 8,\n );\n let start = Instant::now();\n\n // The retransmission timeout is the measured (Jacobson/Karels) timeout raised to a\n // path-derived floor: the largest delay the configured path can impose (the forward jitter\n // bound), one whole receiver acknowledgement interval (the receiver emits at most one ACK per\n // `1 / ack_rate_limit` seconds, so an acknowledgement can legitimately lag its send by that\n // much), and an absolute scheduling floor. A lossless path therefore never mistakes a slow but\n // successful acknowledgement for loss.\n let ack_interval = if config.ack_rate_limit == 0 {\n POLL_INTERVAL\n } else {\n Duration::from_secs_f64(1.0 / f64::from(config.ack_rate_limit))\n };\n let path_floor = config\n .maximum_jitter\n .saturating_add(ack_interval)\n .saturating_add(POLL_INTERVAL.saturating_mul(8));\n let minimum_rto = path_floor\n .max(MINIMUM_MEASURED_RTO)\n .max(config.minimum_loss_delay);\n let mut estimator = RttEstimator::new(config.rto_variance_multiplier, minimum_rto);\n // Largest round trip observed so far, so the timeout never drops back below a round trip the\n // path has actually been measured to produce.\n let mut maximum_observed_rtt = Duration::ZERO;\n let mut congestion_window = INITIAL_CONGESTION_WINDOW.min(window_cap);\n let mut slow_start_threshold = window_cap;\n let mut maximum_congestion_window = congestion_window;\n let mut next_packet_number = 0u64;\n let mut produced = 0u64;\n let mut in_flight: BTreeMap<(u32, u64), InFlight> = BTreeMap::new();\n let mut acknowledged: BTreeSet<(u32, u64)> = BTreeSet::new();\n let mut retransmissions = 0u64;\n let mut cumulative_ack_datagrams = 0u64;\n let mut decoy_source_addresses: BTreeSet = BTreeSet::new();\n let mut buffer = vec![0; 2_048];\n\n while acknowledged.len() < packet_count as usize && start.elapsed() < LOOPBACK_TIMEOUT {\n let now = start.elapsed();\n let window = congestion_window.min(window_cap).max(1);\n\n // Admit new packets while the congestion window has room. Only unacknowledged packets are\n // counted here; the jitter queue has its own separate capacity, so a queued packet is not\n // counted twice against the send window.\n while produced < packet_count && in_flight.len() < window {\n let stream_id = STREAM_IDS[(produced as usize) % stream_count];\n let sequence = produced / stream_count as u64;\n let packet_number = next_packet_number;\n next_packet_number += 1;\n jitter\n .push(\n now,\n packet_number,\n Pending {\n stream_id,\n sequence,\n packet_number,\n attempt: 0,\n },\n )\n .expect(\"jitter capacity exceeds the send window\");\n in_flight.insert(\n (stream_id, sequence),\n InFlight {\n sent_at: now,\n attempt: 0,\n },\n );\n produced += 1;\n }\n\n // Release jittered packets that are due, applying the configured loss path to every\n // transmission (original or retransmission) by attempt number.\n for pending in jitter.drain_due(now) {\n if deterministically_lost(\n config.seed,\n pending.sequence,\n pending.attempt,\n config.udp_loss_percent,\n false,\n ) {\n continue;\n }\n let mut frame = Frame::Data(DataFrame {\n session_id: SESSION_ID,\n stream_id: pending.stream_id,\n sequence: pending.sequence,\n packet_number: pending.packet_number,\n tag: 0,\n payload: format!(\"packet-{}\", pending.sequence).into_bytes(),\n });\n security.sign_forward(&mut frame);\n send_frame(&sender_data, data_destination, &frame)?;\n if let Some(state) = in_flight.get_mut(&(pending.stream_id, pending.sequence)) {\n state.sent_at = now;\n }\n }\n\n // Retransmit anything still unacknowledged past the measured timeout, re-entering the\n // jitter and loss path with a fresh packet number rather than resending an old identity.\n // The timeout is never allowed below the largest round trip the path has actually\n // produced, so a slow-but-successful acknowledgement is not mistaken for loss.\n let timeout = estimator.timeout().max(maximum_observed_rtt);\n let timed_out: Vec<(u32, u64)> = in_flight\n .iter()\n .filter_map(|(&key, state)| {\n (now.saturating_sub(state.sent_at) >= timeout).then_some(key)\n })\n .collect();\n let mut timeout_fired = false;\n for key in timed_out {\n let packet_number = next_packet_number;\n next_packet_number += 1;\n if let Some(state) = in_flight.get_mut(&key) {\n state.attempt += 1;\n state.sent_at = now;\n let pending = Pending {\n stream_id: key.0,\n sequence: key.1,\n packet_number,\n attempt: state.attempt,\n };\n let _ = jitter.push(now, packet_number, pending);\n retransmissions += 1;\n timeout_fired = true;\n }\n }\n if timeout_fired {\n slow_start_threshold = (congestion_window / 2).max(MINIMUM_CONGESTION_WINDOW);\n congestion_window = MINIMUM_CONGESTION_WINDOW.max(slow_start_threshold / 2);\n }\n\n // Read one datagram from the return path if one is waiting.\n match sender_ack.recv_from(&mut buffer) {\n Ok((length, source)) => {\n if let Ok(Frame::Ack(ack)) = Frame::decode(\n &buffer[..length],\n config.maximum_payload_bytes,\n config.maximum_ack_ranges,\n ) && security.verify_return(&Frame::Ack(ack.clone()))\n && ack.session_id == SESSION_ID\n {\n if is_decoy_ack(&ack) {\n // Cover traffic only: a decoy never acknowledges real data, so it is\n // recorded as an observed decoy source rather than applied to delivery.\n decoy_source_addresses.insert(source);\n } else {\n let mut newly_acked = 0usize;\n for range in &ack.ranges {\n let span = range.end.saturating_sub(range.start).saturating_add(1);\n // A range wider than the whole transfer is a forgery and is ignored.\n if span == 0 || span > packet_count {\n continue;\n }\n for sequence in range.start..=range.end {\n if let Some(state) = in_flight.remove(&(ack.stream_id, sequence)) {\n acknowledged.insert((ack.stream_id, sequence));\n // Karn's algorithm: only use round trips from packets that were not\n // retransmitted, so the estimate stays unambiguous.\n if state.attempt == 0 {\n let sample = now.saturating_sub(state.sent_at);\n maximum_observed_rtt = maximum_observed_rtt.max(sample);\n estimator.sample(sample);\n }\n newly_acked += 1;\n }\n }\n }\n if newly_acked > 0 {\n if newly_acked >= 2 {\n cumulative_ack_datagrams += 1;\n }\n let growth = if congestion_window < slow_start_threshold {\n newly_acked\n } else {\n 1\n };\n congestion_window = (congestion_window + growth).min(window_cap);\n maximum_congestion_window =\n maximum_congestion_window.max(congestion_window);\n }\n }\n }\n }\n Err(error)\n if matches!(\n error.kind(),\n io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut\n ) => {}\n Err(error) => return Err(error),\n }\n }\n\n // Give the decoy destinations a short window to be observed before stopping them, so the sender\n // records a source address for each one even when the transfer finishes quickly.\n let drain_deadline = Instant::now() + DECOY_LIFETIME;\n while Instant::now() < drain_deadline && decoy_source_addresses.len() < DECOY_DESTINATION_COUNT\n {\n match sender_ack.recv_from(&mut buffer) {\n Ok((length, source)) => {\n if let Ok(Frame::Ack(ack)) = Frame::decode(\n &buffer[..length],\n config.maximum_payload_bytes,\n config.maximum_ack_ranges,\n ) && security.verify_return(&Frame::Ack(ack.clone()))\n && ack.session_id == SESSION_ID\n && is_decoy_ack(&ack)\n {\n decoy_source_addresses.insert(source);\n }\n }\n Err(error)\n if matches!(\n error.kind(),\n io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut\n ) => {}\n Err(error) => return Err(error),\n }\n }\n decoy_stop.store(true, Ordering::Relaxed);\n\n let mut decoy_datagrams = 0u64;\n let mut decoy_bytes = 0u64;\n let mut decoy_first_at: Option = None;\n let mut decoy_last_at = Duration::ZERO;\n let mut decoy_destination_addresses = Vec::with_capacity(DECOY_DESTINATION_COUNT);\n for handle in decoy_handles {\n let metrics = handle\n .join()\n .map_err(|_| io::Error::other(\"decoy destination thread panicked\"))??;\n decoy_datagrams += metrics.datagrams;\n decoy_bytes += metrics.bytes;\n decoy_destination_addresses.push(metrics.local_address);\n decoy_first_at =\n Some(decoy_first_at.map_or(metrics.first_at, |current| current.min(metrics.first_at)));\n decoy_last_at = decoy_last_at.max(metrics.last_at);\n }\n decoy_destination_addresses.sort();\n decoy_destination_addresses.dedup();\n let decoy_source_addresses: Vec = decoy_source_addresses.into_iter().collect();\n\n let receiver_metrics = receiver\n .join()\n .map_err(|_| io::Error::other(\"loopback receiver thread panicked\"))??;\n Ok(LoopbackMetrics {\n packets_sent: packet_count,\n packets_received: receiver_metrics.packets_received,\n ack_datagrams: receiver_metrics.ack_datagrams,\n decoy_datagrams,\n decoy_bytes,\n decoy_lifetime: decoy_last_at.saturating_sub(decoy_first_at.unwrap_or(Duration::ZERO)),\n decoy_destinations: decoy_destination_addresses.len() as u32,\n decoy_destination_addresses,\n decoy_source_addresses,\n receiver_return_address,\n cumulative_ack_datagrams,\n retransmissions,\n stream_id: STREAM_IDS[0],\n stream_count: stream_count as u32,\n measured_rtt: estimator.smoothed_rtt(),\n retransmission_timeout: estimator.timeout(),\n congestion_window_packets: maximum_congestion_window,\n maximum_receiver_packets: receiver_metrics.maximum_receiver_packets,\n completed: acknowledged.len() == packet_count as usize,\n })\n}\n\nfn send_frame(socket: &UdpSocket, destination: SocketAddr, frame: &Frame) -> io::Result<()> {\n socket.send_to(&frame.encode(), destination)?;\n Ok(())\n}\n\n/// Runs one decoy destination: an independent endpoint that emits authenticated,\n/// acknowledgement-shaped cover traffic on the return path. Unlike the reviewed revision, the cover\n/// does not originate at the real receiver; each decoy is its own endpoint with its own source\n/// address, so the real destination is not the only, or a uniquely identifiable, source of return\n/// traffic.\n///\n/// The cover is emitted until the shared `stop` flag is set (or the loopback timeout elapses), so it\n/// spans the whole transfer.\nfn run_decoy_destination(\n destination: SocketAddr,\n config: &TransportConfig,\n security: SessionSecurity,\n stop: Arc,\n) -> io::Result {\n let socket = UdpSocket::bind(\"127.0.0.1:0\")?;\n let local_address = socket.local_addr()?;\n let start = Instant::now();\n let mut datagrams = 0u64;\n let mut bytes = 0u64;\n let mut first_at: Option = None;\n let mut last_at = Duration::ZERO;\n let mut next_decoy_at = Duration::ZERO;\n while !stop.load(Ordering::Relaxed) && start.elapsed() < LOOPBACK_TIMEOUT {\n let now = start.elapsed();\n if now >= next_decoy_at {\n let encoded = decoy_frame(config, security).encode();\n socket.send_to(&encoded, destination)?;\n datagrams += 1;\n bytes += encoded.len() as u64;\n first_at.get_or_insert(now);\n last_at = now;\n next_decoy_at = now + DECOY_INTERVAL;\n }\n thread::sleep(POLL_INTERVAL);\n }\n Ok(DecoyMetrics {\n local_address,\n datagrams,\n bytes,\n first_at: first_at.unwrap_or(Duration::ZERO),\n last_at,\n })\n}\n\n/// Builds an authenticated acknowledgement-shaped decoy carrying only reserved sequences.\nfn decoy_frame(config: &TransportConfig, security: SessionSecurity) -> Frame {\n let count = DECOY_RANGE_COUNT\n .min(config.maximum_ack_ranges as u64)\n .max(1);\n let base = DECOY_SEQUENCE_BASE;\n let ranges = (0..count)\n .map(|offset| {\n let sequence = base + offset;\n AckRange::new(sequence, sequence).expect(\"reserved decoy range is valid\")\n })\n .collect();\n let mut frame = Frame::Ack(AckFrame {\n session_id: SESSION_ID,\n stream_id: STREAM_IDS[0],\n receiver_delay_ms: 0,\n tag: 0,\n ranges,\n });\n security.sign_return(&mut frame);\n frame\n}\n\n/// Reports whether an acknowledgement is decoy cover traffic: every range lies in the reserved\n/// decoy sequence region, which is far above any real transfer sequence and can never acknowledge\n/// real data.\nfn is_decoy_ack(ack: &AckFrame) -> bool {\n !ack.ranges.is_empty()\n && ack\n .ranges\n .iter()\n .all(|range| range.start >= DECOY_SEQUENCE_BASE)\n}\n\nfn receive_loop(\n data_socket: UdpSocket,\n ack_socket: UdpSocket,\n ack_destination: SocketAddr,\n config: &TransportConfig,\n packet_count: u64,\n stream_count: usize,\n) -> io::Result {\n let start = Instant::now();\n let security = SessionSecurity::new(config.session_seed(SESSION_ID) ^ 0x5eed_5ec0_de00_d1e5);\n let mut ack_states: Vec = (0..stream_count)\n .map(|index| {\n AckState::new(\n SESSION_ID,\n config.send_window_packets,\n config.maximum_ack_ranges,\n config.ack_rate_limit,\n )\n .with_stream_id(STREAM_IDS[index])\n })\n .collect();\n let mut distinct: BTreeSet<(u32, u64)> = BTreeSet::new();\n let mut ack_datagrams = 0;\n let mut maximum_receiver_packets = 0;\n let mut buffer = vec![0; config.maximum_payload_bytes + 64];\n let mut all_delivered = false;\n let mut acknowledged_after_delivery = vec![false; stream_count];\n let mut completed_at: Option = None;\n\n loop {\n let now = start.elapsed();\n if now >= LOOPBACK_TIMEOUT {\n break;\n }\n if completed_at.is_some() {\n // The receiver keeps no decoy role, so it can stop as soon as delivery is complete and\n // the caller has been told. Cover traffic is emitted by the decoy destinations instead.\n break;\n }\n match data_socket.recv(&mut buffer) {\n Ok(length) => {\n if let Ok(Frame::Data(data)) = Frame::decode(\n &buffer[..length],\n config.maximum_payload_bytes,\n config.maximum_ack_ranges,\n ) && security.verify_forward(&Frame::Data(data.clone()))\n && data.session_id == SESSION_ID\n && let Some(index) = STREAM_IDS\n .iter()\n .position(|&id| id == data.stream_id)\n .filter(|&index| index < stream_count)\n {\n let state = &mut ack_states[index];\n if state.record(data.sequence) != RecordOutcome::OutsideWindow {\n // Only frames the receive window actually accepted count towards delivery,\n // so the receiver never reports a transfer complete while its own anchor\n // still has a hole the sender has not been told about.\n distinct.insert((data.stream_id, data.sequence));\n }\n maximum_receiver_packets =\n maximum_receiver_packets.max(state.retained_packets());\n }\n }\n Err(error)\n if matches!(\n error.kind(),\n io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut\n ) => {}\n Err(error) => return Err(error),\n }\n\n let now = start.elapsed();\n for (index, state) in ack_states.iter_mut().enumerate() {\n if let Some(mut ack) = state.emit_if_due(now, Duration::ZERO) {\n let mut frame = Frame::Ack(ack.clone());\n security.sign_return(&mut frame);\n if let Frame::Ack(signed) = &frame {\n ack = signed.clone();\n }\n ack_socket.send_to(&Frame::Ack(ack).encode(), ack_destination)?;\n ack_datagrams += 1;\n if all_delivered {\n acknowledged_after_delivery[index] = true;\n }\n }\n }\n if distinct.len() == packet_count as usize {\n all_delivered = true;\n }\n if all_delivered && acknowledged_after_delivery.iter().all(|&sent| sent) {\n completed_at.get_or_insert(now);\n }\n }\n Ok(ReceiverMetrics {\n packets_received: distinct.len() as u64,\n ack_datagrams,\n maximum_receiver_packets,\n })\n}\n\n#[cfg(test)]\nmod tests {\n use std::time::Duration;\n\n use super::run_udp_loopback;\n use crate::TransportConfig;\n\n #[test]\n fn separate_udp_paths_complete_on_loopback() {\n let config = TransportConfig {\n send_window_packets: 128,\n maximum_jitter: std::time::Duration::from_millis(10),\n ..TransportConfig::default()\n };\n let metrics = run_udp_loopback(&config, 80).unwrap();\n\n assert!(metrics.completed);\n assert_eq!(metrics.packets_sent, 80);\n assert_eq!(metrics.packets_received, 80);\n assert!(metrics.ack_datagrams > 0);\n assert!(metrics.maximum_receiver_packets <= config.send_window_packets);\n assert!(metrics.stream_count >= 2);\n assert!(metrics.congestion_window_packets >= 1);\n }\n\n #[test]\n fn loopback_retransmits_dropped_packets_and_sends_decoys() {\n let config = TransportConfig {\n udp_loss_percent: 20,\n ..TransportConfig::default()\n };\n let metrics = run_udp_loopback(&config, 80).unwrap();\n\n assert!(metrics.completed);\n assert_eq!(metrics.packets_received, 80);\n assert!(metrics.retransmissions > 0);\n assert!(metrics.decoy_datagrams > 0);\n assert!(metrics.decoy_bytes > metrics.decoy_datagrams * 24);\n assert!(metrics.decoy_lifetime > Duration::ZERO);\n assert_eq!(metrics.stream_id, super::STREAM_IDS[0]);\n }\n\n #[test]\n fn lossless_loopback_measures_rtt_and_does_not_retransmit() {\n let config = TransportConfig {\n udp_loss_percent: 0,\n ..TransportConfig::default()\n };\n let metrics = run_udp_loopback(&config, 128).unwrap();\n\n assert!(metrics.completed);\n assert_eq!(metrics.packets_received, 128);\n // Zero-loss path: measured timeout must exceed the measured RTT, so no spurious resends.\n assert_eq!(\n metrics.retransmissions, 0,\n \"lossless path produced {} retransmissions\",\n metrics.retransmissions\n );\n assert!(metrics.measured_rtt > std::time::Duration::ZERO);\n assert!(metrics.retransmission_timeout >= metrics.measured_rtt);\n }\n}\n","Poc/22-asymmetric-transport/src/main.rs":"//! Runs the deterministic scenario matrix, the real localhost UDP demonstration, and the measured\n//! timing-signature, cross-session/path-linkability, and masking-cost evidence for the asymmetric\n//! transport proof of concept.\n//!\n//! This crate implements [Datura issue 161](http://gdatura24gtdy23lxd7ht3xzx6mi7mdlkabpvuefhrjn4t5jduviw5ad.onion/Nowhere/Datura-Network/issues/161)\n//! against the dual uni-directional stream model in `spec/specification.md`. The program exits\n//! non-zero when the UDP loopback does not complete, so a failing run is visible to scripts and CI\n//! instead of silently succeeding. `DATURA_JITTER_MS` and `DATURA_LOOPBACK_PACKETS` override the\n//! jitter bound and packet count so a deliberate failure can be reproduced.\n\nuse std::process::ExitCode;\nuse std::time::Duration;\n\nuse asymmetric_transport::{\n JitterQueue, TransportConfig, compare_timing_linkability, cross_session_linkability,\n data_loss_pattern, longest_run, masking_cost, path_linkability, required_scenarios,\n run_asymmetric_transport, run_naive_baseline, run_udp_loopback, session_delay_trace,\n shape_with_queue, signature_for_seed,\n};\n\nconst SIGNATURE_SAMPLES: u64 = 512;\nconst SIGNATURE_SESSIONS: u64 = 5;\nconst MEASUREMENT_SESSION_ID: u64 = 0;\n\nfn env_u64(name: &str, default: u64) -> u64 {\n std::env::var(name)\n .ok()\n .and_then(|value| value.parse().ok())\n .unwrap_or(default)\n}\n\nfn main() -> ExitCode {\n let mut config = TransportConfig::default();\n config.maximum_jitter = Duration::from_millis(env_u64(\n \"DATURA_JITTER_MS\",\n config.maximum_jitter.as_millis() as u64,\n ));\n let loopback_packets = env_u64(\"DATURA_LOOPBACK_PACKETS\", 16);\n\n println!(\n \"seed={} window={} ack_limit={}/s jitter_bound_ms={} timing_tick_ms={} rto_variance={} minimum_loss_delay_ms={}\",\n config.seed,\n config.send_window_packets,\n config.ack_rate_limit,\n config.maximum_jitter.as_millis(),\n config.timing_tick.as_millis(),\n config.rto_variance_multiplier,\n config.minimum_loss_delay.as_millis()\n );\n\n for scenario in required_scenarios() {\n let baseline = run_naive_baseline(&config, &scenario, 256);\n let transport = run_asymmetric_transport(&config, &scenario, 256);\n println!(\n \"{}: {} forward hops, {} return hops, {} ms return delay; asymmetric={}ms rtx={} spurious={} loss_events={} min_cwnd={} max_cwnd={} acks={} measured_rtt={:?} rto={:?} tail={:?}; naive={}ms rtx={} timeouts={} min_cwnd={} acks={}; max_jitter_queue={}\",\n scenario.name,\n scenario.forward_hops,\n scenario.return_hops,\n scenario.return_delay.as_millis(),\n transport.completion_time.as_millis(),\n transport.retransmissions,\n transport.spurious_retransmissions,\n transport.loss_events,\n transport.minimum_window_packets,\n transport.maximum_window_packets,\n transport.ack_packets,\n transport.measured_rtt,\n transport.retransmission_timeout,\n transport.tail_latency,\n baseline.completion_time.as_millis(),\n baseline.retransmissions,\n baseline.timeout_events,\n baseline.minimum_window_packets,\n baseline.ack_packets,\n baseline.maximum_jitter_queue_packets\n );\n }\n\n // Show that burst loss is a genuinely different model from random loss rather than the same\n // random drops at a higher percentage.\n let seed = config.seed;\n let scenarios = required_scenarios();\n for name in [\"burst_loss\", \"random_loss\"] {\n let scenario = scenarios\n .iter()\n .find(|scenario| scenario.name == name)\n .expect(\"scenario exists\");\n let pattern = data_loss_pattern(seed, scenario, 256);\n let drops = pattern.iter().filter(|dropped| **dropped).count();\n println!(\n \"loss_pattern {}: drops={}/{}, longest_run={}\",\n name,\n drops,\n pattern.len(),\n longest_run(&pattern)\n );\n }\n\n let mut failed = false;\n match run_udp_loopback(&config, loopback_packets) {\n Ok(loopback) => {\n println!(\n \"udp_loopback: completed={} sent={} received={} ack_datagrams={} cumulative_acks={} decoys={} decoy_destinations={} decoy_bytes={} decoy_lifetime_ms={} retransmissions={} stream_id={} max_receiver_packets={}\",\n loopback.completed,\n loopback.packets_sent,\n loopback.packets_received,\n loopback.ack_datagrams,\n loopback.cumulative_ack_datagrams,\n loopback.decoy_datagrams,\n loopback.decoy_destinations,\n loopback.decoy_bytes,\n loopback.decoy_lifetime.as_millis(),\n loopback.retransmissions,\n loopback.stream_id,\n loopback.maximum_receiver_packets\n );\n if !loopback.completed || loopback.packets_received != loopback_packets {\n eprintln!(\n \"error: udp loopback did not complete: received {} of {} packets\",\n loopback.packets_received, loopback_packets\n );\n failed = true;\n }\n }\n Err(error) => {\n eprintln!(\"error: udp loopback failed: {error}\");\n failed = true;\n }\n }\n\n // Timing linkability of a trace shaped by the real jitter queue, not a hard-coded sample.\n let source_ms: [u64; 8] = [0, 10, 35, 50, 90, 105, 160, 180];\n let shaping_queue = JitterQueue::::new(\n config.session_seed(MEASUREMENT_SESSION_ID),\n config.maximum_jitter,\n source_ms.len(),\n );\n let shaped_ms = shape_with_queue(&source_ms, &shaping_queue);\n let timing = compare_timing_linkability(&source_ms, &source_ms, &shaped_ms);\n println!(\n \"timing_linkability: raw={:.3} shaped={:.3} metric=absolute_gap_correlation queue_shaped=true\",\n timing.raw_score, timing.shaped_score\n );\n\n // Cross-session and path linkability: before (one constant seed shared by every session) and\n // after (a seed derived per session). Unlike a rounded modal mean, these compare whole traces,\n // so the score does not drift toward \"fully linkable\" as the sample count grows.\n let shared_traces: Vec> = (0..SIGNATURE_SESSIONS)\n .map(|_| session_delay_trace(config.seed, config.maximum_jitter, SIGNATURE_SAMPLES))\n .collect();\n let per_session_traces: Vec> = (0..SIGNATURE_SESSIONS)\n .map(|session| {\n session_delay_trace(\n config.session_seed(session),\n config.maximum_jitter,\n SIGNATURE_SAMPLES,\n )\n })\n .collect();\n let per_session_seeds: Vec = (0..SIGNATURE_SESSIONS)\n .map(|session| config.session_seed(session))\n .collect();\n println!(\n \"cross_session_linkability shared={:.3} per_session={:.3} samples={} sessions={}\",\n cross_session_linkability(&shared_traces),\n cross_session_linkability(&per_session_traces),\n SIGNATURE_SAMPLES,\n SIGNATURE_SESSIONS\n );\n println!(\n \"path_linkability shaped={:.3} raw_reference=1.000 sessions={}\",\n path_linkability(&source_ms, &per_session_seeds, config.maximum_jitter),\n SIGNATURE_SESSIONS\n );\n\n // Per-session measured RTT signature and masking cost.\n for session in 0..SIGNATURE_SESSIONS {\n let signature = signature_for_seed(\n config.session_seed(session),\n config.maximum_jitter,\n SIGNATURE_SAMPLES,\n );\n let cost = masking_cost(&config, session, SIGNATURE_SAMPLES);\n println!(\n \"session_seed={} rtt_mean_ms={:.2} rtt_stddev_ms={:.2} rtt_max_ms={:.2} masking_mean_ms={:.2} masking_p95_ms={:.2} masking_max_ms={:.2} masking_total_ms={:.2} queue_bound={}\",\n session,\n signature.mean_added_ms,\n signature.stddev_added_ms,\n signature.maximum_added_ms,\n cost.mean_added_ms,\n cost.p95_added_ms,\n cost.maximum_added_ms,\n cost.total_added_ms,\n cost.queue_bound_packets\n );\n }\n\n if failed {\n ExitCode::FAILURE\n } else {\n ExitCode::SUCCESS\n }\n}\n","Poc/22-asymmetric-transport/tests/decoys.rs":"//! Decoy-destination acceptance test for the asymmetric UDP transport proof of concept.\n//!\n//! rand0 review item: \"Decoys still come from the real receiver, there are no decoy destinations\".\n//! This test drives the real localhost UDP loopback and asserts that the decoy cover traffic on the\n//! return path originates from dedicated decoy destinations, not from the real receiver's return\n//! endpoint. It lives in its own file so the property is not hidden among the other review checks.\n\nuse asymmetric_transport::{TransportConfig, run_udp_loopback};\nuse std::time::Duration;\n\n#[test]\nfn decoy_traffic_comes_from_decoy_destinations_not_the_real_receiver() {\n // Inject forward loss so the transfer lasts long enough for the decoy destinations to emit\n // cover traffic on the real return UDP path.\n let config = TransportConfig {\n udp_loss_percent: 20,\n ..TransportConfig::default()\n };\n let metrics = run_udp_loopback(&config, 80).unwrap();\n\n assert!(metrics.completed, \"loopback did not complete\");\n assert_eq!(metrics.packets_received, 80);\n\n // There is decoy cover traffic on the return path and it is acknowledgement-shaped.\n assert!(\n metrics.decoy_datagrams > 0,\n \"expected decoy return datagrams, got 0\"\n );\n assert!(\n metrics.decoy_bytes > metrics.decoy_datagrams * 24,\n \"decoys must be ACK-sized, got {} bytes over {} datagrams\",\n metrics.decoy_bytes,\n metrics.decoy_datagrams\n );\n assert!(metrics.decoy_lifetime > Duration::ZERO);\n\n // The cover comes from dedicated decoy destinations.\n assert!(\n metrics.decoy_destinations >= 2,\n \"expected at least two decoy destinations, got {}\",\n metrics.decoy_destinations\n );\n assert_eq!(\n metrics.decoy_destination_addresses.len(),\n metrics.decoy_destinations as usize,\n \"each decoy destination must be counted exactly once\"\n );\n assert!(\n metrics.decoy_source_addresses.len() >= 2,\n \"expected cover from several decoy destinations, observed {:?}\",\n metrics.decoy_source_addresses\n );\n\n // Every observed decoy source is one of the decoy destinations...\n assert!(\n metrics\n .decoy_source_addresses\n .iter()\n .all(|address| metrics.decoy_destination_addresses.contains(address)),\n \"every observed decoy source must be a decoy destination: observed {:?}, destinations {:?}\",\n metrics.decoy_source_addresses,\n metrics.decoy_destination_addresses\n );\n\n // ...and none of them is the real receiver's return endpoint.\n assert!(\n !metrics\n .decoy_source_addresses\n .contains(&metrics.receiver_return_address),\n \"decoy traffic must not come from the real receiver at {}\",\n metrics.receiver_return_address\n );\n}\n"},"notes":"Fix for the rand0 decoy item on PR #187. Root cause: the return-path decoy cover was emitted by the real receiver's receive_loop on its own return socket, so there were no decoy destinations. The patch adds dedicated decoy destination endpoints that emit the cover, keeps real ACKs on the real receiver's socket, and exposes the source addresses so the property is testable. tests/decoys.rs is the required proof test in its own file. cargo fmt --check clean, clippy -D warnings clean, `cargo test --all-targets` all suites green (decoys, demonstrates, review, threat_model, unit), `cargo run --quiet` exits 0. The patch applies with `git am --3way` on the PR head. No secrets; no wallet or pricing changes; no maintainer contact; no unrelated refactor."},"attachments":[],"redactions":0,"note":"Shown as stored."}