netget/server/fix/
codec.rs

1//! FIX tag=value framing (FIX 4.x and FIXT.1.1 session layer): BeginString, BodyLength and
2//! CheckSum are checked here, length-prefixed data fields may carry SOH, and repeated tags
3//! (repeating groups) keep their order.
4use anyhow::{bail, ensure, Context, Result};
5
6pub const SOH: u8 = 0x01;
7/// Largest BodyLength accepted, and the bound on buffered unparsed input.
8pub const MAX_BODY: usize = 256 * 1024;
9pub const MAX_FIELDS: usize = 2000;
10
11/// Length tag → the data tag it measures (FIX 4.4 data fields).
12const DATA_PAIRS: &[(u32, u32)] = &[
13    (90, 91),
14    (93, 89),
15    (95, 96),
16    (212, 213),
17    (348, 349),
18    (350, 351),
19    (352, 353),
20    (354, 355),
21    (356, 357),
22    (358, 359),
23    (360, 361),
24    (362, 363),
25    (364, 365),
26    (445, 446),
27    (618, 619),
28    (621, 622),
29];
30
31#[derive(Debug, Clone, PartialEq)]
32pub struct Message {
33    /// Every field in wire order, BeginString through CheckSum excluded.
34    pub fields: Vec<(u32, String)>,
35}
36
37impl Message {
38    pub fn get(&self, tag: u32) -> Option<&str> {
39        self.fields
40            .iter()
41            .find(|(t, _)| *t == tag)
42            .map(|(_, v)| v.as_str())
43    }
44    pub fn msg_type(&self) -> &str {
45        self.get(35).unwrap_or_default()
46    }
47    pub fn seq(&self) -> Option<u32> {
48        self.get(34).and_then(|s| s.parse().ok())
49    }
50}
51
52/// What the framer found at the front of the buffer.
53pub enum Frame {
54    /// Need more bytes.
55    Incomplete,
56    /// A whole message; `consumed` bytes may be dropped.
57    Message {
58        begin: String,
59        message: Message,
60        consumed: usize,
61    },
62    /// `consumed` bytes are garbled (FIX: ignore them and carry on).
63    Garbled { consumed: usize, reason: String },
64}
65
66fn find(hay: &[u8], needle: &[u8], from: usize) -> Option<usize> {
67    hay.get(from..)?
68        .windows(needle.len())
69        .position(|w| w == needle)
70        .map(|p| p + from)
71}
72
73pub fn checksum(bytes: &[u8]) -> u8 {
74    bytes.iter().fold(0u8, |a, b| a.wrapping_add(*b))
75}
76
77/// Frame one message from the front of `buf`.
78pub fn frame(buf: &[u8]) -> Frame {
79    if buf.is_empty() {
80        return Frame::Incomplete;
81    }
82    if !buf.starts_with(b"8=") {
83        // Resynchronise on the next BeginString after a field separator.
84        return match find(buf, b"\x018=", 0) {
85            Some(p) => Frame::Garbled {
86                consumed: p + 1,
87                reason: "bytes before BeginString".into(),
88            },
89            None if buf.len() > 1 => Frame::Garbled {
90                consumed: buf.len() - 1,
91                reason: "no BeginString".into(),
92            },
93            None => Frame::Incomplete,
94        };
95    }
96    let Some(e1) = buf.iter().position(|b| *b == SOH) else {
97        return if buf.len() > 32 {
98            Frame::Garbled {
99                consumed: 2,
100                reason: "BeginString too long".into(),
101            }
102        } else {
103            Frame::Incomplete
104        };
105    };
106    let begin = String::from_utf8_lossy(&buf[2..e1]).into_owned();
107    let rest = &buf[e1 + 1..];
108    if rest.len() < 2 {
109        return Frame::Incomplete;
110    }
111    if !rest.starts_with(b"9=") {
112        return Frame::Garbled {
113            consumed: e1 + 1,
114            reason: "BodyLength is not the second field".into(),
115        };
116    }
117    let Some(e2) = rest.iter().position(|b| *b == SOH) else {
118        return if rest.len() > 12 {
119            Frame::Garbled {
120                consumed: e1 + 1,
121                reason: "BodyLength too long".into(),
122            }
123        } else {
124            Frame::Incomplete
125        };
126    };
127    let len: usize = match std::str::from_utf8(&rest[2..e2])
128        .ok()
129        .filter(|s| !s.is_empty() && s.len() <= 7 && s.bytes().all(|b| b.is_ascii_digit()))
130        .and_then(|s| s.parse().ok())
131    {
132        Some(n) if n <= MAX_BODY => n,
133        _ => {
134            return Frame::Garbled {
135                consumed: e1 + 1 + e2 + 1,
136                reason: "BodyLength is not a number within bounds".into(),
137            }
138        }
139    };
140    let body_start = e1 + 1 + e2 + 1;
141    let total = body_start + len + 7;
142    if buf.len() < total {
143        return Frame::Incomplete;
144    }
145    let trailer = &buf[body_start + len..total];
146    if !trailer.starts_with(b"10=")
147        || trailer[6] != SOH
148        || !trailer[3..6].iter().all(u8::is_ascii_digit)
149    {
150        return Frame::Garbled {
151            consumed: body_start,
152            reason: "BodyLength does not end at CheckSum".into(),
153        };
154    }
155    let declared: u32 = std::str::from_utf8(&trailer[3..6])
156        .ok()
157        .and_then(|s| s.parse().ok())
158        .unwrap_or(1000);
159    if declared != checksum(&buf[..body_start + len]) as u32 {
160        return Frame::Garbled {
161            consumed: total,
162            reason: "CheckSum mismatch".into(),
163        };
164    }
165    match parse_fields(&buf[body_start..body_start + len]) {
166        Ok(fields) => Frame::Message {
167            begin,
168            message: Message { fields },
169            consumed: total,
170        },
171        Err(e) => Frame::Garbled {
172            consumed: total,
173            reason: format!("{e:#}"),
174        },
175    }
176}
177
178fn parse_fields(body: &[u8]) -> Result<Vec<(u32, String)>> {
179    let mut out: Vec<(u32, String)> = Vec::new();
180    let mut i = 0;
181    let mut data_len: Option<(u32, usize)> = None;
182    while i < body.len() {
183        ensure!(out.len() < MAX_FIELDS, "more than {MAX_FIELDS} fields");
184        let eq = body[i..]
185            .iter()
186            .position(|b| *b == b'=')
187            .context("field without =")?
188            + i;
189        let tag_s = std::str::from_utf8(&body[i..eq])
190            .ok()
191            .filter(|s| {
192                !s.is_empty()
193                    && s.len() <= 9
194                    && !s.starts_with('0')
195                    && s.bytes().all(|b| b.is_ascii_digit())
196            })
197            .context("invalid tag")?;
198        let tag: u32 = tag_s.parse()?;
199        let value = match data_len.take() {
200            Some((data_tag, n)) if data_tag == tag => {
201                let v = body
202                    .get(eq + 1..eq + 1 + n)
203                    .context("data field shorter than its length")?;
204                ensure!(
205                    body.get(eq + 1 + n) == Some(&SOH),
206                    "data field not followed by SOH"
207                );
208                i = eq + 1 + n + 1;
209                String::from_utf8_lossy(v).into_owned()
210            }
211            _ => {
212                let end = body[eq + 1..]
213                    .iter()
214                    .position(|b| *b == SOH)
215                    .context("field not terminated by SOH")?
216                    + eq
217                    + 1;
218                let v = &body[eq + 1..end];
219                ensure!(!v.is_empty(), "tag {tag} has an empty value");
220                i = end + 1;
221                String::from_utf8_lossy(v).into_owned()
222            }
223        };
224        if let Some((_, data_tag)) = DATA_PAIRS.iter().find(|(l, _)| *l == tag) {
225            let n: usize = value
226                .parse()
227                .ok()
228                .filter(|n| *n <= MAX_BODY)
229                .context("invalid data length")?;
230            data_len = Some((*data_tag, n));
231        }
232        out.push((tag, value));
233    }
234    ensure!(
235        !out.is_empty() && out[0].0 == 35,
236        "MsgType is not the third field"
237    );
238    Ok(out)
239}
240
241/// Encode a message: BeginString, BodyLength, `fields` (MsgType first), CheckSum.
242pub fn encode(begin: &str, fields: &[(u32, String)]) -> Result<Vec<u8>> {
243    let mut body = Vec::new();
244    for (tag, value) in fields {
245        if value.as_bytes().contains(&SOH) && !DATA_PAIRS.iter().any(|(_, d)| d == tag) {
246            bail!("tag {tag} contains SOH");
247        }
248        body.extend_from_slice(format!("{tag}=").as_bytes());
249        body.extend_from_slice(value.as_bytes());
250        body.push(SOH);
251    }
252    ensure!(body.len() <= MAX_BODY, "message body over {MAX_BODY} bytes");
253    let mut out = format!("8={begin}\x019={}\x01", body.len()).into_bytes();
254    out.extend(body);
255    let sum = checksum(&out);
256    out.extend_from_slice(format!("10={sum:03}\x01").as_bytes());
257    Ok(out)
258}
259
260/// FIX UTCTimestamp with milliseconds.
261pub fn timestamp() -> String {
262    let now = time::OffsetDateTime::now_utc();
263    format!(
264        "{:04}{:02}{:02}-{:02}:{:02}:{:02}.{:03}",
265        now.year(),
266        now.month() as u8,
267        now.day(),
268        now.hour(),
269        now.minute(),
270        now.second(),
271        now.millisecond()
272    )
273}
netget/server/fix/codec.rs

netget/server/fix/
codec.rs
