reth_ipc/
stream_codec.rs

1// Copyright (c) 2015-2017 Parity Technologies Limited
2//
3// Permission is hereby granted, free of charge, to any
4// person obtaining a copy of this software and associated
5// documentation files (the "Software"), to deal in the
6// Software without restriction, including without
7// limitation the rights to use, copy, modify, merge,
8// publish, distribute, sublicense, and/or sell copies of
9// the Software, and to permit persons to whom the Software
10// is furnished to do so, subject to the following
11// conditions:
12//
13// The above copyright notice and this permission notice
14// shall be included in all copies or substantial portions
15// of the Software.
16//
17// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
18// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
19// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
20// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
21// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
22// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
23// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
24// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
25// DEALINGS IN THE SOFTWARE.
26
27// This basis of this file has been taken from the deprecated jsonrpc codebase:
28// https://github.com/paritytech/jsonrpc
29
30use bytes::{Buf, BufMut, BytesMut};
31use std::{io, str};
32
33/// Separator for enveloping messages in streaming codecs
34#[derive(Debug, Clone)]
35pub enum Separator {
36    /// No envelope is expected between messages. Decoder will try to figure out
37    /// message boundaries by accumulating incoming bytes until valid JSON is formed.
38    /// Encoder will send messages without any boundaries between requests.
39    Empty,
40    /// Byte is used as a sentinel between messages
41    Byte(u8),
42}
43
44impl Default for Separator {
45    fn default() -> Self {
46        Self::Byte(b'\n')
47    }
48}
49
50/// Stream codec for streaming protocols (ipc, tcp)
51#[derive(Debug)]
52pub struct StreamCodec {
53    incoming_separator: Separator,
54    outgoing_separator: Separator,
55    max_length: usize,
56    /// The first `max_length + 1` bytes of the message being skipped.
57    truncated: Option<BytesMut>,
58    scan: ScanState,
59}
60
61impl StreamCodec {
62    /// Default codec with streaming input data. Input can be both enveloped and not.
63    pub fn stream_incoming() -> Self {
64        Self::new(Separator::Empty, Default::default())
65    }
66
67    /// New custom stream codec
68    pub const fn new(incoming_separator: Separator, outgoing_separator: Separator) -> Self {
69        Self {
70            incoming_separator,
71            outgoing_separator,
72            max_length: usize::MAX,
73            truncated: None,
74            scan: ScanState::new(),
75        }
76    }
77
78    /// Truncates messages longer than `max_length` bytes to `max_length + 1` bytes, so the
79    /// reader can reject them without buffering the whole message.
80    pub const fn with_max_length(mut self, max_length: usize) -> Self {
81        self.max_length = max_length;
82        self
83    }
84
85    /// Starts skipping the message at the start of `buf`, keeping its first bytes.
86    fn truncate(&mut self, buf: &mut BytesMut) {
87        self.truncated = Some(buf.split_to(self.max_length + 1));
88        buf.clear();
89    }
90}
91
92impl Default for StreamCodec {
93    fn default() -> Self {
94        Self::new(Default::default(), Default::default())
95    }
96}
97
98/// Scan state of the [`Separator::Empty`] decoder, carried across `decode` calls so bytes
99/// scanned by a previous call are not scanned again.
100///
101/// Indices refer to the current read buffer and must be reset whenever bytes are consumed from
102/// it.
103#[derive(Debug)]
104struct ScanState {
105    depth: i32,
106    in_str: bool,
107    is_escaped: bool,
108    start_idx: usize,
109    whitespaces: usize,
110    cursor: usize,
111}
112
113impl ScanState {
114    const fn new() -> Self {
115        Self { depth: 0, in_str: false, is_escaped: false, start_idx: 0, whitespaces: 0, cursor: 0 }
116    }
117}
118
119impl Default for ScanState {
120    fn default() -> Self {
121        Self::new()
122    }
123}
124
125#[inline]
126const fn is_whitespace(byte: u8) -> bool {
127    matches!(byte, 0x0D | 0x0A | 0x20 | 0x09)
128}
129
130/// Converts the start of a truncated message, which may end inside a UTF-8 sequence.
131///
132/// The reader rejects the message by its length alone, so non-ASCII bytes are replaced with `?`
133/// in place, which keeps the length.
134fn lossy(truncated: BytesMut) -> String {
135    String::from_utf8(truncated.into()).unwrap_or_else(|err| {
136        let mut bytes = err.into_bytes();
137        bytes.iter_mut().filter(|b| !b.is_ascii()).for_each(|b| *b = b'?');
138        String::from_utf8(bytes).expect("ASCII is valid UTF-8")
139    })
140}
141
142impl tokio_util::codec::Decoder for StreamCodec {
143    type Item = String;
144    type Error = io::Error;
145
146    fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
147        if let Separator::Byte(separator) = self.incoming_separator {
148            if let Some(i) = buf.as_ref().iter().position(|&b| b == separator) {
149                let line = buf.split_to(i);
150                buf.advance(1);
151                if let Some(truncated) = self.truncated.take() {
152                    return Ok(Some(lossy(truncated)))
153                }
154
155                match str::from_utf8(line.as_ref()) {
156                    Ok(s) => Ok(Some(s.to_string())),
157                    Err(_) => Err(io::Error::other("invalid UTF-8")),
158                }
159            } else {
160                if self.truncated.is_some() {
161                    buf.clear();
162                } else if buf.len() > self.max_length {
163                    self.truncate(buf);
164                }
165                Ok(None)
166            }
167        } else {
168            // resume scanning at the byte the previous call stopped at
169            while self.scan.cursor < buf.len() {
170                let idx = self.scan.cursor;
171                let byte = buf[idx];
172
173                if (byte == b'{' || byte == b'[') && !self.scan.in_str {
174                    if self.scan.depth == 0 {
175                        self.scan.start_idx = idx;
176                    }
177                    self.scan.depth += 1;
178                } else if (byte == b'}' || byte == b']') && !self.scan.in_str {
179                    self.scan.depth -= 1;
180                } else if byte == b'"' && !self.scan.is_escaped {
181                    self.scan.in_str = !self.scan.in_str;
182                } else if is_whitespace(byte) {
183                    self.scan.whitespaces += 1;
184                }
185                self.scan.is_escaped = byte == b'\\' && !self.scan.is_escaped && self.scan.in_str;
186
187                if self.scan.depth == 0 &&
188                    let Some(truncated) = self.truncated.take()
189                {
190                    self.scan = ScanState::new();
191                    buf.advance(idx + 1);
192                    return Ok(Some(lossy(truncated)))
193                }
194                if self.scan.depth == 0 &&
195                    idx != self.scan.start_idx &&
196                    idx - self.scan.start_idx + 1 > self.scan.whitespaces
197                {
198                    let start = self.scan.start_idx;
199                    let end = idx + 1;
200                    // reset before advancing the buffer because the stored indices go stale
201                    self.scan = ScanState::new();
202                    if start > 0 {
203                        buf.advance(start);
204                    }
205                    let bts = buf.split_to(end - start);
206                    // Skip invalid UTF-8 and keep scanning, the rest of the buffer may hold more
207                    // messages.
208                    match String::from_utf8(bts.into()) {
209                        Ok(msg) => return Ok(Some(msg)),
210                        Err(_) => continue,
211                    }
212                }
213
214                self.scan.cursor += 1;
215            }
216
217            // Keep at most `max_length` bytes buffered between calls.
218            if self.truncated.is_some() {
219                buf.clear();
220                self.scan.cursor = 0;
221            } else if buf.len() > self.max_length {
222                if self.scan.depth <= 0 {
223                    return Err(io::Error::new(io::ErrorKind::InvalidData, "message too large"))
224                }
225                // The bytes before an open value are never part of a message.
226                buf.advance(self.scan.start_idx);
227                self.scan.cursor -= self.scan.start_idx;
228                self.scan.start_idx = 0;
229                if buf.len() > self.max_length {
230                    self.truncate(buf);
231                    self.scan.cursor = 0;
232                }
233            }
234            Ok(None)
235        }
236    }
237}
238
239impl tokio_util::codec::Encoder<String> for StreamCodec {
240    type Error = io::Error;
241
242    fn encode(&mut self, msg: String, buf: &mut BytesMut) -> io::Result<()> {
243        match self.outgoing_separator {
244            Separator::Byte(separator) => {
245                buf.reserve(msg.len() + 1);
246                buf.extend_from_slice(msg.as_bytes());
247                buf.put_u8(separator);
248            }
249            Separator::Empty => buf.extend_from_slice(msg.as_bytes()),
250        }
251        Ok(())
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258    use bytes::BufMut;
259    use rand::{rngs::StdRng, RngCore, SeedableRng};
260    use tokio_util::codec::Decoder;
261
262    #[test]
263    fn simple_encode() {
264        let mut buf = BytesMut::with_capacity(2048);
265        buf.put_slice(b"{ test: 1 }{ test: 2 }{ test: 3 }");
266
267        let mut codec = StreamCodec::stream_incoming();
268
269        let request = codec
270            .decode(&mut buf)
271            .expect("There should be no error in simple test")
272            .expect("There should be at least one request in simple test");
273
274        assert_eq!(request, "{ test: 1 }");
275    }
276
277    #[test]
278    fn escape() {
279        let mut buf = BytesMut::with_capacity(2048);
280        buf.put_slice(br#"{ test: "\"\\" }{ test: "\ " }{ test: "\}" }[ test: "\]" ]"#);
281
282        let mut codec = StreamCodec::stream_incoming();
283
284        let request = codec
285            .decode(&mut buf)
286            .expect("There should be no error in first escape test")
287            .expect("There should be a request in first escape test");
288
289        assert_eq!(request, r#"{ test: "\"\\" }"#);
290
291        let request2 = codec
292            .decode(&mut buf)
293            .expect("There should be no error in 2nd escape test")
294            .expect("There should be a request in 2nd escape test");
295        assert_eq!(request2, r#"{ test: "\ " }"#);
296
297        let request3 = codec
298            .decode(&mut buf)
299            .expect("There should be no error in 3rd escape test")
300            .expect("There should be a request in 3rd escape test");
301        assert_eq!(request3, r#"{ test: "\}" }"#);
302
303        let request4 = codec
304            .decode(&mut buf)
305            .expect("There should be no error in 4th escape test")
306            .expect("There should be a request in 4th escape test");
307        assert_eq!(request4, r#"[ test: "\]" ]"#);
308    }
309
310    #[test]
311    fn whitespace() {
312        let mut buf = BytesMut::with_capacity(2048);
313        buf.put_slice(b"{ test: 1 }\n\n\n\n{ test: 2 }\n\r{\n test: 3 }  ");
314
315        let mut codec = StreamCodec::stream_incoming();
316
317        let request = codec
318            .decode(&mut buf)
319            .expect("There should be no error in first whitespace test")
320            .expect("There should be a request in first whitespace test");
321
322        assert_eq!(request, "{ test: 1 }");
323
324        let request2 = codec
325            .decode(&mut buf)
326            .expect("There should be no error in first 2nd test")
327            .expect("There should be a request in 2nd whitespace test");
328        assert_eq!(request2, "{ test: 2 }");
329
330        let request3 = codec
331            .decode(&mut buf)
332            .expect("There should be no error in first 3rd test")
333            .expect("There should be a request in 3rd whitespace test");
334        assert_eq!(request3, "{\n test: 3 }");
335
336        let request4 = codec.decode(&mut buf).expect("There should be no error in first 4th test");
337        assert!(
338            request4.is_none(),
339            "There should be no 4th request because it contains only whitespaces"
340        );
341    }
342
343    #[test]
344    fn fragmented_encode() {
345        let mut buf = BytesMut::with_capacity(2048);
346        buf.put_slice(b"{ test: 1 }{ test: 2 }{ tes");
347
348        let mut codec = StreamCodec::stream_incoming();
349
350        let request = codec
351            .decode(&mut buf)
352            .expect("There should be no error in first fragmented test")
353            .expect("There should be at least one request in first fragmented test");
354        assert_eq!(request, "{ test: 1 }");
355        codec
356            .decode(&mut buf)
357            .expect("There should be no error in second fragmented test")
358            .expect("There should be at least one request in second fragmented test");
359        assert_eq!(String::from_utf8(buf.as_ref().to_vec()).unwrap(), "{ tes");
360
361        buf.put_slice(b"t: 3 }");
362        let request = codec
363            .decode(&mut buf)
364            .expect("There should be no error in third fragmented test")
365            .expect("There should be at least one request in third fragmented test");
366        assert_eq!(request, "{ test: 3 }");
367    }
368
369    #[test]
370    fn huge() {
371        let request = r#"{
372			"jsonrpc":"2.0",
373			"method":"say_hello",
374			"params": [
375				42,
376				0,
377				{
378					"from":"0xb60e8dd61c5d32be8058bb8eb970870f07233155",
379					"gas":"0x2dc6c0",
380					"data":"0x606060405260003411156010576002565b6001805433600160a060020a0319918216811790925560028054909116909117905561291f806100406000396000f3606060405236156100e55760e060020a600035046304029f2381146100ed5780630a1273621461015f57806317c1dd87146102335780631f9ea25d14610271578063266fa0e91461029357806349593f5314610429578063569aa0d8146104fc57806359a4669f14610673578063647a4d5f14610759578063656104f5146108095780636e9febfe1461082b57806370de8c6e1461090d57806371bde852146109ed5780638f30435d14610ab4578063916dbc1714610da35780639f5a7cd414610eef578063c91540f614610fe6578063eae99e1c146110b5578063fedc2a281461115a575b61122d610002565b61122d6004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050604435915050606435600154600090600160a060020a03908116339091161461233357610002565b61122f6004808035906020019082018035906020019191908080601f016020809104026020016040519081016040528093929190818152602001838380828437509496505093359350506044359150506064355b60006000600060005086604051808280519060200190808383829060006004602084601f0104600f02600301f1509050019150509081526020016040518091039020600050905042816005016000508560ff1660028110156100025760040201835060010154604060020a90046001604060020a0316116115df576115d6565b6112416004355b604080516001604060020a038316408152606060020a33600160a060020a031602602082015290519081900360340190205b919050565b61122d600435600254600160a060020a0390811633909116146128e357610002565b61125e6004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050505060006000600060006000600060005087604051808280519060200190808383829060006004602084601f0104600f02600301f1509050019150509081526020016040518091039020600050905080600001600050600087600160a060020a0316815260200190815260200160002060005060000160059054906101000a90046001604060020a03169450845080600001600050600087600160a060020a03168152602001908152602001600020600050600001600d9054906101000a90046001604060020a03169350835080600001600050600087600160a060020a0316815260200190815260200160002060005060000160009054906101000a900460ff169250825080600001600050600087600160a060020a0316815260200190815260200160002060005060000160019054906101000a900463ffffffff16915081505092959194509250565b61122d6004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050604435915050606435608435600060006000600060005088604051808280519060200190808383829060006004602084601f0104600f02600301f15090500191505090815260200160405180910390206000509250346000141515611c0e5760405133600160a060020a0316908290349082818181858883f193505050501515611c1a57610002565b6112996004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050604435915050600060006000600060006000600060006000508a604051808280519060200190808383829060006004602084601f0104600f02600301f15090500191505090815260200160405180910390206000509050806001016000508960ff16600281101561000257600160a060020a038a168452828101600101602052604084205463ffffffff1698506002811015610002576040842054606060020a90046001604060020a031697506002811015610002576040842054640100000000900463ffffffff169650600281101561000257604084206001015495506002811015610002576040842054604060020a900463ffffffff169450600281101561000257505060409091205495999498509296509094509260a060020a90046001604060020a0316919050565b61122d6004808035906020019082018035906020019191908080601f016020809104026020016040519081016040528093929190818152602001838380828437509496505050505050506000600060005082604051808280519060200190808383829060006004602084601f0104600f02600301f15090500191505090815260200160405180910390206000509050348160050160005082600d0160009054906101000a900460ff1660ff16600281101561000257600402830160070180546001608060020a0381169093016001608060020a03199390931692909217909155505b5050565b6112e26004808035906020019082018035906020019191908080601f01602080910003423423094734987103498712093847102938740192387401349857109487501938475"
381				}
382			]
383		}"#;
384
385        let mut buf = BytesMut::with_capacity(65536);
386        buf.put_slice(request.as_bytes());
387
388        let mut codec = StreamCodec::stream_incoming();
389
390        let parsed_request = codec
391            .decode(&mut buf)
392            .expect("There should be no error in huge test")
393            .expect("There should be at least one request huge test");
394        assert_eq!(request, parsed_request);
395    }
396
397    #[test]
398    fn simple_line_codec() {
399        let mut buf = BytesMut::with_capacity(2048);
400        buf.put_slice(b"{ test: 1 }\n{ test: 2 }\n{ test: 3 }");
401
402        let mut codec = StreamCodec::default();
403
404        let request = codec
405            .decode(&mut buf)
406            .expect("There should be no error in simple test")
407            .expect("There should be at least one request in simple test");
408        let request2 = codec
409            .decode(&mut buf)
410            .expect("There should be no error in simple test")
411            .expect("There should be at least one request in simple test");
412
413        assert_eq!(request, "{ test: 1 }");
414        assert_eq!(request2, "{ test: 2 }");
415    }
416
417    #[test]
418    fn serde_json_accepts_whitespace_wrapped_json() {
419        let json = "   { \"key\": \"value\" }   ";
420
421        #[derive(serde::Deserialize, Debug, PartialEq)]
422        struct Obj {
423            key: String,
424        }
425
426        let parsed: Result<Obj, _> = serde_json::from_str(json);
427        assert!(parsed.is_ok(), "serde_json should accept whitespace-wrapped JSON");
428        assert_eq!(parsed.unwrap(), Obj { key: "value".into() });
429    }
430
431    /// Multiple messages fed one byte at a time.
432    #[test]
433    fn pipelined_messages_byte_by_byte() {
434        let payload = br#"{"a":1}{"b":2}{"c":3}"#;
435        let mut buf = BytesMut::with_capacity(256);
436        let mut codec = StreamCodec::stream_incoming();
437        let mut decoded = Vec::new();
438
439        for byte in payload {
440            buf.put_u8(*byte);
441            while let Some(m) = codec.decode(&mut buf).unwrap() {
442                decoded.push(m);
443            }
444        }
445
446        assert_eq!(decoded, vec![r#"{"a":1}"#, r#"{"b":2}"#, r#"{"c":3}"#]);
447        assert!(buf.is_empty());
448    }
449
450    /// The stateless `Separator::Empty` decode loop that `ScanState` replaced, kept as the
451    /// reference implementation for the differential tests below.
452    ///
453    /// Messages with invalid UTF-8 are skipped.
454    struct ReferenceDecoder;
455
456    impl Decoder for ReferenceDecoder {
457        type Item = String;
458        type Error = io::Error;
459
460        fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
461            let mut depth = 0i32;
462            let mut in_str = false;
463            let mut is_escaped = false;
464            let mut start_idx = 0usize;
465            let mut whitespaces = 0usize;
466
467            for idx in 0..buf.as_ref().len() {
468                let byte = buf.as_ref()[idx];
469
470                if (byte == b'{' || byte == b'[') && !in_str {
471                    if depth == 0 {
472                        start_idx = idx;
473                    }
474                    depth += 1;
475                } else if (byte == b'}' || byte == b']') && !in_str {
476                    depth -= 1;
477                } else if byte == b'"' && !is_escaped {
478                    in_str = !in_str;
479                } else if is_whitespace(byte) {
480                    whitespaces += 1;
481                }
482                is_escaped = byte == b'\\' && !is_escaped && in_str;
483
484                if depth == 0 && idx != start_idx && idx - start_idx + 1 > whitespaces {
485                    if start_idx > 0 {
486                        buf.advance(start_idx);
487                    }
488                    let bts = buf.split_to(idx + 1 - start_idx);
489                    return match String::from_utf8(bts.into()) {
490                        Ok(msg) => Ok(Some(msg)),
491                        Err(_) => self.decode(buf),
492                    }
493                }
494            }
495            Ok(None)
496        }
497    }
498
499    /// Feeds the same bytes with the same chunk boundaries to the codec and the reference
500    /// decoder and returns the decode results of both. Asserts that both consume the same bytes.
501    fn drain_both(bytes: &[u8], chunks: &[usize]) -> (Vec<Option<String>>, Vec<Option<String>>) {
502        let mut ours = StreamCodec::stream_incoming();
503        let mut theirs = ReferenceDecoder;
504
505        let mut buf_ours = BytesMut::with_capacity(bytes.len() + 16);
506        let mut buf_theirs = BytesMut::with_capacity(bytes.len() + 16);
507        let mut out_ours = Vec::new();
508        let mut out_theirs = Vec::new();
509
510        let mut cur = 0usize;
511        for &c in chunks {
512            let end = cur.saturating_add(c.max(1)).min(bytes.len());
513            if cur == end {
514                break;
515            }
516            buf_ours.put_slice(&bytes[cur..end]);
517            buf_theirs.put_slice(&bytes[cur..end]);
518            loop {
519                let a = ours.decode(&mut buf_ours).unwrap();
520                let b = theirs.decode(&mut buf_theirs).unwrap();
521                out_ours.push(a.clone());
522                out_theirs.push(b.clone());
523                if a.is_none() && b.is_none() {
524                    break;
525                }
526            }
527            cur = end;
528        }
529        if cur < bytes.len() {
530            buf_ours.put_slice(&bytes[cur..]);
531            buf_theirs.put_slice(&bytes[cur..]);
532            loop {
533                let a = ours.decode(&mut buf_ours).unwrap();
534                let b = theirs.decode(&mut buf_theirs).unwrap();
535                out_ours.push(a.clone());
536                out_theirs.push(b.clone());
537                if a.is_none() && b.is_none() {
538                    break;
539                }
540            }
541        }
542
543        assert_eq!(buf_ours, buf_theirs, "residual buffers diverged");
544
545        (out_ours, out_theirs)
546    }
547
548    /// Hand-picked adversarial inputs under several chunk schedules.
549    #[test]
550    fn differential_curated_corpus() {
551        let inputs: &[&[u8]] = &[
552            b"",
553            b"{}",
554            b"[]",
555            b"{}{}",
556            br#"{"a":1}"#,
557            br#"{"a":1}{"b":2}"#,
558            b"   {}   ",
559            b"   { }   ",
560            b"   {\n}\n   ",
561            br#"{"a":"\""}"#,
562            br#"{"a":"\\"}"#,
563            br#"{"a":"\\\""}"#,
564            b"{\xff}",
565            b"{\xff}{\"a\":1}",
566            b"]{}",
567            b"}{}",
568            b"]{}{}",
569            b"}{}{}",
570            b"][{}",
571            b"abc",
572            b"a",
573            b"\n\n\n",
574            br#"{"\":1}"#,
575            br#"{"escaped":"line\nline"}"#,
576            b"{[]}",
577            b"[{}]",
578            b"[{},{},{}]",
579        ];
580        let chunk_variants: &[&[usize]] =
581            &[&[usize::MAX], &[1], &[2], &[3], &[5], &[7], &[1, 3, 2, 7, 1, 4]];
582
583        for input in inputs {
584            for chunks in chunk_variants {
585                let (ours, theirs) = drain_both(input, chunks);
586                assert_eq!(
587                    ours,
588                    theirs,
589                    "differential mismatch on {:?} with chunks {:?}",
590                    std::str::from_utf8(input).unwrap_or("<non-utf8>"),
591                    chunks
592                );
593            }
594        }
595    }
596
597    /// Random byte sequences over a JSON-heavy alphabet under random chunk schedules.
598    #[test]
599    fn differential_random_bytes() {
600        const SEQUENCES: usize = 200;
601        const SCHEDULES: usize = 5;
602        const MAX_LEN: usize = 128;
603
604        let mut rng = StdRng::seed_from_u64(0x00CC_2259_C0DE_C0DE);
605        let alphabet: &[u8] = b"{}[]\":,\\ \n \t a1{}[]\"";
606
607        for seq in 0..SEQUENCES {
608            let len = 1 + (rng.next_u32() as usize % MAX_LEN);
609            let mut bytes = Vec::with_capacity(len);
610            for _ in 0..len {
611                let b = alphabet[rng.next_u32() as usize % alphabet.len()];
612                bytes.push(b);
613            }
614            for sched in 0..SCHEDULES {
615                let mut chunks = Vec::new();
616                let mut remaining = len;
617                while remaining > 0 {
618                    let c = 1 + (rng.next_u32() as usize % remaining.min(16));
619                    chunks.push(c);
620                    remaining -= c;
621                }
622                let (ours, theirs) = drain_both(&bytes, &chunks);
623                assert_eq!(
624                    ours, theirs,
625                    "differential mismatch on seq={} sched={} bytes={:?}",
626                    seq, sched, bytes
627                );
628            }
629        }
630    }
631
632    /// Randomly chunked well-formed JSON-RPC envelopes.
633    #[test]
634    fn differential_random_wellformed_json_rpc() {
635        let mut rng = StdRng::seed_from_u64(0x00DE_C0DE_C002_2259);
636        for _ in 0..20 {
637            let str_len = 500 + (rng.next_u32() as usize % 4096);
638            let filler: String = (0..str_len)
639                .map(|_| {
640                    let b = rng.next_u32() as u8 & 0x7F;
641                    // printable ASCII without quotes or backslashes
642                    if b < 0x20 || b == b'"' || b == b'\\' {
643                        b'a'
644                    } else {
645                        b
646                    }
647                })
648                .map(|b| b as char)
649                .collect();
650            let msg = format!(
651                r#"{{"jsonrpc":"2.0","method":"engine_test","params":["{}"],"id":1}}"#,
652                filler
653            );
654            let bytes = msg.as_bytes();
655
656            let mut chunks = Vec::new();
657            let mut remaining = bytes.len();
658            while remaining > 0 {
659                let c = 1 + (rng.next_u32() as usize % remaining.min(64));
660                chunks.push(c);
661                remaining -= c;
662            }
663
664            let (ours, theirs) = drain_both(bytes, &chunks);
665            assert_eq!(ours, theirs, "differential mismatch on well-formed JSON-RPC");
666            assert!(
667                ours.iter().any(|o| o.is_some()),
668                "well-formed message should decode to at least one frame"
669            );
670        }
671    }
672
673    #[test]
674    fn max_length() {
675        let mut codec = StreamCodec::stream_incoming().with_max_length(8);
676        let mut buf = BytesMut::from(&br#"{"a":1}  {"b":"#[..]);
677        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), r#"{"a":1}"#);
678        assert!(codec.decode(&mut buf).unwrap().is_none());
679        buf.put_slice(br#""long","c":["#);
680        assert!(codec.decode(&mut buf).unwrap().is_none());
681        buf.put_slice(br#"]}{"d":2}"#);
682        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), r#"{"b":"lon"#);
683        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), r#"{"d":2}"#);
684        assert!(buf.is_empty());
685
686        let mut buf = BytesMut::from(&br#"      {"a":"#[..]);
687        assert!(codec.decode(&mut buf).unwrap().is_none());
688        assert_eq!(buf, br#"{"a":"#[..]);
689        buf.put_slice(b"1}");
690        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), r#"{"a":1}"#);
691
692        let mut buf = BytesMut::from(&b"          "[..]);
693        assert_eq!(codec.decode(&mut buf).unwrap_err().kind(), io::ErrorKind::InvalidData);
694
695        let mut codec = StreamCodec::stream_incoming().with_max_length(8);
696        let mut buf = BytesMut::from(&b"}aaaaaaaa"[..]);
697        assert_eq!(codec.decode(&mut buf).unwrap_err().kind(), io::ErrorKind::InvalidData);
698
699        let mut codec = StreamCodec::default().with_max_length(4);
700        let mut buf = BytesMut::from(&b"abcd\nabcdef"[..]);
701        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), "abcd");
702        assert!(codec.decode(&mut buf).unwrap().is_none());
703        buf.put_slice(b"gh\nij\n");
704        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), "abcde");
705        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), "ij");
706
707        let mut buf = BytesMut::from(&b"\xff\xffcd\xe2\x82\xacgh"[..]);
708        assert!(codec.decode(&mut buf).unwrap().is_none());
709        buf.put_slice(b"\n");
710        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), "??cd?");
711    }
712
713    /// Random input under random chunk schedules never leaves more than `max_length` bytes
714    /// buffered between calls, in either framing mode.
715    #[test]
716    fn max_length_bounds_buffer() {
717        const MAX_CHUNK: usize = 16;
718
719        let mut rng = StdRng::seed_from_u64(0x00B0_0B5E_C0DE_C0DE);
720        let alphabet: &[u8] = b"{}[]\":,\\ \n\t a1\xff";
721
722        for _ in 0..200 {
723            let max_length = 1 + rng.next_u32() as usize % 16;
724            let mut codec = if rng.next_u32() % 2 == 0 {
725                StreamCodec::stream_incoming()
726            } else {
727                StreamCodec::default()
728            }
729            .with_max_length(max_length);
730            let mut buf = BytesMut::new();
731
732            'input: for _ in 0..rng.next_u32() % 64 {
733                for _ in 0..1 + rng.next_u32() as usize % MAX_CHUNK {
734                    buf.put_u8(alphabet[rng.next_u32() as usize % alphabet.len()]);
735                }
736                loop {
737                    match codec.decode(&mut buf) {
738                        Ok(Some(msg)) => assert!(msg.len() <= max_length + MAX_CHUNK),
739                        Ok(None) => break,
740                        Err(_) => break 'input,
741                    }
742                }
743                assert!(buf.len() <= max_length, "{} bytes buffered", buf.len());
744            }
745        }
746    }
747}
reth_ipc/stream_codec.rs

reth_ipc/
stream_codec.rs
