1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
use crate::sql_read_bytes::SqlReadBytes;
// Decode a partially length-prefixed type.
//
// NOTE: the length/chunk prefixes are read with the packet-aware
// `read_u16_le` (fixed-size total length), `read_u64_le` (PLP total-length
// field) and `read_u32_le` (per-chunk length) helpers, and the value bytes
// themselves via the packet-aware bulk reader `read_bytes_into`. All of these
// transparently span TDS packet boundaries. The generic
// `AsyncReadExt::read_exact` must NOT be used to read the value bytes here: a
// PLP value can span multiple packets, and `read_exact` treats a
// packet-boundary `Ok(0)` as EOF (truncating the value), whereas
// `read_bytes_into` continues across the boundary.
pub(crate) async fn decode<R>(src: &mut R, len: usize) -> crate::Result<Option<Vec<u8>>>
where
R: SqlReadBytes + Unpin,
{
match len {
// Fixed size
len if len < 0xffff => {
let len = src.read_u16_le().await? as usize;
match len {
// NULL
0xffff => Ok(None),
_ => {
let mut data = Vec::new();
crate::sql_read_bytes::read_bytes_into(
src,
&mut data,
len,
super::MAX_PREALLOC,
)
.await?;
Ok(Some(data))
}
}
}
// Unknown size, length-prefixed blobs
_ => {
let len = src.read_u64_le().await?;
let mut data = match len {
// NULL
0xffffffffffffffff => return Ok(None),
// Unknown size
0xfffffffffffffffe => Vec::new(),
// Known size. `len` is an untrusted 64-bit wire value; cap the
// up-front reservation (avoids memory-exhaustion and the
// `Vec` capacity-overflow panic for values near u64::MAX).
_ => Vec::with_capacity((len as usize).min(super::MAX_PREALLOC)),
};
loop {
let chunk_size = src.read_u32_le().await? as usize;
if chunk_size == 0 {
break; // found a sentinel, we're done
}
// The number of chunks in an "unknown length" PLP value is
// unbounded on the wire. Cap the running total so a hostile
// server cannot stream chunks forever and exhaust memory on
// a single value.
if data.len().saturating_add(chunk_size) > super::MAX_PLP_SIZE {
return Err(crate::Error::Protocol(
format!(
"PLP value exceeds the maximum supported size of {} bytes",
super::MAX_PLP_SIZE
)
.into(),
));
}
// Bulk-read the whole chunk (packet-aware), appending it to the
// running accumulation buffer.
crate::sql_read_bytes::read_bytes_into(
src,
&mut data,
chunk_size,
super::MAX_PREALLOC,
)
.await?;
}
Ok(Some(data))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::sql_read_bytes::test_utils::IntoSqlReadBytes;
use bytes::{BufMut, BytesMut};
// At the boundary `len == 0xffff` the decoder takes the chunked (PLP) branch,
// reading a `u64` length; the chunked stream decodes to [0xAA, 0xBB, 0xCC].
#[tokio::test]
async fn decode_boundary_len_uses_chunked_branch() {
let mut buf = BytesMut::new();
buf.put_u64_le(5); // known-size PLP total length
buf.put_u32_le(3); // chunk size
buf.put_slice(&[0xAA, 0xBB, 0xCC]);
buf.put_u32_le(0); // terminating chunk
let data = decode(&mut buf.into_sql_read_bytes(), 0xffff)
.await
.unwrap();
assert_eq!(data, Some(vec![0xAA, 0xBB, 0xCC]));
}
// A chunk whose running total strictly exceeds `MAX_PLP_SIZE` must be
// rejected with the "exceeds the maximum" protocol error before any chunk
// data is read.
#[tokio::test]
async fn decode_oversized_chunk_is_rejected() {
let mut buf = BytesMut::new();
buf.put_u64_le(10); // known-size marker -> chunked branch bookkeeping
buf.put_u32_le(u32::MAX); // chunk size well past MAX_PLP_SIZE
let err = decode(&mut buf.into_sql_read_bytes(), 0x10000)
.await
.expect_err("oversized PLP chunk must be rejected");
match err {
crate::Error::Protocol(msg) => {
assert!(
msg.contains("exceeds the maximum"),
"unexpected protocol message: {msg}"
);
}
other => panic!("expected a protocol error, got {other:?}"),
}
}
}