1use std::collections::HashMap;
19
20use chia_protocol::Bytes32;
21
22use crate::constants::{FRESHNESS_WINDOW_MS, MAX_TRACKED_SENDERS, REPLAY_WINDOW};
23
24const WINDOW_WORDS: usize = REPLAY_WINDOW / 64;
26
27type SenderKey = (Bytes32, u32);
30
31#[derive(Clone)]
34struct SenderWindow {
35 highest: u64,
36 bits: [u64; WINDOW_WORDS],
38 last_seen_ms: u64,
39 started: bool,
40}
41
42impl SenderWindow {
43 fn new() -> Self {
44 Self {
45 highest: 0,
46 bits: [0; WINDOW_WORDS],
47 last_seen_ms: 0,
48 started: false,
49 }
50 }
51
52 fn get_bit(&self, offset: u64) -> bool {
53 let i = offset as usize;
54 (self.bits[i / 64] >> (i % 64)) & 1 == 1
55 }
56
57 fn set_bit(&mut self, offset: u64) {
58 let i = offset as usize;
59 self.bits[i / 64] |= 1 << (i % 64);
60 }
61
62 fn shift_left(&mut self, diff: u64) {
65 if diff as usize >= REPLAY_WINDOW {
66 self.bits = [0; WINDOW_WORDS];
67 return;
68 }
69 let shift = diff as usize;
70 let word_shift = shift / 64;
71 let bit_shift = shift % 64;
72 let mut out = [0u64; WINDOW_WORDS];
73 for i in (0..WINDOW_WORDS).rev() {
74 let mut v = 0u64;
75 if i >= word_shift {
76 v = self.bits[i - word_shift] << bit_shift;
77 if bit_shift > 0 && i > word_shift {
78 v |= self.bits[i - word_shift - 1] >> (64 - bit_shift);
79 }
80 }
81 out[i] = v;
82 }
83 self.bits = out;
84 }
85
86 fn admit(&mut self, counter: u64) -> bool {
89 if !self.started {
90 self.started = true;
91 self.highest = counter;
92 self.set_bit(0);
93 return true;
94 }
95 if counter > self.highest {
96 let diff = counter - self.highest;
97 self.shift_left(diff);
98 self.highest = counter;
99 self.set_bit(0);
100 true
101 } else {
102 let offset = self.highest - counter;
103 if offset as usize >= REPLAY_WINDOW || self.get_bit(offset) {
104 false
105 } else {
106 self.set_bit(offset);
107 true
108 }
109 }
110 }
111}
112
113#[derive(Default)]
116pub struct ReplayGuard {
117 senders: HashMap<SenderKey, SenderWindow>,
118}
119
120impl ReplayGuard {
121 #[must_use]
123 pub fn new() -> Self {
124 Self::default()
125 }
126
127 #[must_use]
129 pub fn tracked_senders(&self) -> usize {
130 self.senders.len()
131 }
132
133 #[must_use]
139 pub fn check_and_admit(
140 &mut self,
141 sender: Bytes32,
142 sender_epoch: u32,
143 counter: u64,
144 timestamp_ms: u64,
145 now_ms: u64,
146 ) -> bool {
147 let lower = now_ms.saturating_sub(FRESHNESS_WINDOW_MS);
149 let upper = now_ms.saturating_add(FRESHNESS_WINDOW_MS);
150 if timestamp_ms < lower || timestamp_ms > upper {
151 return false;
152 }
153
154 let key = (sender, sender_epoch);
155 if !self.senders.contains_key(&key) {
156 self.evict_if_full();
157 }
158 let window = self.senders.entry(key).or_insert_with(SenderWindow::new);
159 if window.admit(counter) {
160 window.last_seen_ms = window.last_seen_ms.max(timestamp_ms);
161 true
162 } else {
163 false
164 }
165 }
166
167 fn evict_if_full(&mut self) {
171 if self.senders.len() < MAX_TRACKED_SENDERS {
172 return;
173 }
174 if let Some(victim) = self
175 .senders
176 .iter()
177 .min_by_key(|(_, w)| w.last_seen_ms)
178 .map(|(k, _)| *k)
179 {
180 self.senders.remove(&victim);
181 }
182 }
183}
184
185#[cfg(test)]
186mod tests {
187 use super::*;
188
189 fn sender(n: u8) -> Bytes32 {
190 Bytes32::new([n; 32])
191 }
192
193 const NOW: u64 = 1_700_000_000_000;
194
195 #[test]
196 fn first_message_is_accepted() {
197 let mut g = ReplayGuard::new();
198 assert!(g.check_and_admit(sender(1), 0, 0, NOW, NOW));
199 }
200
201 #[test]
202 fn duplicate_counter_is_rejected() {
203 let mut g = ReplayGuard::new();
204 assert!(g.check_and_admit(sender(1), 0, 5, NOW, NOW));
205 assert!(!g.check_and_admit(sender(1), 0, 5, NOW, NOW));
206 }
207
208 #[test]
209 fn monotonic_advance_accepts() {
210 let mut g = ReplayGuard::new();
211 for c in 0..100 {
212 assert!(g.check_and_admit(sender(1), 0, c, NOW, NOW), "counter {c}");
213 }
214 }
215
216 #[test]
217 fn in_window_reorder_accepts_then_rejects_replay() {
218 let mut g = ReplayGuard::new();
219 assert!(g.check_and_admit(sender(1), 0, 10, NOW, NOW));
220 assert!(g.check_and_admit(sender(1), 0, 7, NOW, NOW));
222 assert!(g.check_and_admit(sender(1), 0, 3, NOW, NOW));
223 assert!(!g.check_and_admit(sender(1), 0, 7, NOW, NOW));
225 assert!(!g.check_and_admit(sender(1), 0, 10, NOW, NOW));
226 }
227
228 #[test]
229 fn counter_below_the_window_is_rejected() {
230 let mut g = ReplayGuard::new();
231 assert!(g.check_and_admit(sender(1), 0, REPLAY_WINDOW as u64 + 100, NOW, NOW));
232 assert!(!g.check_and_admit(sender(1), 0, 1, NOW, NOW));
234 }
235
236 #[test]
237 fn far_future_jump_clears_and_accepts() {
238 let mut g = ReplayGuard::new();
239 assert!(g.check_and_admit(sender(1), 0, 5, NOW, NOW));
240 let far = 5 + REPLAY_WINDOW as u64 * 3;
242 assert!(g.check_and_admit(sender(1), 0, far, NOW, NOW));
243 assert!(!g.check_and_admit(sender(1), 0, 5, NOW, NOW));
245 }
246
247 #[test]
248 fn stale_timestamp_rejected() {
249 let mut g = ReplayGuard::new();
250 assert!(!g.check_and_admit(sender(1), 0, 0, NOW - FRESHNESS_WINDOW_MS - 1, NOW));
251 }
252
253 #[test]
254 fn future_timestamp_beyond_window_rejected() {
255 let mut g = ReplayGuard::new();
256 assert!(!g.check_and_admit(sender(1), 0, 0, NOW + FRESHNESS_WINDOW_MS + 1, NOW));
257 }
258
259 #[test]
260 fn timestamp_at_window_edges_accepted() {
261 let mut g = ReplayGuard::new();
262 assert!(g.check_and_admit(sender(1), 0, 0, NOW - FRESHNESS_WINDOW_MS, NOW));
263 assert!(g.check_and_admit(sender(2), 0, 0, NOW + FRESHNESS_WINDOW_MS, NOW));
264 }
265
266 #[test]
267 fn distinct_senders_have_independent_counters() {
268 let mut g = ReplayGuard::new();
269 assert!(g.check_and_admit(sender(1), 0, 0, NOW, NOW));
270 assert!(g.check_and_admit(sender(2), 0, 0, NOW, NOW));
272 }
273
274 #[test]
275 fn distinct_epochs_are_independent() {
276 let mut g = ReplayGuard::new();
277 assert!(g.check_and_admit(sender(1), 0, 0, NOW, NOW));
278 assert!(g.check_and_admit(sender(1), 1, 0, NOW, NOW));
279 }
280
281 #[test]
282 fn self_pair_is_a_valid_independent_stream() {
283 let mut g = ReplayGuard::new();
286 let me = sender(7);
287 assert!(g.check_and_admit(me, 0, 0, NOW, NOW));
288 assert!(g.check_and_admit(me, 0, 1, NOW, NOW));
289 assert!(!g.check_and_admit(me, 0, 0, NOW, NOW));
290 }
291
292 #[test]
293 fn per_sender_state_is_bounded_under_a_counter_flood() {
294 let mut g = ReplayGuard::new();
296 for c in (0..10_000).step_by(1) {
297 let _ = g.check_and_admit(sender(1), 0, c, NOW, NOW);
298 }
299 assert_eq!(g.tracked_senders(), 1, "one sender = one bounded entry");
300 }
301}