ocular_protocol/
mongodb.rs

1//! MongoDB wire protocol parser (OP_MSG only, modern MongoDB 3.6+)
2//! All integers are little-endian.
3
4const OP_MSG: i32 = 2013;
5const OP_COMPRESSED: i32 = 2012;
6
7/// Get the total message length from a MongoDB wire protocol header.
8/// Returns None if buffer is too small or length is invalid.
9pub fn mongo_msg_len(buf: &[u8]) -> Option<usize> {
10    if buf.len() < 4 { return None; }
11    let len = i32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
12    if !(16..=48 * 1024 * 1024).contains(&len) { return None; }
13    Some(len)
14}
15
16/// Parse a MongoDB request (client→server), returning a command summary.
17pub fn parse_mongo_request(buf: &[u8]) -> Option<String> {
18    let doc = extract_body_doc(buf)?;
19    let cmd = first_key(&doc)?;
20    let db = get_string_field(&doc, "$db").unwrap_or_default();
21    let detail = match cmd.as_str() {
22        "find" => {
23            let coll = get_string_field(&doc, "find").unwrap_or_default();
24            let filter = get_doc_field_summary(&doc, "filter");
25            format!("find {}.{} {}", db, coll, filter)
26        }
27        "insert" => {
28            let coll = get_string_field(&doc, "insert").unwrap_or_default();
29            let n = get_array_len(&doc, "documents");
30            format!("insert {}.{} ({} docs)", db, coll, n)
31        }
32        "update" => {
33            let coll = get_string_field(&doc, "update").unwrap_or_default();
34            let n = get_array_len(&doc, "updates");
35            format!("update {}.{} ({} ops)", db, coll, n)
36        }
37        "delete" => {
38            let coll = get_string_field(&doc, "delete").unwrap_or_default();
39            let n = get_array_len(&doc, "deletes");
40            format!("delete {}.{} ({} ops)", db, coll, n)
41        }
42        "aggregate" => {
43            let coll = get_string_field(&doc, "aggregate").unwrap_or_default();
44            format!("aggregate {}.{}", db, coll)
45        }
46        "getMore" => {
47            let coll = get_string_field(&doc, "collection").unwrap_or_default();
48            format!("getMore {}.{}", db, coll)
49        }
50        _ => {
51            if db.is_empty() { cmd.clone() } else { format!("{} {}", cmd, db) }
52        }
53    };
54    Some(detail)
55}
56
57/// Extract full command detail (for Detail panel) — mongosh-style replayable statements.
58pub fn extract_mongo_full_command(buf: &[u8]) -> Option<String> {
59    let doc = extract_body_doc(buf)?;
60    let cmd = first_key(&doc)?;
61    let db = get_string_field(&doc, "$db").unwrap_or_default();
62    match cmd.as_str() {
63        "find" => {
64            let coll = get_string_field(&doc, "find").unwrap_or_default();
65            let filter = get_doc_field_summary(&doc, "filter");
66            let limit = get_i32_field(&doc, "limit");
67            let sort = get_raw_doc_field(&doc, "sort").map(|d| bson_doc_to_json_like(&d));
68            let mut s = format!("db.{}.find({})", coll, filter);
69            if let Some(sort_str) = sort { s.push_str(&format!(".sort({})", sort_str)); }
70            if let Some(l) = limit { s.push_str(&format!(".limit({})", l)); }
71            Some(s)
72        }
73        "insert" => {
74            let coll = get_string_field(&doc, "insert").unwrap_or_default();
75            let docs = get_array_docs(&doc, "documents");
76            if docs.len() == 1 {
77                Some(format!("db.{}.insertOne({})", coll, bson_doc_to_json_like(&docs[0])))
78            } else {
79                let items: Vec<String> = docs.iter().take(10).map(|d| bson_doc_to_json_like(d)).collect();
80                let mut s = format!("db.{}.insertMany([{}])", coll, items.join(", "));
81                if docs.len() > 10 { s.push_str(&format!(" // +{} more", docs.len() - 10)); }
82                Some(s)
83            }
84        }
85        "update" => {
86            let coll = get_string_field(&doc, "update").unwrap_or_default();
87            let updates = get_array_docs(&doc, "updates");
88            if updates.len() == 1 {
89                let q = get_doc_field_summary(&updates[0], "q");
90                let u = get_doc_field_summary(&updates[0], "u");
91                let multi = get_i32_field(&updates[0], "multi").unwrap_or(0) != 0
92                    || has_field(&updates[0], "multi") && get_f64_field(&updates[0], "multi") == Some(1.0);
93                let method = if multi { "updateMany" } else { "updateOne" };
94                Some(format!("db.{}.{}({}, {})", coll, method, q, u))
95            } else {
96                Some(format!("db.{}.bulkWrite([...{} ops])", coll, updates.len()))
97            }
98        }
99        "delete" => {
100            let coll = get_string_field(&doc, "delete").unwrap_or_default();
101            let deletes = get_array_docs(&doc, "deletes");
102            if deletes.len() == 1 {
103                let q = get_doc_field_summary(&deletes[0], "q");
104                let limit = get_i32_field(&deletes[0], "limit").unwrap_or(0);
105                let method = if limit == 1 { "deleteOne" } else { "deleteMany" };
106                Some(format!("db.{}.{}({})", coll, method, q))
107            } else {
108                Some(format!("db.{}.bulkWrite([...{} ops])", coll, deletes.len()))
109            }
110        }
111        "aggregate" => {
112            let coll = get_string_field(&doc, "aggregate").unwrap_or_default();
113            Some(format!("db.{}.aggregate([...])", coll))
114        }
115        "findAndModify" => {
116            let coll = get_string_field(&doc, "findAndModify").unwrap_or_default();
117            let query = get_doc_field_summary(&doc, "query");
118            let update = get_doc_field_summary(&doc, "update");
119            Some(format!("db.{}.findOneAndUpdate({}, {})", coll, query, update))
120        }
121        "count" | "countDocuments" => {
122            let coll = get_string_field(&doc, &cmd).unwrap_or_default();
123            let query = get_doc_field_summary(&doc, "query");
124            Some(format!("db.{}.countDocuments({})", coll, query))
125        }
126        _ => {
127            if db.is_empty() { Some(cmd) } else { Some(format!("{} {}", cmd, db)) }
128        }
129    }
130}
131
132/// Parse a MongoDB response (server→client), returning a summary.
133pub fn parse_mongo_response(buf: &[u8]) -> Option<String> {
134    let doc = extract_body_doc(buf)?;
135    let ok = get_f64_field(&doc, "ok");
136    if ok == Some(0.0) {
137        let errmsg = get_string_field(&doc, "errmsg").unwrap_or("error".into());
138        let code = get_i32_field(&doc, "code").map(|c| format!(" ({})", c)).unwrap_or_default();
139        return Some(format!("ERR{} {}", code, errmsg));
140    }
141    // Check for cursor result
142    if let Some(cursor_doc) = get_raw_doc_field(&doc, "cursor") {
143        let batch_key = if has_field(&cursor_doc, "firstBatch") { "firstBatch" } else { "nextBatch" };
144        let n = get_array_len(&cursor_doc, batch_key);
145        return Some(format!("OK ({} docs)", n));
146    }
147    // Check for n (insert/update/delete result)
148    if let Some(n) = get_i32_field(&doc, "n") {
149        let modified = get_i32_field(&doc, "nModified");
150        if let Some(m) = modified {
151            return Some(format!("OK (n={}, modified={})", n, m));
152        }
153        return Some(format!("OK (n={})", n));
154    }
155    Some("OK".into())
156}
157
158/// Format detailed response for the detail panel.
159pub fn format_mongo_response_detail(buf: &[u8]) -> Option<String> {
160    let doc = extract_body_doc(buf)?;
161    let ok = get_f64_field(&doc, "ok");
162    if ok == Some(0.0) {
163        let errmsg = get_string_field(&doc, "errmsg").unwrap_or("error".into());
164        let code = get_i32_field(&doc, "code").unwrap_or(0);
165        let codename = get_string_field(&doc, "codeName").unwrap_or_default();
166        return Some(format!("ERROR {} ({}): {}", code, codename, errmsg));
167    }
168    if let Some(cursor_doc) = get_raw_doc_field(&doc, "cursor") {
169        let batch_key = if has_field(&cursor_doc, "firstBatch") { "firstBatch" } else { "nextBatch" };
170        let docs = get_array_docs(&cursor_doc, batch_key);
171        let mut lines = Vec::new();
172        lines.push(format!("{} documents:", docs.len()));
173        for (i, d) in docs.iter().enumerate().take(20) {
174            lines.push(format!("  [{}] {}", i, bson_doc_to_json_like(d)));
175        }
176        if docs.len() > 20 {
177            lines.push(format!("  ... ({} more)", docs.len() - 20));
178        }
179        return Some(lines.join("\n"));
180    }
181    parse_mongo_response(buf)
182}
183
184// --- Internal helpers ---
185
186/// Extract the Kind 0 body BSON document from an OP_MSG.
187fn extract_body_doc(buf: &[u8]) -> Option<Vec<u8>> {
188    if buf.len() < 21 { return None; } // header(16) + flags(4) + kind(1)
189    let opcode = i32::from_le_bytes([buf[12], buf[13], buf[14], buf[15]]);
190    if opcode != OP_MSG && opcode != OP_COMPRESSED { return None; }
191    if opcode == OP_COMPRESSED { return decompress_op_compressed(buf); }
192    // flags at offset 16, sections start at offset 20
193    let mut pos = 20;
194    while pos < buf.len() {
195        let kind = buf[pos];
196        pos += 1;
197        if kind == 0 {
198            // Kind 0: single BSON document
199            if pos + 4 > buf.len() { return None; }
200            let doc_len = i32::from_le_bytes([buf[pos], buf[pos+1], buf[pos+2], buf[pos+3]]) as usize;
201            if pos + doc_len > buf.len() { return None; }
202            return Some(buf[pos..pos+doc_len].to_vec());
203        } else if kind == 1 {
204            // Kind 1: document sequence, skip
205            if pos + 4 > buf.len() { return None; }
206            let sec_len = i32::from_le_bytes([buf[pos], buf[pos+1], buf[pos+2], buf[pos+3]]) as usize;
207            pos += sec_len;
208        } else {
209            break;
210        }
211    }
212    None
213}
214
215/// Decompress an OP_COMPRESSED message and extract the body doc from the inner OP_MSG.
216fn decompress_op_compressed(buf: &[u8]) -> Option<Vec<u8>> {
217    if buf.len() < 25 { return None; }
218    let original_opcode = i32::from_le_bytes([buf[16], buf[17], buf[18], buf[19]]);
219    if original_opcode != OP_MSG { return None; }
220    let uncompressed_size = i32::from_le_bytes([buf[20], buf[21], buf[22], buf[23]]) as usize;
221    let compressor_id = buf[24];
222    let compressed = &buf[25..];
223
224    let decompressed = match compressor_id {
225        0 => compressed.to_vec(),
226        1 => snap::raw::Decoder::new().decompress_vec(compressed).ok()?,
227        2 => {
228            use std::io::Read;
229            let mut decoder = flate2::read::ZlibDecoder::new(compressed);
230            let mut out = Vec::with_capacity(uncompressed_size);
231            decoder.read_to_end(&mut out).ok()?;
232            out
233        }
234        3 => zstd::decode_all(compressed).ok()?,
235        _ => return None,
236    };
237
238    // decompressed = flags(4) + sections... (OP_MSG body without 16-byte header)
239    if decompressed.len() < 5 { return None; }
240    let mut pos = 4; // skip flags
241    while pos < decompressed.len() {
242        let kind = decompressed[pos];
243        pos += 1;
244        if kind == 0 {
245            if pos + 4 > decompressed.len() { return None; }
246            let doc_len = i32::from_le_bytes([decompressed[pos], decompressed[pos+1], decompressed[pos+2], decompressed[pos+3]]) as usize;
247            if pos + doc_len > decompressed.len() { return None; }
248            return Some(decompressed[pos..pos+doc_len].to_vec());
249        } else if kind == 1 {
250            if pos + 4 > decompressed.len() { return None; }
251            let sec_len = i32::from_le_bytes([decompressed[pos], decompressed[pos+1], decompressed[pos+2], decompressed[pos+3]]) as usize;
252            pos += sec_len;
253        } else {
254            break;
255        }
256    }
257    None
258}
259
260/// Get the first key name from a BSON document (the command name).
261fn first_key(doc: &[u8]) -> Option<String> {
262    if doc.len() < 6 { return None; }
263    // doc[0..4] = size, doc[4] = element type, doc[5..] = cstring key
264    let key = read_cstr(&doc[5..])?;
265    Some(key)
266}
267
268/// Read a null-terminated C string.
269fn read_cstr(buf: &[u8]) -> Option<String> {
270    let end = buf.iter().position(|&b| b == 0)?;
271    Some(String::from_utf8_lossy(&buf[..end]).to_string())
272}
273
274/// Get a string field value from a BSON document.
275fn get_string_field(doc: &[u8], name: &str) -> Option<String> {
276    let mut pos = 4; // skip doc size
277    while pos < doc.len() - 1 {
278        let etype = doc[pos];
279        if etype == 0 { break; } // end of doc
280        pos += 1;
281        let key = read_cstr(&doc[pos..])?;
282        pos += key.len() + 1;
283        match etype {
284            0x02 => { // string
285                if pos + 4 > doc.len() { return None; }
286                let slen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize;
287                pos += 4;
288                if key == name {
289                    let s = String::from_utf8_lossy(&doc[pos..pos+slen.saturating_sub(1)]).to_string();
290                    return Some(s);
291                }
292                pos += slen;
293            }
294            0x01 => { pos += 8; } // double
295            0x03 | 0x04 => { // document or array
296                if pos + 4 > doc.len() { return None; }
297                let dlen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize;
298                pos += dlen;
299            }
300            0x05 => { // binary
301                if pos + 4 > doc.len() { return None; }
302                let blen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize;
303                pos += 5 + blen;
304            }
305            0x07 => { pos += 12; } // ObjectId
306            0x08 => { pos += 1; } // boolean
307            0x09 | 0x11 | 0x12 => { pos += 8; } // datetime, timestamp, int64
308            0x0A => {} // null
309            0x10 => { pos += 4; } // int32
310            0x13 => { pos += 16; } // decimal128
311            _ => { return None; } // unknown type, bail
312        }
313    }
314    None
315}
316
317fn get_f64_field(doc: &[u8], name: &str) -> Option<f64> {
318    let mut pos = 4;
319    while pos < doc.len() - 1 {
320        let etype = doc[pos];
321        if etype == 0 { break; }
322        pos += 1;
323        let key = read_cstr(&doc[pos..])?;
324        pos += key.len() + 1;
325        match etype {
326            0x01 => {
327                if key == name && pos + 8 <= doc.len() {
328                    return Some(f64::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3], doc[pos+4], doc[pos+5], doc[pos+6], doc[pos+7]]));
329                }
330                pos += 8;
331            }
332            0x10 => {
333                if key == name && pos + 4 <= doc.len() {
334                    let v = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]);
335                    return Some(v as f64);
336                }
337                pos += 4;
338            }
339            0x02 => { if pos + 4 > doc.len() { return None; } let slen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 4 + slen; }
340            0x03 | 0x04 => { if pos + 4 > doc.len() { return None; } let dlen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += dlen; }
341            0x05 => { if pos + 4 > doc.len() { return None; } let blen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 5 + blen; }
342            0x07 => { pos += 12; }
343            0x08 => { pos += 1; }
344            0x09 | 0x11 | 0x12 => { pos += 8; }
345            0x0A => {}
346            0x13 => { pos += 16; }
347            _ => { return None; }
348        }
349    }
350    None
351}
352
353fn get_i32_field(doc: &[u8], name: &str) -> Option<i32> {
354    let mut pos = 4;
355    while pos < doc.len() - 1 {
356        let etype = doc[pos];
357        if etype == 0 { break; }
358        pos += 1;
359        let key = read_cstr(&doc[pos..])?;
360        pos += key.len() + 1;
361        match etype {
362            0x10 => {
363                if key == name && pos + 4 <= doc.len() {
364                    return Some(i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]));
365                }
366                pos += 4;
367            }
368            0x01 => { pos += 8; }
369            0x02 => { if pos + 4 > doc.len() { return None; } let slen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 4 + slen; }
370            0x03 | 0x04 => { if pos + 4 > doc.len() { return None; } let dlen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += dlen; }
371            0x05 => { if pos + 4 > doc.len() { return None; } let blen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 5 + blen; }
372            0x07 => { pos += 12; }
373            0x08 => { pos += 1; }
374            0x09 | 0x11 | 0x12 => { pos += 8; }
375            0x0A => {}
376            0x13 => { pos += 16; }
377            _ => { return None; }
378        }
379    }
380    None
381}
382
383fn get_raw_doc_field(doc: &[u8], name: &str) -> Option<Vec<u8>> {
384    let mut pos = 4;
385    while pos < doc.len() - 1 {
386        let etype = doc[pos];
387        if etype == 0 { break; }
388        pos += 1;
389        let key = read_cstr(&doc[pos..])?;
390        pos += key.len() + 1;
391        match etype {
392            0x03 | 0x04 => {
393                if pos + 4 > doc.len() { return None; }
394                let dlen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize;
395                if key == name {
396                    return Some(doc[pos..pos+dlen].to_vec());
397                }
398                pos += dlen;
399            }
400            0x01 => { pos += 8; }
401            0x02 => { if pos + 4 > doc.len() { return None; } let slen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 4 + slen; }
402            0x05 => { if pos + 4 > doc.len() { return None; } let blen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 5 + blen; }
403            0x07 => { pos += 12; }
404            0x08 => { pos += 1; }
405            0x09 | 0x10 | 0x11 | 0x12 => { pos += if etype == 0x10 { 4 } else { 8 }; }
406            0x0A => {}
407            0x13 => { pos += 16; }
408            _ => { return None; }
409        }
410    }
411    None
412}
413
414fn has_field(doc: &[u8], name: &str) -> bool {
415    let mut pos = 4;
416    while pos < doc.len().saturating_sub(1) {
417        let etype = doc[pos];
418        if etype == 0 { break; }
419        pos += 1;
420        let Some(key) = read_cstr(&doc[pos..]) else { break };
421        if key == name { return true; }
422        pos += key.len() + 1;
423        match etype {
424            0x01 => { pos += 8; }
425            0x02 => { if pos + 4 > doc.len() { break; } let slen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 4 + slen; }
426            0x03 | 0x04 => { if pos + 4 > doc.len() { break; } let dlen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += dlen; }
427            0x05 => { if pos + 4 > doc.len() { break; } let blen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 5 + blen; }
428            0x07 => { pos += 12; }
429            0x08 => { pos += 1; }
430            0x09 | 0x11 | 0x12 => { pos += 8; }
431            0x0A => {}
432            0x10 => { pos += 4; }
433            0x13 => { pos += 16; }
434            _ => { break; }
435        }
436    }
437    false
438}
439
440fn get_array_len(doc: &[u8], name: &str) -> usize {
441    let Some(arr) = get_raw_doc_field(doc, name) else { return 0 };
442    // BSON array is a document with "0", "1", ... keys
443    let mut count = 0;
444    let mut pos = 4;
445    while pos < arr.len().saturating_sub(1) {
446        if arr[pos] == 0 { break; }
447        count += 1;
448        pos += 1;
449        let Some(key) = read_cstr(&arr[pos..]) else { break };
450        pos += key.len() + 1;
451        // skip value based on type
452        let etype = arr[pos - key.len() - 2];
453        match etype {
454            0x01 => { pos += 8; }
455            0x02 => { if pos + 4 > arr.len() { break; } let slen = i32::from_le_bytes([arr[pos], arr[pos+1], arr[pos+2], arr[pos+3]]) as usize; pos += 4 + slen; }
456            0x03 | 0x04 => { if pos + 4 > arr.len() { break; } let dlen = i32::from_le_bytes([arr[pos], arr[pos+1], arr[pos+2], arr[pos+3]]) as usize; pos += dlen; }
457            0x05 => { if pos + 4 > arr.len() { break; } let blen = i32::from_le_bytes([arr[pos], arr[pos+1], arr[pos+2], arr[pos+3]]) as usize; pos += 5 + blen; }
458            0x07 => { pos += 12; }
459            0x08 => { pos += 1; }
460            0x09 | 0x11 | 0x12 => { pos += 8; }
461            0x0A => {}
462            0x10 => { pos += 4; }
463            0x13 => { pos += 16; }
464            _ => { break; }
465        }
466    }
467    count
468}
469
470fn get_array_docs(doc: &[u8], name: &str) -> Vec<Vec<u8>> {
471    let Some(arr) = get_raw_doc_field(doc, name) else { return vec![] };
472    let mut docs = Vec::new();
473    let mut pos = 4;
474    while pos < arr.len().saturating_sub(1) {
475        let etype = arr[pos];
476        if etype == 0 { break; }
477        pos += 1;
478        let Some(key) = read_cstr(&arr[pos..]) else { break };
479        pos += key.len() + 1;
480        if etype == 0x03 {
481            if pos + 4 > arr.len() { break; }
482            let dlen = i32::from_le_bytes([arr[pos], arr[pos+1], arr[pos+2], arr[pos+3]]) as usize;
483            if pos + dlen <= arr.len() {
484                docs.push(arr[pos..pos+dlen].to_vec());
485            }
486            pos += dlen;
487        } else {
488            break; // unexpected type in result array
489        }
490    }
491    docs
492}
493
494fn get_doc_field_summary(doc: &[u8], name: &str) -> String {
495    let Some(subdoc) = get_raw_doc_field(doc, name) else { return "{}".into() };
496    bson_doc_to_json_like(&subdoc)
497}
498
499/// Simple BSON doc to JSON-like string (for display, not full fidelity).
500fn bson_doc_to_json_like(doc: &[u8]) -> String {
501    let mut parts = Vec::new();
502    let mut pos = 4;
503    while pos < doc.len().saturating_sub(1) {
504        let etype = doc[pos];
505        if etype == 0 { break; }
506        pos += 1;
507        let Some(key) = read_cstr(&doc[pos..]) else { break };
508        pos += key.len() + 1;
509        let val = match etype {
510            0x01 => { let v = if pos + 8 <= doc.len() { f64::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3], doc[pos+4], doc[pos+5], doc[pos+6], doc[pos+7]]) } else { 0.0 }; pos += 8; format!("{}", v) }
511            0x02 => { if pos + 4 > doc.len() { break; } let slen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += 4; let s = String::from_utf8_lossy(&doc[pos..pos+slen.saturating_sub(1)]).to_string(); pos += slen; format!("\"{}\"", s) }
512            0x03 => { if pos + 4 > doc.len() { break; } let dlen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; let s = bson_doc_to_json_like(&doc[pos..pos+dlen]); pos += dlen; s }
513            0x04 => { if pos + 4 > doc.len() { break; } let dlen = i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) as usize; pos += dlen; "[...]".into() }
514            0x07 => { pos += 12; "ObjectId(...)".into() }
515            0x08 => { let v = doc[pos] != 0; pos += 1; format!("{}", v) }
516            0x09 => { pos += 8; "Date(...)".into() }
517            0x0A => { "null".into() }
518            0x10 => { let v = if pos + 4 <= doc.len() { i32::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3]]) } else { 0 }; pos += 4; format!("{}", v) }
519            0x12 => { let v = if pos + 8 <= doc.len() { i64::from_le_bytes([doc[pos], doc[pos+1], doc[pos+2], doc[pos+3], doc[pos+4], doc[pos+5], doc[pos+6], doc[pos+7]]) } else { 0 }; pos += 8; format!("{}", v) }
520            _ => { break; }
521        };
522        if key == "_id" || key == "lsid" { continue; }
523        parts.push(format!("{}: {}", key, val));
524        if parts.len() >= 8 { parts.push("...".into()); break; }
525    }
526    format!("{{{}}}", parts.join(", "))
527}
528
529#[cfg(test)]
530mod tests {
531    use super::*;
532
533    /// Build a minimal OP_MSG with a Kind 0 BSON body document.
534    fn build_op_msg(doc: &[u8]) -> Vec<u8> {
535        let msg_len = 16 + 4 + 1 + doc.len(); // header + flags + kind + doc
536        let mut buf = Vec::new();
537        buf.extend_from_slice(&(msg_len as i32).to_le_bytes()); // messageLength
538        buf.extend_from_slice(&1i32.to_le_bytes()); // requestID
539        buf.extend_from_slice(&0i32.to_le_bytes()); // responseTo
540        buf.extend_from_slice(&OP_MSG.to_le_bytes()); // opCode
541        buf.extend_from_slice(&0u32.to_le_bytes()); // flagBits
542        buf.push(0); // kind 0
543        buf.extend_from_slice(doc);
544        buf
545    }
546
547    /// Build a simple BSON document: {"cmd": "coll", "$db": "testdb"}
548    fn build_simple_cmd(cmd: &str, coll: &str) -> Vec<u8> {
549        let mut doc = Vec::new();
550        doc.extend_from_slice(&[0; 4]); // placeholder for size
551        // cmd: coll (string)
552        doc.push(0x02); // string type
553        doc.extend_from_slice(cmd.as_bytes());
554        doc.push(0);
555        let val = format!("{}\0", coll);
556        doc.extend_from_slice(&(val.len() as i32).to_le_bytes());
557        doc.extend_from_slice(val.as_bytes());
558        // $db: "testdb" (string)
559        doc.push(0x02);
560        doc.extend_from_slice(b"$db\0");
561        let db = "testdb\0";
562        doc.extend_from_slice(&(db.len() as i32).to_le_bytes());
563        doc.extend_from_slice(db.as_bytes());
564        // end
565        doc.push(0);
566        let len = doc.len() as i32;
567        doc[0..4].copy_from_slice(&len.to_le_bytes());
568        doc
569    }
570
571    #[test]
572    fn test_parse_find_request() {
573        let doc = build_simple_cmd("find", "users");
574        let buf = build_op_msg(&doc);
575        let result = parse_mongo_request(&buf).unwrap();
576        assert!(result.contains("find"));
577        assert!(result.contains("testdb"));
578        assert!(result.contains("users"));
579    }
580
581    #[test]
582    fn test_parse_insert_request() {
583        let doc = build_simple_cmd("insert", "users");
584        let buf = build_op_msg(&doc);
585        let result = parse_mongo_request(&buf).unwrap();
586        assert!(result.contains("insert"));
587        assert!(result.contains("testdb.users"));
588    }
589
590    #[test]
591    fn test_parse_response_ok() {
592        // {"ok": 1.0}
593        let mut doc = Vec::new();
594        doc.extend_from_slice(&[0; 4]);
595        doc.push(0x01); // double
596        doc.extend_from_slice(b"ok\0");
597        doc.extend_from_slice(&1.0f64.to_le_bytes());
598        doc.push(0);
599        let len = doc.len() as i32;
600        doc[0..4].copy_from_slice(&len.to_le_bytes());
601
602        let buf = build_op_msg(&doc);
603        let result = parse_mongo_response(&buf).unwrap();
604        assert_eq!(result, "OK");
605    }
606
607    #[test]
608    fn test_parse_response_error() {
609        // {"ok": 0.0, "errmsg": "not found", "code": 26}
610        let mut doc = Vec::new();
611        doc.extend_from_slice(&[0; 4]);
612        // ok: 0.0
613        doc.push(0x01);
614        doc.extend_from_slice(b"ok\0");
615        doc.extend_from_slice(&0.0f64.to_le_bytes());
616        // errmsg: "not found"
617        doc.push(0x02);
618        doc.extend_from_slice(b"errmsg\0");
619        let msg = "not found\0";
620        doc.extend_from_slice(&(msg.len() as i32).to_le_bytes());
621        doc.extend_from_slice(msg.as_bytes());
622        // code: 26
623        doc.push(0x10);
624        doc.extend_from_slice(b"code\0");
625        doc.extend_from_slice(&26i32.to_le_bytes());
626        doc.push(0);
627        let len = doc.len() as i32;
628        doc[0..4].copy_from_slice(&len.to_le_bytes());
629
630        let buf = build_op_msg(&doc);
631        let result = parse_mongo_response(&buf).unwrap();
632        assert!(result.contains("ERR"));
633        assert!(result.contains("26"));
634        assert!(result.contains("not found"));
635    }
636
637    #[test]
638    fn test_mongo_msg_len() {
639        let buf = build_op_msg(&build_simple_cmd("ping", "admin"));
640        assert_eq!(mongo_msg_len(&buf), Some(buf.len()));
641    }
642
643    #[test]
644    fn test_mongo_msg_len_too_short() {
645        assert_eq!(mongo_msg_len(&[1, 2, 3]), None);
646    }
647
648    #[test]
649    fn test_extract_full_command_find() {
650        let doc = build_simple_cmd("find", "users");
651        let buf = build_op_msg(&doc);
652        let result = extract_mongo_full_command(&buf).unwrap();
653        assert!(result.contains("db.users.find"));
654    }
655}
ocular_protocol/mongodb.rs

ocular_protocol/
mongodb.rs
