tako_rs_core/grpc/
framing.rs

1//! gRPC length-prefix framing: the message-size cap, encode/decode of a
2//! single `[compressed][length][bytes]` frame, and the `GrpcError` type those
3//! operations surface.
4
5use prost::Message;
6
7use super::status::GrpcStatusCode;
8use super::status::build_grpc_error_response;
9use crate::responder::Responder;
10use crate::types::Response;
11
12/// Cap on the `length` prefix of a single gRPC frame. Without it any client
13/// can advertise a 4 GiB message and force the parser to either pre-allocate
14/// that much space or treat the body as well-formed-but-truncated. 4 MiB
15/// matches the default `grpc-go` and `tonic` server limits.
16pub const MAX_GRPC_MESSAGE_SIZE: usize = 4 * 1024 * 1024;
17
18/// Error types for gRPC extraction.
19#[derive(Debug)]
20pub enum GrpcError {
21  /// Content-Type is not application/grpc.
22  InvalidContentType,
23  /// Failed to read the request body.
24  BodyReadError(String),
25  /// gRPC frame is too short or malformed.
26  InvalidFrame,
27  /// Length-prefix advertises a message larger than [`MAX_GRPC_MESSAGE_SIZE`].
28  ///
29  /// Mapped to gRPC status `ResourceExhausted` (8) per the spec — `grpc-go`,
30  /// `tonic`, and the upstream issue (grpc/grpc#23454) all use it for
31  /// `received message larger than max`. Returning `InvalidArgument` would
32  /// be wire-level wrong: clients that backoff-retry on `ResourceExhausted`
33  /// would never retry on `InvalidArgument`.
34  MessageTooLarge,
35  /// Protobuf decoding failed.
36  DecodeError(String),
37  /// Frame's compressed flag was set but the server does not advertise
38  /// any compression codec. Mapped to gRPC status `Unimplemented` per
39  /// the spec (<https://grpc.io/docs/guides/wire>/) so clients fall back
40  /// to uncompressed.
41  CompressionUnsupported,
42}
43
44impl Responder for GrpcError {
45  fn into_response(self) -> Response {
46    let (status_code, message) = match self {
47      GrpcError::InvalidContentType => (
48        // Spec maps wrong/missing content-type to `Unimplemented` (12) —
49        // see PROTOCOL-HTTP2.md ("If Content-Type does not begin with
50        // 'application/grpc', gRPC servers SHOULD respond with HTTP
51        // status of 415 (Unsupported Media Type)"). grpcurl/Envoy
52        // route on this distinction; `InvalidArgument` would suggest
53        // a request-payload bug instead of an unsupported protocol.
54        GrpcStatusCode::Unimplemented,
55        "invalid content-type; expected application/grpc",
56      ),
57      GrpcError::BodyReadError(_) => (GrpcStatusCode::Internal, "failed to read request body"),
58      GrpcError::InvalidFrame => (GrpcStatusCode::InvalidArgument, "malformed gRPC frame"),
59      GrpcError::MessageTooLarge => (
60        GrpcStatusCode::ResourceExhausted,
61        "grpc message exceeds MAX_GRPC_MESSAGE_SIZE",
62      ),
63      GrpcError::DecodeError(_) => (
64        GrpcStatusCode::InvalidArgument,
65        "failed to decode protobuf message",
66      ),
67      GrpcError::CompressionUnsupported => (
68        GrpcStatusCode::Unimplemented,
69        "frame is compressed but no codec is configured",
70      ),
71    };
72
73    build_grpc_error_response(status_code, message)
74  }
75}
76
77/// Encode a protobuf message with gRPC length-prefix framing.
78///
79/// Format: `[compressed: u8][length: u32 BE][message bytes]`
80///
81/// # Panics
82///
83/// Panics if the encoded message exceeds `u32::MAX` (≈ 4 GiB). gRPC's wire
84/// format uses a 4-byte big-endian length prefix, so anything larger would
85/// silently wrap to a wrong length and produce undecodable frames. The assert
86/// turns that silent corruption into a loud server-side crash with a clear
87/// site. (Outbound messages this large already indicate a serious
88/// memory-pressure problem in the calling handler.)
89pub fn grpc_encode<T: Message>(msg: &T) -> Vec<u8> {
90  let msg_bytes = msg.encode_to_vec();
91  assert!(
92    u32::try_from(msg_bytes.len()).is_ok(),
93    "grpc_encode: message of {} bytes exceeds u32::MAX (4 GiB) — gRPC length-prefix would wrap",
94    msg_bytes.len()
95  );
96  let len = msg_bytes.len() as u32;
97
98  let mut frame = Vec::with_capacity(5 + msg_bytes.len());
99  frame.push(0); // not compressed
100  frame.extend_from_slice(&len.to_be_bytes());
101  frame.extend_from_slice(&msg_bytes);
102  frame
103}
104
105/// Decode a gRPC length-prefix framed message.
106///
107/// Returns the decoded message and whether compression was indicated.
108pub fn grpc_decode<T: Message + Default>(data: &[u8]) -> Result<(T, bool), GrpcError> {
109  if data.len() < 5 {
110    return Err(GrpcError::InvalidFrame);
111  }
112
113  let compressed = data[0] != 0;
114  if compressed {
115    return Err(GrpcError::CompressionUnsupported);
116  }
117  let msg_len = u32::from_be_bytes([data[1], data[2], data[3], data[4]]) as usize;
118
119  if msg_len > MAX_GRPC_MESSAGE_SIZE {
120    return Err(GrpcError::MessageTooLarge);
121  }
122  if data.len() < 5 + msg_len {
123    return Err(GrpcError::InvalidFrame);
124  }
125
126  let msg = T::decode(&data[5..5 + msg_len]).map_err(|e| GrpcError::DecodeError(e.to_string()))?;
127  Ok((msg, compressed))
128}
tako_rs_core/grpc/framing.rs

tako_rs_core/grpc/
framing.rs
