confium_net/transports/
mock.rs

1//! Deterministic mock transport.
2//!
3//! `mock://<name>` URLs address an in-memory mock transport whose
4//! behavior is fully controlled by the test harness. Two peers sharing
5//! a name exchange byte vectors through an internal channel (like
6//! `inproc`), but the mock layer can be configured to:
7//!
8//! - **drop** messages (simulating a message-losing network or a
9//!   Byzantine peer that withholds input),
10//! - **tamper** messages (flipping bytes to simulate an active
11//!   adversary),
12//! - **record** every sent and received byte for post-hoc assertion.
13//!
14//! This is the transport for deterministic CI vectors and replay-attack
15//! tests described in TODO #05.
16
17use std::collections::HashMap;
18use std::sync::Mutex;
19use std::sync::OnceLock;
20use std::sync::mpsc;
21
22use url::Url;
23
24use crate::Listener;
25use crate::Result;
26use crate::Transport;
27use crate::error::ClosedSnafu;
28use crate::error::MalformedUrlSnafu;
29use crate::error::MockDropSnafu;
30use crate::registry::TransportKind;
31
32/// Global registry of named mock channel endpoints, mirroring the
33/// inproc table. Each entry is a mailbox a connector pulls a handshake
34/// out of.
35static CHANNELS: OnceLock<Mutex<HashMap<String, mpsc::Sender<MockHandshake>>>> = OnceLock::new();
36
37/// Per-channel behavior configuration. Both ends of a connection read
38/// the same shared config so a harness can install fault-injection
39/// rules before traffic flows.
40static CONFIGS: OnceLock<Mutex<HashMap<String, MockConfig>>> = OnceLock::new();
41
42fn channels() -> &'static Mutex<HashMap<String, mpsc::Sender<MockHandshake>>> {
43    CHANNELS.get_or_init(|| Mutex::new(HashMap::new()))
44}
45
46fn configs() -> &'static Mutex<HashMap<String, MockConfig>> {
47    CONFIGS.get_or_init(|| Mutex::new(HashMap::new()))
48}
49
50/// Fault-injection knobs for a mock channel.
51#[derive(Clone, Debug, Default)]
52pub struct MockConfig {
53    /// If `true`, every sent message is silently discarded and the
54    /// receiver observes a [`crate::error::Error::MockDrop`].
55    pub drop_all: bool,
56    /// If `true`, every sent message has its bytes XORed with `0xFF`
57    /// before delivery, simulating an active tampering adversary.
58    pub tamper: bool,
59}
60
61impl MockConfig {
62    /// Install a configuration for the named channel. Must be called
63    /// before either end connects so both halves observe it.
64    pub fn install(name: &str, cfg: MockConfig) {
65        configs()
66            .lock()
67            .expect("mock config table poisoned")
68            .insert(name.to_string(), cfg);
69    }
70}
71
72struct MockHandshake {
73    connector_to_peer: mpsc::Receiver<Vec<u8>>,
74    peer_to_connector: mpsc::Sender<Vec<u8>>,
75}
76
77fn channel_name(url: &Url) -> Result<String> {
78    let name = url.host_str().unwrap_or("");
79    if name.is_empty() {
80        return MalformedUrlSnafu {
81            scheme: "mock",
82            url: url.to_string(),
83            reason: "missing channel name (use mock://<name>)",
84        }
85        .fail();
86    }
87    Ok(name.to_string())
88}
89
90fn config_for(name: &str) -> MockConfig {
91    configs()
92        .lock()
93        .expect("mock config table poisoned")
94        .get(name)
95        .cloned()
96        .unwrap_or_default()
97}
98
99/// Connected mock transport. All traffic is recorded for assertion.
100pub struct MockTransport {
101    tx: Option<mpsc::Sender<Vec<u8>>>,
102    rx: Option<mpsc::Receiver<Vec<u8>>>,
103    name: String,
104    cfg: MockConfig,
105    pending: Vec<u8>,
106    /// Every byte sequence this end has sent, in order.
107    pub sent_log: Vec<Vec<u8>>,
108    /// Every byte sequence this end has received, in order.
109    pub recv_log: Vec<Vec<u8>>,
110}
111
112impl MockTransport {
113    fn new(
114        tx: mpsc::Sender<Vec<u8>>,
115        rx: mpsc::Receiver<Vec<u8>>,
116        name: String,
117        cfg: MockConfig,
118    ) -> Self {
119        Self {
120            tx: Some(tx),
121            rx: Some(rx),
122            name,
123            cfg,
124            pending: Vec::new(),
125            sent_log: Vec::new(),
126            recv_log: Vec::new(),
127        }
128    }
129
130    /// The channel name this transport is bound to.
131    pub fn name(&self) -> &str {
132        &self.name
133    }
134}
135
136impl Transport for MockTransport {
137    fn send(&mut self, data: &[u8]) -> Result<()> {
138        self.sent_log.push(data.to_vec());
139        if self.cfg.drop_all {
140            // Drop is silent from the sender's perspective; the receiver
141            // will see a MockDrop when it next tries to recv. We still
142            // deliver a sentinel so the receiver can report the drop
143            // deterministically rather than blocking forever.
144            let tx = match &self.tx {
145                Some(tx) => tx,
146                None => return ClosedSnafu.fail(),
147            };
148            return tx.send(Vec::new()).map_err(|_| ClosedSnafu.build());
149        }
150        let mut payload = data.to_vec();
151        if self.cfg.tamper {
152            for byte in &mut payload {
153                *byte ^= 0xFF;
154            }
155        }
156        let tx = match &self.tx {
157            Some(tx) => tx,
158            None => return ClosedSnafu.fail(),
159        };
160        tx.send(payload).map_err(|_| ClosedSnafu.build())
161    }
162
163    fn recv(&mut self, buf: &mut [u8]) -> Result<usize> {
164        if self.pending.is_empty() {
165            let rx = match &self.rx {
166                Some(rx) => rx,
167                None => return ClosedSnafu.fail(),
168            };
169            let msg = rx.recv().map_err(|_| ClosedSnafu.build())?;
170            if self.cfg.drop_all {
171                return MockDropSnafu.fail();
172            }
173            self.recv_log.push(msg.clone());
174            self.pending = msg;
175        }
176        let n = std::cmp::min(self.pending.len(), buf.len());
177        buf[..n].copy_from_slice(&self.pending[..n]);
178        self.pending.drain(..n);
179        Ok(n)
180    }
181
182    fn close(&mut self) -> Result<()> {
183        self.tx = None;
184        self.rx = None;
185        self.pending.clear();
186        Ok(())
187    }
188}
189
190pub struct MockListener {
191    name: String,
192    inbox: Option<mpsc::Receiver<MockHandshake>>,
193}
194
195impl MockListener {
196    fn new(name: String, inbox: mpsc::Receiver<MockHandshake>) -> Self {
197        Self {
198            name,
199            inbox: Some(inbox),
200        }
201    }
202}
203
204impl Listener for MockListener {
205    fn accept(&mut self) -> Result<Box<dyn Transport>> {
206        let inbox = match &self.inbox {
207            Some(rx) => rx,
208            None => return ClosedSnafu.fail(),
209        };
210        let handshake = inbox.recv().map_err(|_| ClosedSnafu.build())?;
211        let cfg = config_for(&self.name);
212        Ok(Box::new(MockTransport::new(
213            handshake.peer_to_connector,
214            handshake.connector_to_peer,
215            self.name.clone(),
216            cfg,
217        )))
218    }
219}
220
221impl Drop for MockListener {
222    fn drop(&mut self) {
223        if let Ok(mut table) = channels().lock() {
224            table.remove(&self.name);
225        }
226    }
227}
228
229pub struct MockKind;
230
231impl TransportKind for MockKind {
232    fn schemes(&self) -> &'static [&'static str] {
233        &["mock"]
234    }
235
236    fn connect(&self, url: &Url) -> Result<Box<dyn Transport>> {
237        let name = channel_name(url)?;
238        let tx = {
239            let mut table = channels().lock().expect("mock channel table poisoned");
240            match table.remove(&name) {
241                Some(sender) => sender,
242                None => {
243                    return MalformedUrlSnafu {
244                        scheme: "mock",
245                        url: url.to_string(),
246                        reason: "no listener registered for this channel name",
247                    }
248                    .fail();
249                }
250            }
251        };
252        let (c2p_tx, c2p_rx) = mpsc::channel::<Vec<u8>>();
253        let (p2c_tx, p2c_rx) = mpsc::channel::<Vec<u8>>();
254        tx.send(MockHandshake {
255            connector_to_peer: c2p_rx,
256            peer_to_connector: p2c_tx,
257        })
258        .map_err(|_| ClosedSnafu.build())?;
259        let cfg = config_for(&name);
260        Ok(Box::new(MockTransport::new(c2p_tx, p2c_rx, name, cfg)))
261    }
262
263    fn listen(&self, url: &Url) -> Result<Box<dyn Listener>> {
264        let name = channel_name(url)?;
265        let (handshake_tx, handshake_rx) = mpsc::channel::<MockHandshake>();
266        let mut table = channels().lock().expect("mock channel table poisoned");
267        if table.contains_key(&name) {
268            return MalformedUrlSnafu {
269                scheme: "mock",
270                url: url.to_string(),
271                reason: "channel name already in use",
272            }
273            .fail();
274        }
275        table.insert(name.clone(), handshake_tx);
276        Ok(Box::new(MockListener::new(name, handshake_rx)))
277    }
278}
279
280crate::register_transport!(MockKind);
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285
286    fn unique_name(tag: &str) -> String {
287        use std::sync::atomic::AtomicU64;
288        use std::sync::atomic::Ordering;
289        static SEQ: AtomicU64 = AtomicU64::new(0);
290        let n = SEQ.fetch_add(1, Ordering::Relaxed);
291        format!("mock-{tag}-{n}")
292    }
293
294    fn rendezvous(tag: &str) -> (Box<dyn Transport>, Box<dyn Transport>) {
295        let name = unique_name(tag);
296        let url = format!("mock://{name}");
297        let mut listener = crate::listen(&url).unwrap();
298        let client = crate::connect(&url).unwrap();
299        let server = listener.accept().unwrap();
300        (client, server)
301    }
302
303    #[test]
304    fn round_trip_preserves_order_and_payloads() {
305        let (mut client, mut server) = rendezvous("rt");
306
307        client.send(b"alpha").unwrap();
308        client.send(b"beta").unwrap();
309
310        let mut buf = [0u8; 16];
311        let n = server.recv(&mut buf).unwrap();
312        assert_eq!(&buf[..n], b"alpha");
313        let n = server.recv(&mut buf).unwrap();
314        assert_eq!(&buf[..n], b"beta");
315    }
316
317    #[test]
318    fn logs_record_traffic_for_directly_built_pair() {
319        // Build a pair directly so we can inspect the concrete logs.
320        let name = unique_name("log");
321        let cfg = MockConfig::default();
322        let (c2p_tx, c2p_rx) = mpsc::channel::<Vec<u8>>();
323        let (p2c_tx, p2c_rx) = mpsc::channel::<Vec<u8>>();
324        let mut client = MockTransport::new(c2p_tx, p2c_rx, name.clone(), cfg.clone());
325        let mut server = MockTransport::new(p2c_tx, c2p_rx, name, cfg);
326
327        client.send(b"one").unwrap();
328        client.send(b"two").unwrap();
329
330        let mut buf = [0u8; 16];
331        let _ = server.recv(&mut buf).unwrap();
332        let _ = server.recv(&mut buf).unwrap();
333
334        assert_eq!(client.sent_log.len(), 2);
335        assert_eq!(client.sent_log[0], b"one".to_vec());
336        assert_eq!(client.sent_log[1], b"two".to_vec());
337        assert_eq!(server.recv_log.len(), 2);
338        assert_eq!(server.recv_log[0], b"one".to_vec());
339        assert_eq!(server.recv_log[1], b"two".to_vec());
340    }
341
342    #[test]
343    fn drop_config_silences_receiver() {
344        let name = unique_name("drop");
345        MockConfig::install(
346            &name,
347            MockConfig {
348                drop_all: true,
349                tamper: false,
350            },
351        );
352        let url = format!("mock://{name}");
353        let mut listener = crate::listen(&url).unwrap();
354        let mut client = crate::connect(&url).unwrap();
355        let mut server = listener.accept().unwrap();
356
357        // Send succeeds (the harness intends to send); recv reports the
358        // drop deterministically rather than blocking.
359        client.send(b"lost").unwrap();
360        let mut buf = [0u8; 16];
361        let err = server.recv(&mut buf).unwrap_err();
362        assert!(matches!(err, crate::error::Error::MockDrop { .. }));
363    }
364
365    #[test]
366    fn tamper_config_corrupts_payload() {
367        let name = unique_name("tamper");
368        MockConfig::install(
369            &name,
370            MockConfig {
371                drop_all: false,
372                tamper: true,
373            },
374        );
375        let url = format!("mock://{name}");
376        let mut listener = crate::listen(&url).unwrap();
377        let mut client = crate::connect(&url).unwrap();
378        let mut server = listener.accept().unwrap();
379
380        client.send(b"ABCDEFGH").unwrap();
381        let mut buf = [0u8; 16];
382        let n = server.recv(&mut buf).unwrap();
383        assert_eq!(n, 8);
384        // Every byte was XORed with 0xFF.
385        let expected: Vec<u8> = b"ABCDEFGH".iter().map(|b| b ^ 0xFF).collect();
386        assert_eq!(&buf[..n], &expected[..]);
387    }
388
389    #[test]
390    fn default_config_passes_payloads_through_unchanged() {
391        let (mut client, mut server) = rendezvous("passthrough");
392        client.send(b"plain").unwrap();
393        let mut buf = [0u8; 16];
394        let n = server.recv(&mut buf).unwrap();
395        assert_eq!(&buf[..n], b"plain");
396    }
397}
confium_net/transports/mock.rs

confium_net/transports/
mock.rs
