typhoon/flow/decoy/
heavy.rs

1/// Heavy mode: sends large decoy packets at a low cadence (~0.1 pkt/s base, capped at 0.2 pkt/s), resembling background heartbeat / metric-push / software-update-poll traffic.
2use std::sync::atomic::AtomicU32;
3use std::sync::{Arc, Weak};
4use std::time::Duration;
5
6use async_trait::async_trait;
7use log::warn;
8
9use crate::bytes::{ByteBuffer, DynamicByteBuffer};
10use crate::cache::DerivedValue;
11use crate::flow::decoy::common::{DecoyCommunicationMode, DecoyFlowSender, DecoyProvider, DecoyState, exponential_variance, maintenance_timer_task, random_uniform, try_replicate};
12use crate::settings::Settings;
13use crate::settings::consts::FG_OFFSET;
14use crate::settings::keys::*;
15use crate::tailer::{IdentityType, PacketFlags};
16use crate::utils::sync::{AsyncExecutor, RwLock, sleep};
17use crate::utils::unix_timestamp_ms;
18
19/// Heavy mode implements sending large decoy packets at a low cadence (background-heartbeat shape).
20pub struct HeavyDecoyProvider<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> {
21    manager: Weak<dyn DecoyFlowSender>,
22    state: Arc<RwLock<DecoyState<T, AE>>>,
23}
24
25impl<T: IdentityType + Clone, AE: AsyncExecutor> HeavyDecoyProvider<T, AE> {
26    fn calculate_delay(state: &DecoyState<T, AE>) -> u64 {
27        let base_rate_rnd = state.settings.get(&DECOY_BASE_RATE_RND);
28        let heavy_base_rate = state.settings.get(&DECOY_HEAVY_BASE_RATE);
29        let quietness_factor = state.settings.get(&DECOY_HEAVY_QUIETNESS_FACTOR);
30        let delay_min = state.settings.get(&DECOY_HEAVY_DELAY_MIN);
31        let delay_max = state.settings.get(&DECOY_HEAVY_DELAY_MAX);
32        let delay_default = state.settings.get(&DECOY_HEAVY_DELAY_DEFAULT);
33
34        let base_rate = heavy_base_rate * random_uniform(1.0 - base_rate_rnd, 1.0 + base_rate_rnd);
35        let quietness = state.quietness_index();
36        let rate = base_rate * quietness.powf(quietness_factor) * (-state.packet_rate / state.reference_rate).exp();
37
38        let delay = if rate > 0.0 {
39            exponential_variance(rate) * 1000.0
40        } else {
41            delay_default as f64
42        };
43
44        (delay as u64).clamp(delay_min, delay_max)
45    }
46
47    pub(crate) fn calculate_length(state: &DecoyState<T, AE>) -> usize {
48        let base_length_factor = state.settings.get(&DECOY_HEAVY_BASE_LENGTH);
49        let quietness_length = state.settings.get(&DECOY_HEAVY_QUIETNESS_LENGTH);
50        let decoy_length_factor = state.settings.get(&DECOY_HEAVY_DECOY_LENGTH_FACTOR);
51        let length_min = state.settings.get(&DECOY_HEAVY_LENGTH_MIN) as usize;
52
53        let quietness = state.quietness_index();
54        let base_length = (state.packet_length_cap as f64) * (base_length_factor + quietness_length * quietness);
55        let decoy_length = random_uniform(decoy_length_factor * base_length, base_length);
56
57        (decoy_length as usize).clamp(length_min, state.packet_length_cap)
58    }
59
60    async fn timer_task(manager: Weak<dyn DecoyFlowSender>, state: Arc<RwLock<DecoyState<T, AE>>>) {
61        loop {
62            let delay = {
63                let state_guard = state.read().await;
64                let remaining = state_guard.next_decoy_time.saturating_sub(unix_timestamp_ms());
65                Duration::from_millis(remaining as u64)
66            };
67
68            sleep(delay).await;
69
70            let Some(manager_arc) = manager.upgrade() else {
71                warn!("HeavyDecoyProvider: manager dropped, stopping timer");
72                break;
73            };
74
75            {
76                let mut state_guard = state.write().await;
77                let decoy_length = state_guard.pending_length;
78
79                if state_guard.try_spend_budget(decoy_length) {
80                    let decoy_packet = state_guard.create_decoy_packet(decoy_length, false);
81                    let should_rep = state_guard.should_replicate(false);
82                    let fallthrough = state_guard.should_fallthrough();
83                    let settings = Arc::clone(&state_guard.settings);
84                    drop(state_guard);
85
86                    let body_buf = should_rep.then(|| settings.pool().allocate_precise_from_slice_with_capacity(decoy_packet.slice_end(decoy_length), 0, 0));
87                    if let Err(err) = manager_arc.send_decoy_packet(decoy_packet, fallthrough, false).await {
88                        warn!("HeavyDecoyProvider: failed to send decoy packet: {err:?}");
89                    } else if let Some(body) = body_buf {
90                        try_replicate(&state, &manager, false, body).await;
91                    }
92                }
93            }
94
95            {
96                let mut state_guard = state.write().await;
97                let delay = Self::calculate_delay(&state_guard);
98                let length = Self::calculate_length(&state_guard);
99                state_guard.schedule_next(delay, length);
100            }
101        }
102    }
103}
104
105#[async_trait]
106impl<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> DecoyProvider for HeavyDecoyProvider<T, AE> {
107    #[inline]
108    fn name(&self) -> &'static str {
109        "HeavyDecoyProvider"
110    }
111
112    async fn start(&self) {
113        let executor = {
114            let lock = self.state.read().await;
115            lock.settings.executor().clone()
116        };
117
118        let manager = self.manager.clone();
119        let state = self.state.clone();
120        executor.spawn(Self::timer_task(manager.clone(), state.clone()));
121        executor.spawn(maintenance_timer_task(manager, state));
122    }
123
124    async fn feed_input(&self, packet: DynamicByteBuffer, _tailer_buf: DynamicByteBuffer) -> Option<DynamicByteBuffer> {
125        let mut state = self.state.write().await;
126        state.update(packet.len(), false);
127        Some(packet)
128    }
129
130    async fn feed_output(&self, body: DynamicByteBuffer, tailer_buf: DynamicByteBuffer) -> Option<DynamicByteBuffer> {
131        let flags = PacketFlags::from_bits_truncate(*tailer_buf.get(FG_OFFSET));
132        if !flags.is_discardable() {
133            let mut state = self.state.write().await;
134            state.update(body.len() + tailer_buf.len(), true);
135        }
136        Some(body)
137    }
138}
139
140impl<T: IdentityType + Clone, AE: AsyncExecutor> DecoyCommunicationMode<T, AE> for HeavyDecoyProvider<T, AE> {
141    fn new(manager: Weak<dyn DecoyFlowSender>, settings: Arc<Settings<AE>>, identity: DerivedValue<T>, counter: Arc<AtomicU32>, fallthrough_probability: Option<f64>) -> Self {
142        let state = DecoyState::new(settings.clone(), identity, counter, fallthrough_probability);
143        let delay = Self::calculate_delay(&state);
144        let length = Self::calculate_length(&state);
145        let mut state = state;
146        state.schedule_next(delay, length);
147
148        Self {
149            manager,
150            state: Arc::new(RwLock::new(state)),
151        }
152    }
153}
typhoon/flow/decoy/heavy.rs

typhoon/flow/decoy/
heavy.rs
