manycastr/worker/outbound/mod.rs
1mod probe;
2mod trace;
3
4use log::{info, warn};
5use std::net::SocketAddr;
6use std::num::NonZeroU32;
7use std::sync::Arc;
8use std::sync::atomic::AtomicBool;
9use std::thread;
10use tokio::sync::mpsc::Receiver;
11
12use crate::ALL_ORIGINS;
13use crate::custom_module::Separated;
14use crate::custom_module::manycastr::instruction::InstructionType;
15use crate::custom_module::manycastr::task::TaskType;
16use crate::custom_module::manycastr::{Address, ProtocolType};
17use crate::worker::outbound::probe::send_probe;
18use crate::worker::outbound::trace::send_trace;
19use ratelimit_meter::{DirectRateLimiter, LeakyBucket};
20use socket2::{SockAddr, Socket};
21
22const DISCOVERY_WORKER_ID_OFFSET: u32 = u16::MAX as u32;
23
24/// Configuration for the outbound/sending thread
25pub struct OutboundConfig {
26 /// The unique ID of this specific worker.
27 pub worker_id: u16,
28 /// Shared signal to forcefully shut down the worker (e.g., when the CLI disconnects).
29 pub abort_outbound: Arc<AtomicBool>,
30 /// The 16-bit measurement ID
31 pub m_id: u32,
32 /// Protocol type used
33 pub p_type: ProtocolType,
34 /// Optional domain name to query in DNS measurement probes.
35 pub qname: Option<String>,
36 /// Optional URL to be embedded in the probe's payload (e.g., an opt-out link).
37 pub info_url: Option<String>,
38 /// The target rate for sending probes, measured in packets per second (pps).
39 pub probing_rate: u32,
40 /// Source address to use
41 pub src: Address,
42 /// Source port to use
43 pub sport: u16,
44 /// Destination port to use
45 pub dport: u16,
46 /// Origin ID associated with this outbound sender
47 pub origin_id: u32,
48}
49
50/// Starts the outbound worker thread that awaits tasks and sends probes.
51///
52/// # Arguments
53/// * `config` - configuration for the outbound worker thread
54/// * `outbound_rx` - on this channel we receive future tasks that are part of the current measurement
55/// * `socket` - the sender object to send probe/discovery/trace packets
56///
57/// # Returns
58/// The join handle of the outbound thread, used to await its completion at the end of a measurement
59pub fn outbound(
60 config: OutboundConfig,
61 mut outbound_rx: Receiver<InstructionType>,
62 socket: Arc<Socket>,
63) -> thread::JoinHandle<()> {
64 thread::Builder::new()
65 .name("outbound".to_string())
66 .spawn(move || {
67 let mut sent = 0u32;
68 let mut sent_discovery = 0u32;
69 let mut traces_sent = 0u32;
70 let mut failed = 0u32;
71 let mut packet_buffer = Vec::with_capacity(256);
72
73 let total_rate = config.probing_rate;
74 // Rate limiter bucket
75 let mut limiter =
76 DirectRateLimiter::<LeakyBucket>::per_second(NonZeroU32::new(total_rate).unwrap());
77
78 while let Some(instruction) = outbound_rx.blocking_recv() {
79 if config
80 .abort_outbound
81 .load(std::sync::atomic::Ordering::SeqCst)
82 {
83 // Forcefully abort the thread (discard any instructions left in the channel)
84 warn!("[Worker outbound] Abort signal received, stopping.");
85 break;
86 }
87
88 match instruction {
89 // Measurement finished
90 InstructionType::End(_) => break,
91 // Probe tasks to send
92 InstructionType::Tasks(payload) => {
93 for task in payload.tasks.iter() {
94 if task.origin_id != config.origin_id && task.origin_id != ALL_ORIGINS {
95 continue; // Not for us
96 }
97 // Skip targets of the other IP version (mixed-version measurements)
98 let dst = match &task.task_type {
99 Some(TaskType::Probe(p)) | Some(TaskType::Discovery(p)) => p.dst,
100 Some(TaskType::Trace(t)) => t.dst,
101 None => None,
102 };
103 if dst.is_some_and(|dst| dst.is_v6() != config.src.is_v6()) {
104 continue;
105 }
106 match &task.task_type {
107 Some(TaskType::Probe(probe)) => {
108 let (s, f) = send_probe(
109 &config,
110 &probe.dst.unwrap(),
111 task.session_id,
112 &socket,
113 &mut limiter,
114 false,
115 &mut packet_buffer,
116 );
117 sent += s;
118 failed += f;
119 }
120 Some(TaskType::Discovery(probe)) => {
121 let (s, f) = send_probe(
122 &config,
123 &probe.dst.unwrap(),
124 task.session_id,
125 &socket,
126 &mut limiter,
127 true,
128 &mut packet_buffer,
129 );
130 sent_discovery += s;
131 failed += f;
132 }
133 Some(TaskType::Trace(trace)) => {
134 let (s, f) = send_trace(&config, trace, &socket);
135 traces_sent += s;
136 failed += f;
137 }
138 _ => continue, // Invalid task type
139 };
140 }
141 }
142 _ => continue, // Invalid measurement
143 };
144 }
145 info!(
146 "[Worker outbound] Finished. Sent: {} ({} discovery), Traces: {}, Failed: {}",
147 sent.with_separator(),
148 sent_discovery.with_separator(),
149 traces_sent.with_separator(),
150 failed.with_separator()
151 );
152 })
153 .expect("Failed to spawn outbound thread")
154}
155
156/// Send a packet (vector of bytes) to a destination using the raw socket.
157/// IPv4: Send IPv4 header and IP payload
158/// IPv6: Send only payload (kernel writes IPv6 header)
159///
160/// The port in the destination `SockAddr` is ignored for raw sockets (the real
161/// destination lives in the IP header we craft), so it is always 0.
162///
163/// # Arguments
164/// * `socket` - attached raw socket to send probes from
165/// * `packet_buffer` - Packet to send (as bytes)
166/// * `dst` - Destination address to send to
167pub fn send_packet(
168 socket: &Socket,
169 packet_buffer: &[u8],
170 dst: &Address,
171) -> Result<(), std::io::Error> {
172 let dest_addr = SockAddr::from(SocketAddr::new(dst.into(), 0));
173 socket.send_to(packet_buffer, &dest_addr)?;
174
175 Ok(())
176}