Line data Source code
1 : use std::{collections::HashMap, fmt::Error};
2 : use base64::Engine;
3 : use serde::Serialize;
4 : use http::{header::CONTENT_TYPE, Response, StatusCode};
5 :
6 : use crate::proxy_headers::{
7 : HEADER_NAME_ERROR, HEADER_NAME_GRPC_MESSAGE, HEADER_NAME_GRPC_STATUS,
8 : HEADER_NAME_GRPC_STATUS_DETAILS,
9 : };
10 :
11 : #[derive(Debug)]
12 : pub enum GrpcStreamElement {
13 : Message,
14 : Trailers,
15 : }
16 :
17 : #[derive(Debug)]
18 : pub enum Status {
19 : GrpcCallFailed{ status: tonic::Status },
20 : GrpcChannelIdIsUnknown{ channel_id: usize },
21 : GrpcEndpointConnectFailed{ message: String },
22 : GrpcTlsConfigInvalid { message: String },
23 : GrpcStreamClosed { channel_id: usize, stream_id: usize },
24 : GrpcStreamIsUnknown { channel_id: usize, stream_id: usize },
25 : GrpcStreamReadFailed { channel_id: usize, stream_id: usize, element: GrpcStreamElement, status: tonic::Status },
26 : GrpcStreamReadTimedOut { channel_id: usize, stream_id: usize },
27 : HeaderHasInvalidEncoding{ header: &'static str, message: String },
28 : HeaderIsNotAValidEndpoint{ header: &'static str, message: String },
29 : HeaderIsNotAnUsize{ header: &'static str, message: String },
30 : HeaderPathIsInvalid{ header: &'static str, path: String, message: String },
31 : HeaderRequiredButMissing{ header: &'static str },
32 : HttpClientConfigInvalid{ message: String },
33 : HttpEndpointIsInvalid{ header: &'static str, endpoint: String, message: String },
34 : HttpMethodIsInvalid{ header: &'static str, method: String, message: String },
35 : HttpRequestFailed { stream_id: usize, error: String },
36 : HttpStreamClosed { stream_id: usize },
37 : HttpStreamFailed { stream_id: usize, error: String },
38 : HttpStreamIsUnknown { stream_id: usize },
39 : HttpStreamReadFailed { stream_id: usize, error: String },
40 : HttpStreamReadTimedOut { stream_id: usize },
41 : HttpUrlIsInvalid{ message: String },
42 : }
43 :
44 : /// The JSON error body sent for a failed proxy request
45 : #[derive(Serialize)]
46 : pub struct StatusMessage {
47 : /// The human-readable error message
48 : message: String,
49 : /// Additional error details, e.g. the offending header or the underlying error
50 : details: HashMap<&'static str, String>,
51 : }
52 :
53 : impl TryFrom<&Status> for StatusMessage {
54 : type Error = std::fmt::Error;
55 :
56 71 : fn try_from(s: &Status) -> Result<StatusMessage, Error> {
57 71 : match s {
58 1 : Status::HeaderHasInvalidEncoding { header, message } => Ok(StatusMessage {
59 1 : message: "header has an invalid encoding".to_string(),
60 1 : details: HashMap::from_iter([
61 1 : ("header", header.to_string()),
62 1 : ("error", message.to_string())
63 1 : ]),
64 1 : }),
65 2 : Status::HeaderIsNotAValidEndpoint { header, message } => Ok(StatusMessage {
66 2 : message: "header is not a valid endpoint".to_string(),
67 2 : details: HashMap::from_iter([
68 2 : ("header", header.to_string()),
69 2 : ("error", message.to_string())
70 2 : ]),
71 2 : }),
72 15 : Status::HeaderRequiredButMissing { header } => Ok(StatusMessage {
73 15 : message: "header is missing".to_string(),
74 15 : details: HashMap::from_iter([
75 15 : ("header", header.to_string())
76 15 : ]),
77 15 : }),
78 3 : Status::HeaderIsNotAnUsize { header, message } => Ok(StatusMessage {
79 3 : message: "header is not an unsigned integer".to_string(),
80 3 : details: HashMap::from_iter([
81 3 : ("header", header.to_string()),
82 3 : ("error", message.to_string())
83 3 : ]),
84 3 : }),
85 3 : Status::HeaderPathIsInvalid { header, path, message } => Ok(StatusMessage {
86 3 : message: "header stores invalid path".to_string(),
87 3 : details: HashMap::from_iter([
88 3 : ("header", header.to_string()),
89 3 : ("path", path.to_string()),
90 3 : ("error", message.to_string())
91 3 : ]),
92 3 : }),
93 2 : Status::HttpEndpointIsInvalid { header, endpoint, message } => Ok(StatusMessage {
94 2 : message: "header stores invalid endpoint".to_string(),
95 2 : details: HashMap::from_iter([
96 2 : ("header", header.to_string()),
97 2 : ("endpoint", endpoint.to_string()),
98 2 : ("error", message.to_string())
99 2 : ]),
100 2 : }),
101 2 : Status::HttpMethodIsInvalid { header, method, message } => Ok(StatusMessage {
102 2 : message: "header stores invalid method".to_string(),
103 2 : details: HashMap::from_iter([
104 2 : ("header", header.to_string()),
105 2 : ("method", method.to_string()),
106 2 : ("error", message.to_string())
107 2 : ]),
108 2 : }),
109 5 : Status::GrpcChannelIdIsUnknown { channel_id } => Ok(StatusMessage {
110 5 : message: "gRPC channel id is unknown".to_string(),
111 5 : details: HashMap::from_iter([
112 5 : ("channel", channel_id.to_string()),
113 5 : ]),
114 5 : }),
115 3 : Status::GrpcEndpointConnectFailed { message } => Ok(StatusMessage {
116 3 : message: format!("connecting to gRPC endpoint failed"),
117 3 : details: HashMap::from_iter([
118 3 : ("error", message.to_string()),
119 3 : ]),
120 3 : }),
121 3 : Status::GrpcTlsConfigInvalid { message } => Ok(StatusMessage {
122 3 : message: "gRPC tls config is invalid".to_string(),
123 3 : details: HashMap::from_iter([
124 3 : ("error", message.to_string()),
125 3 : ]),
126 3 : }),
127 6 : Status::GrpcCallFailed { status } => Ok(StatusMessage {
128 6 : message: "gRPC call failed".to_string(),
129 6 : details: HashMap::from_iter([
130 6 : ("code", status.code().to_string()),
131 6 : ("error", status.to_string()),
132 6 : ]),
133 6 : }),
134 4 : Status::GrpcStreamReadFailed { channel_id, stream_id, element, status } => Ok(StatusMessage {
135 4 : message: "reading from gRPC stream failed".to_string(),
136 4 : details: HashMap::from_iter([
137 4 : ("target", match element { GrpcStreamElement::Trailers => "trailers", GrpcStreamElement::Message => "message" }.to_string()),
138 4 : ("stream", stream_id.to_string()),
139 4 : ("channel", channel_id.to_string()),
140 4 : ("code", status.code().to_string()),
141 4 : ("error", status.to_string()),
142 : ]),
143 : }),
144 2 : Status::GrpcStreamReadTimedOut { channel_id, stream_id } => Ok(StatusMessage {
145 2 : message: "reading from gRPC stream timed out".to_string(),
146 2 : details: HashMap::from_iter([
147 2 : ("stream", stream_id.to_string()),
148 2 : ("channel", channel_id.to_string()),
149 2 : ]),
150 2 : }),
151 4 : Status::GrpcStreamIsUnknown { channel_id, stream_id } => Ok(StatusMessage {
152 4 : message: "gRPC stream is unknown".to_string(),
153 4 : details: HashMap::from_iter([
154 4 : ("stream", stream_id.to_string()),
155 4 : ("channel", channel_id.to_string()),
156 4 : ]),
157 4 : }),
158 1 : Status::GrpcStreamClosed { channel_id, stream_id } => Ok(StatusMessage {
159 1 : message: "gRPC stream is closed".to_string(),
160 1 : details: HashMap::from_iter([
161 1 : ("stream", stream_id.to_string()),
162 1 : ("channel", channel_id.to_string()),
163 1 : ]),
164 1 : }),
165 2 : Status::HttpRequestFailed { stream_id, error } => Ok(StatusMessage {
166 2 : message: "http request failed".to_string(),
167 2 : details: HashMap::from_iter([
168 2 : ("stream", stream_id.to_string()),
169 2 : ("error", error.to_string()),
170 2 : ]),
171 2 : }),
172 2 : Status::HttpStreamReadFailed { stream_id, error } => Ok(StatusMessage {
173 2 : message: "reading chunk from http stream failed".to_string(),
174 2 : details: HashMap::from_iter([
175 2 : ("stream", stream_id.to_string()),
176 2 : ("error", error.to_string()),
177 2 : ]),
178 2 : }),
179 2 : Status::HttpStreamReadTimedOut { stream_id } => Ok(StatusMessage {
180 2 : message: "reading from http stream timed out".to_string(),
181 2 : details: HashMap::from_iter([
182 2 : ("stream", stream_id.to_string()),
183 2 : ]),
184 2 : }),
185 5 : Status::HttpStreamIsUnknown { stream_id } => Ok(StatusMessage {
186 5 : message: "http stream is unknown".to_string(),
187 5 : details: HashMap::from_iter([
188 5 : ("stream", stream_id.to_string()),
189 5 : ]),
190 5 : }),
191 1 : Status::HttpStreamClosed { stream_id } => Ok(StatusMessage {
192 1 : message: "http stream is closed".to_string(),
193 1 : details: HashMap::from_iter([
194 1 : ("stream", stream_id.to_string()),
195 1 : ]),
196 1 : }),
197 1 : Status::HttpStreamFailed { stream_id, error } => Ok(StatusMessage {
198 1 : message: "http stream failed".to_string(),
199 1 : details: HashMap::from_iter([
200 1 : ("stream", stream_id.to_string()),
201 1 : ("error", error.to_string()),
202 1 : ]),
203 1 : }),
204 1 : Status::HttpClientConfigInvalid { message } => Ok(StatusMessage {
205 1 : message: format!("http client config is invalid"),
206 1 : details: HashMap::from_iter([
207 1 : ("error", message.to_string()),
208 1 : ]),
209 1 : }),
210 1 : Status::HttpUrlIsInvalid { message } => Ok(StatusMessage {
211 1 : message: format!("http url is invalid"),
212 1 : details: HashMap::from_iter([
213 1 : ("error", message.to_string()),
214 1 : ]),
215 1 : }),
216 : }
217 71 : }
218 : }
219 :
220 : impl From<&Status> for StatusCode {
221 71 : fn from(status: &Status) -> StatusCode {
222 71 : match status {
223 6 : Status::GrpcCallFailed { status: _ } => StatusCode::BAD_REQUEST,
224 5 : Status::GrpcChannelIdIsUnknown { channel_id: _ } => StatusCode::NOT_FOUND,
225 3 : Status::GrpcEndpointConnectFailed { message: _ } => StatusCode::BAD_REQUEST,
226 3 : Status::GrpcTlsConfigInvalid { message: _ } => StatusCode::BAD_REQUEST,
227 1 : Status::GrpcStreamClosed { channel_id: _, stream_id: _ } => StatusCode::BAD_REQUEST,
228 4 : Status::GrpcStreamIsUnknown { channel_id: _, stream_id: _ } => StatusCode::NOT_FOUND,
229 4 : Status::GrpcStreamReadFailed { channel_id: _, stream_id: _, element: _, status: _ } => StatusCode::BAD_REQUEST,
230 2 : Status::GrpcStreamReadTimedOut { channel_id: _, stream_id: _ } => StatusCode::BAD_REQUEST,
231 1 : Status::HeaderHasInvalidEncoding { header: _, message: _ } => StatusCode::BAD_REQUEST,
232 2 : Status::HeaderIsNotAValidEndpoint { header: _, message: _ } => StatusCode::BAD_REQUEST,
233 3 : Status::HeaderIsNotAnUsize { header: _, message: _ } => StatusCode::BAD_REQUEST,
234 3 : Status::HeaderPathIsInvalid { header: _, path: _, message: _ } => StatusCode::BAD_REQUEST,
235 15 : Status::HeaderRequiredButMissing { header: _ } => StatusCode::BAD_REQUEST,
236 1 : Status::HttpClientConfigInvalid { message: _ } => StatusCode::BAD_REQUEST,
237 2 : Status::HttpEndpointIsInvalid { header: _, endpoint: _ , message: _ } => StatusCode::BAD_REQUEST,
238 2 : Status::HttpMethodIsInvalid { header: _, method: _, message: _ } => StatusCode::BAD_REQUEST,
239 2 : Status::HttpRequestFailed { stream_id: _, error: _ } => StatusCode::BAD_REQUEST,
240 1 : Status::HttpStreamClosed { stream_id: _ } => StatusCode::BAD_REQUEST,
241 1 : Status::HttpStreamFailed { stream_id: _, error: _ } => StatusCode::BAD_REQUEST,
242 5 : Status::HttpStreamIsUnknown { stream_id: _ } => StatusCode::NOT_FOUND,
243 2 : Status::HttpStreamReadFailed { stream_id: _, error: _ } => StatusCode::BAD_REQUEST,
244 2 : Status::HttpStreamReadTimedOut { stream_id: _ } => StatusCode::BAD_REQUEST,
245 1 : Status::HttpUrlIsInvalid { message: _ } => StatusCode::BAD_REQUEST,
246 : }
247 71 : }
248 : }
249 :
250 : /// Percent-encode a gRPC status message for a header, the encoding of the `grpc-message` trailer.
251 : /// Database errors span lines and quote the text of the user, which a header value cannot hold.
252 10 : fn encode_grpc_message(message: &str) -> String {
253 10 : let mut encoded = String::with_capacity(message.len());
254 162 : for byte in message.bytes() {
255 162 : if (0x20..=0x7E).contains(&byte) && byte != b'%' {
256 150 : encoded.push(byte as char);
257 150 : } else {
258 12 : encoded.push_str(&format!("%{byte:02X}"));
259 12 : }
260 : }
261 10 : encoded
262 10 : }
263 :
264 : /// Build an error response for a failed gRPC call/stream, forwarding the
265 : /// gRPC "richer error model" data to the frontend:
266 : /// - `dashql-grpc-status`: the numeric gRPC status code
267 : /// - `dashql-grpc-message`: the terse gRPC status message, percent-encoded
268 : /// - `dashql-grpc-status-details-bin`: base64 of the `google.rpc.Status`
269 : /// details envelope (carrying Hyper's `ErrorInfo`), if present.
270 10 : fn grpc_error_response(s: &Status, status: &tonic::Status, body: Vec<u8>) -> Response<Vec<u8>> {
271 10 : let grpc_code = (status.code() as usize).to_string();
272 10 : let mut builder = Response::builder()
273 10 : .status(StatusCode::from(s).as_u16())
274 10 : .header(HEADER_NAME_ERROR, "true")
275 10 : .header(HEADER_NAME_GRPC_STATUS, &grpc_code)
276 10 : .header(HEADER_NAME_GRPC_MESSAGE, encode_grpc_message(status.message()))
277 10 : .header(CONTENT_TYPE, mime::APPLICATION_JSON.essence_str());
278 : // Forward the raw status-details-bin (google.rpc.Status) as base64, if any.
279 : // The frontend decodes it to surface Hyper's ErrorInfo (sqlstate, hint,
280 : // customer/system detail, SQL text position, ...).
281 10 : let details = status.details();
282 10 : if !details.is_empty() {
283 3 : let encoded = base64::engine::general_purpose::STANDARD.encode(details);
284 3 : builder = builder.header(HEADER_NAME_GRPC_STATUS_DETAILS, encoded);
285 7 : }
286 10 : builder.body(body).unwrap()
287 10 : }
288 :
289 : impl From<&Status> for Response<Vec<u8>> {
290 71 : fn from(s: &Status) -> Response<Vec<u8>> {
291 71 : let mut body: Vec<u8> = Vec::new();
292 71 : if let Ok(status_msg) = StatusMessage::try_from(s) {
293 71 : body = serde_json::to_vec(&status_msg).unwrap_or_default();
294 71 : }
295 71 : match &s {
296 6 : Status::GrpcCallFailed { ref status }
297 4 : | Status::GrpcStreamReadFailed { channel_id: _, stream_id: _, element: _, ref status } => {
298 10 : grpc_error_response(s, status, body)
299 : },
300 : _ => {
301 61 : Response::builder()
302 61 : .status(StatusCode::from(s).as_u16())
303 61 : .header(HEADER_NAME_ERROR, "true")
304 61 : .header(CONTENT_TYPE, mime::APPLICATION_JSON.essence_str())
305 61 : .body(body)
306 61 : .unwrap()
307 : }
308 : }
309 71 : }
310 : }
311 :
312 : #[cfg(test)]
313 : mod test {
314 : use super::*;
315 :
316 57 : fn header(response: &Response<Vec<u8>>, name: &str) -> Option<String> {
317 57 : response.headers().get(name).map(|value| value.to_str().unwrap().to_string())
318 57 : }
319 :
320 26 : fn json(response: &Response<Vec<u8>>) -> serde_json::Value {
321 26 : serde_json::from_slice(response.body()).unwrap()
322 26 : }
323 :
324 : #[test]
325 1 : fn every_status_describes_itself_as_json() {
326 13 : let text = || "text".to_string();
327 3 : let grpc = || tonic::Status::internal("boom");
328 1 : let cases: Vec<(Status, u16, &str, &[(&str, &str)])> = vec![
329 1 : (Status::GrpcChannelIdIsUnknown { channel_id: 1 }, 404, "gRPC channel id is unknown", &[("channel", "1")]),
330 1 : (Status::GrpcEndpointConnectFailed { message: text() }, 400, "connecting to gRPC endpoint failed", &[("error", "text")]),
331 1 : (Status::GrpcTlsConfigInvalid { message: text() }, 400, "gRPC tls config is invalid", &[("error", "text")]),
332 1 : (Status::GrpcStreamClosed { channel_id: 1, stream_id: 2 }, 400, "gRPC stream is closed", &[("channel", "1"), ("stream", "2")]),
333 1 : (Status::GrpcStreamIsUnknown { channel_id: 1, stream_id: 2 }, 404, "gRPC stream is unknown", &[("channel", "1"), ("stream", "2")]),
334 1 : (Status::GrpcStreamReadTimedOut { channel_id: 1, stream_id: 2 }, 400, "reading from gRPC stream timed out", &[("channel", "1"), ("stream", "2")]),
335 1 : (Status::GrpcCallFailed { status: grpc() }, 400, "gRPC call failed", &[]),
336 1 : (Status::GrpcStreamReadFailed { channel_id: 1, stream_id: 2, element: GrpcStreamElement::Message, status: grpc() }, 400, "reading from gRPC stream failed", &[("target", "message")]),
337 1 : (Status::GrpcStreamReadFailed { channel_id: 1, stream_id: 2, element: GrpcStreamElement::Trailers, status: grpc() }, 400, "reading from gRPC stream failed", &[("target", "trailers")]),
338 1 : (Status::HeaderHasInvalidEncoding { header: "h", message: text() }, 400, "header has an invalid encoding", &[("header", "h"), ("error", "text")]),
339 1 : (Status::HeaderIsNotAValidEndpoint { header: "h", message: text() }, 400, "header is not a valid endpoint", &[("header", "h"), ("error", "text")]),
340 1 : (Status::HeaderIsNotAnUsize { header: "h", message: text() }, 400, "header is not an unsigned integer", &[("header", "h"), ("error", "text")]),
341 1 : (Status::HeaderPathIsInvalid { header: "h", path: "p".to_string(), message: text() }, 400, "header stores invalid path", &[("header", "h"), ("path", "p")]),
342 1 : (Status::HeaderRequiredButMissing { header: "h" }, 400, "header is missing", &[("header", "h")]),
343 1 : (Status::HttpClientConfigInvalid { message: text() }, 400, "http client config is invalid", &[("error", "text")]),
344 1 : (Status::HttpEndpointIsInvalid { header: "h", endpoint: "e".to_string(), message: text() }, 400, "header stores invalid endpoint", &[("endpoint", "e")]),
345 1 : (Status::HttpMethodIsInvalid { header: "h", method: "m".to_string(), message: text() }, 400, "header stores invalid method", &[("method", "m")]),
346 1 : (Status::HttpRequestFailed { stream_id: 2, error: text() }, 400, "http request failed", &[("stream", "2"), ("error", "text")]),
347 1 : (Status::HttpStreamClosed { stream_id: 2 }, 400, "http stream is closed", &[("stream", "2")]),
348 1 : (Status::HttpStreamFailed { stream_id: 2, error: text() }, 400, "http stream failed", &[("stream", "2"), ("error", "text")]),
349 1 : (Status::HttpStreamIsUnknown { stream_id: 2 }, 404, "http stream is unknown", &[("stream", "2")]),
350 1 : (Status::HttpStreamReadFailed { stream_id: 2, error: text() }, 400, "reading chunk from http stream failed", &[("stream", "2"), ("error", "text")]),
351 1 : (Status::HttpStreamReadTimedOut { stream_id: 2 }, 400, "reading from http stream timed out", &[("stream", "2")]),
352 1 : (Status::HttpUrlIsInvalid { message: text() }, 400, "http url is invalid", &[("error", "text")]),
353 : ];
354 24 : for (status, code, message, details) in cases {
355 24 : let response = Response::from(&status);
356 24 : assert_eq!(response.status().as_u16(), code, "{status:?}");
357 24 : assert_eq!(header(&response, HEADER_NAME_ERROR).as_deref(), Some("true"), "{status:?}");
358 24 : assert_eq!(header(&response, "content-type").as_deref(), Some("application/json"), "{status:?}");
359 24 : let body = json(&response);
360 24 : assert_eq!(body["message"], message, "{status:?}");
361 33 : for (name, value) in details {
362 33 : assert_eq!(body["details"][name], *value, "{status:?}");
363 : }
364 : }
365 1 : }
366 :
367 : #[test]
368 1 : fn a_failed_grpc_call_forwards_the_rich_error() {
369 1 : let status = tonic::Status::with_details(tonic::Code::InvalidArgument, "bad query", vec![0, 1, 255].into());
370 1 : let response = Response::from(&Status::GrpcCallFailed { status });
371 1 : assert_eq!(header(&response, HEADER_NAME_GRPC_STATUS).as_deref(), Some("3"));
372 1 : assert_eq!(header(&response, HEADER_NAME_GRPC_MESSAGE).as_deref(), Some("bad query"));
373 1 : assert_eq!(header(&response, HEADER_NAME_GRPC_STATUS_DETAILS).as_deref(), Some("AAH/"));
374 1 : }
375 :
376 : #[test]
377 1 : fn a_failed_grpc_call_without_details_sends_none() {
378 1 : let response = Response::from(&Status::GrpcCallFailed { status: tonic::Status::internal("boom") });
379 1 : assert_eq!(header(&response, HEADER_NAME_GRPC_STATUS).as_deref(), Some("13"));
380 1 : assert_eq!(header(&response, HEADER_NAME_GRPC_STATUS_DETAILS), None);
381 1 : }
382 :
383 : #[test]
384 1 : fn a_grpc_message_that_is_no_header_value_reaches_the_app() {
385 : // Database errors span lines and quote the text of the user
386 1 : let message = "syntax error at line 1\nnear \"caf\u{e9}\" 100%";
387 1 : let status = tonic::Status::new(tonic::Code::InvalidArgument, message);
388 2 : for status in [
389 1 : Status::GrpcCallFailed { status: status.clone() },
390 1 : Status::GrpcStreamReadFailed { channel_id: 1, stream_id: 2, element: GrpcStreamElement::Message, status },
391 1 : ] {
392 2 : let response = Response::from(&status);
393 2 : assert_eq!(header(&response, HEADER_NAME_GRPC_STATUS).as_deref(), Some("3"));
394 : // Percent-encoded like the `grpc-message` trailer it comes from
395 2 : assert_eq!(
396 2 : header(&response, HEADER_NAME_GRPC_MESSAGE).as_deref(),
397 : Some("syntax error at line 1%0Anear \"caf%C3%A9\" 100%25"),
398 : );
399 2 : assert!(json(&response)["details"]["error"].as_str().unwrap().contains("caf\u{e9}"));
400 : }
401 1 : }
402 : }
|
