Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Input bounds wall: every request-shaped resource has
3 : //! a configured cap, rejected with the spec-shaped error. One middleware
4 : //! enforces the transport-level caps (URI length 414, body size 413, JSON
5 : //! nesting 400) — size and depth are checked BEFORE any parse. The
6 : //! per-feature caps (batch count, joinLevel, @context fetch
7 : //! count, q= complexity, result ceiling) live at their parse points.
8 : //! Rejections are counted and exported via /q/health.
9 :
10 : use axum::body::{Body, Bytes};
11 : use axum::http::{Request, StatusCode};
12 : use axum::middleware::Next;
13 : use axum::response::{IntoResponse, Response};
14 : use std::sync::atomic::{AtomicU64, Ordering};
15 :
16 : /// Hard caps (v1: compile-time constants — a config file is a later knob;
17 : /// every value is spec-shaped on rejection).
18 : /// → bare 413 (6.3.4). Deployment knob (ANTARES_MAX_BODY_BYTES): the spec
19 : /// names no ceiling; 4 MiB is the DoS bound, raised where a trusted
20 : /// producer legitimately sends bigger batches. Read once at first use.
21 210 : pub static MAX_BODY_BYTES: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
22 210 : std::env::var("ANTARES_MAX_BODY_BYTES")
23 210 : .ok()
24 210 : .and_then(|v| v.parse().ok())
25 210 : .filter(|n| *n > 0)
26 210 : .unwrap_or(4 * 1024 * 1024)
27 210 : });
28 : pub const MAX_URI_BYTES: usize = 8 * 1024; // → bare 414
29 : pub const MAX_JSON_DEPTH: usize = 64; // → 400 BadRequestData
30 : /// → 400 BadRequestData. Maximum coordinate positions in a QUERY geometry
31 : /// (4.10 geoQ, 4.23 ordering reference). The spec sets no ceiling, and the
32 : /// geometry is not bounded by the URI length on the POST query path — the
33 : /// body carries it. Every position is an edge the DE-9IM relate walks once
34 : /// per candidate entity, so the work a single request can buy is capped
35 : /// here: 1024 positions describe an administrative boundary at street
36 : /// resolution, and are already more than the 8 KiB URI ceiling can carry.
37 : pub use antares_ql::geo::MAX_GEO_VERTICES;
38 : /// → 400 BadRequestData. Deployment knob (ANTARES_MAX_BATCH_ITEMS): the
39 : /// spec sets no batch ceiling — 1000 is this broker's DoS-bounds default,
40 : /// raised where a trusted producer legitimately batches larger (e.g. a
41 : /// full-fleet upsert). Read once at first use.
42 54 : pub static MAX_BATCH_ITEMS: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
43 54 : std::env::var("ANTARES_MAX_BATCH_ITEMS")
44 54 : .ok()
45 54 : .and_then(|v| v.parse().ok())
46 54 : .filter(|n| *n > 0)
47 54 : .unwrap_or(1_000)
48 54 : });
49 : // Deployment knob (ANTARES_MAX_FED_RESPONSE_BYTES): ceiling on one forwarded
50 : // (4.3.6) response body. The spec sets no ceiling; an over-cap peer part
51 : // fails like an unparseable payload (Table 6.3.17-1, warning 111) instead of
52 : // ballooning broker memory — one misbehaving peer must not break the
53 : // 500 MB RSS budget. Read once at first use.
54 74 : pub static MAX_FED_RESPONSE_BYTES: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
55 74 : std::env::var("ANTARES_MAX_FED_RESPONSE_BYTES")
56 74 : .ok()
57 74 : .and_then(|v| v.parse().ok())
58 74 : .filter(|n| *n > 0)
59 74 : .unwrap_or(16 * 1024 * 1024)
60 74 : });
61 : // Deployment knob (ANTARES_FED_INFLIGHT): forwarded requests in flight for
62 : // the whole process. Per-request fan-out is bounded below; across requests
63 : // nothing was, and 6 000 open federated queries × 34 sources each held
64 : // 7.7 GB of buffers and connections. Callers over the cap wait their turn.
65 76 : pub static MAX_FED_INFLIGHT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
66 76 : std::env::var("ANTARES_FED_INFLIGHT")
67 76 : .ok()
68 76 : .and_then(|v| v.parse().ok())
69 76 : .filter(|n| *n > 0)
70 76 : .unwrap_or(256)
71 76 : });
72 : pub static FED_INFLIGHT: std::sync::LazyLock<tokio::sync::Semaphore> =
73 58 : std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(*MAX_FED_INFLIGHT));
74 : // Deployment knob (ANTARES_FED_FANOUT): concurrent forwards per distributed
75 : // read. 4.3.6.1 orders the MERGE (4.5.5), never the requests, so forwards
76 : // run concurrently; this bounds how many at once per request.
77 148 : pub static MAX_FED_FANOUT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
78 148 : std::env::var("ANTARES_FED_FANOUT")
79 148 : .ok()
80 148 : .and_then(|v| v.parse().ok())
81 148 : .filter(|n| *n > 0)
82 148 : .unwrap_or(8)
83 148 : });
84 : /// 6.3.17: `NGSILD-Warning` values relayed from ONE Context Source. The
85 : /// clause makes a peer's warnings part of this broker's answer (4.3.6.4 puts
86 : /// a deeper hop's abnormality on the response that survived it), so the list
87 : /// is written by the peer but sent by this broker to a client that never
88 : /// addressed it. Table 6.3.17-1 defines four codes and a cascade adds a few
89 : /// per hop, so eight carries a real cascade; past it a source would make the
90 : /// response grow faster than the fan-out does, and would crowd out the
91 : /// warnings the clause obliges this broker to raise about the other sources.
92 : pub const MAX_PEER_WARNINGS: usize = 8;
93 :
94 : /// → 500 InternalError. How many [`crate::AppState::call`] frames one
95 : /// request may be inside. The in-process handle is the façade seam, and a
96 : /// façade legitimately calls the broker while another façade legitimately
97 : /// calls it, so the ceiling is a depth rather than a refusal of the second
98 : /// call. Without one a `/x/` route that translates into a request its own
99 : /// surface serves recurses until the stack ends, and each frame builds a
100 : /// router of its own. Eight carries a façade over a façade over the broker
101 : /// several times and still ends the loop in milliseconds, an order of
102 : /// magnitude below what ends the process: with the guard removed, a route
103 : /// asking for thirty-two hops overflows a 2 MiB thread stack in a debug
104 : /// build. The count is per task — work a handler spawns starts a new chain,
105 : /// which is right, since a notification is not inside the request that
106 : /// caused it.
107 : pub const MAX_IN_PROCESS_CALL_DEPTH: usize = 8;
108 :
109 : pub const MAX_JOIN_LEVEL: usize = 10; // → 400 BadRequestData
110 : /// → 400 BadRequestData. Documents one @context resolution may fetch, owned
111 : /// by the loader that enforces it (`antares_jsonld`), and the ceiling on how
112 : /// many DISTINCT @contexts one batch may name — without the second, the item
113 : /// count multiplies the first.
114 : pub use antares_jsonld::MAX_CONTEXT_URLS as MAX_CONTEXT_FETCHES;
115 : /// Linked-entity lookup budget per `q=`, owned by the shared evaluator.
116 : pub use antares_ql::eval::MAX_Q_LINK_LOOKUPS;
117 : /// Regex compile ceiling and retention caps, owned by the shared cache
118 : /// (`antares_ql::regex`).
119 : pub use antares_ql::regex::{MAX_REGEX_CACHE, MAX_REGEX_CACHE_BYTES, MAX_REGEX_PROGRAM_BYTES};
120 : /// → 403 TooComplexQuery. The AST size cap, owned by the parser that
121 : /// enforces it (`antares_ql::parse_q`).
122 : pub use antares_ql::MAX_Q_NODES;
123 :
124 : /// → 403 TooManyResults (5.5.6). Deployment knob
125 : /// (ANTARES_DISCOVERY_SCAN_MAX): documents ONE unpaginated whole-tenant fold
126 : /// may read or hold. The folds that carry it are the ones the spec gives no
127 : /// window to push into the store: `/types` and `/attributes` (5.7.5-5.7.10
128 : /// define no pagination) and the registration query (5.10.2.4 filters first
129 : /// and pages second, so the page cannot be pushed down). Without it one
130 : /// request over a tenant at the 100 000-registration target holds every match
131 : /// at once. Read once at first use.
132 38 : pub static MAX_FOLD_DOCS: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
133 38 : std::env::var("ANTARES_DISCOVERY_SCAN_MAX")
134 38 : .ok()
135 38 : .and_then(|v| v.parse().ok())
136 38 : .filter(|n| *n > 0)
137 38 : .unwrap_or(100_000)
138 38 : });
139 :
140 : /// The ceilings on the notification pipeline. Not input bounds: no request
141 : /// names them and none is rejected against them. They are published beside
142 : /// the input caps for the same reason the regex-cache caps above are — an
143 : /// operator reading `/q/health` has to be able to tell a ceiling from a
144 : /// coincidence, and reaching one of these is what a dropped change or a
145 : /// stalled fan-out looks like from outside.
146 : ///
147 : /// Depth of the change→matcher ring, the same size the local bus uses. A
148 : /// full ring drops the batch and counts it
149 : /// (`antares_notification_changes_dropped_total`), so this number is the
150 : /// back-pressure a deployment has before delivery loss starts.
151 : pub const CHANGE_QUEUE: usize = 1024;
152 : /// Notifications in flight at once per drain: one serial POST at a time
153 : /// capped a 9-subscription fan-out at ~600 POST/s and overflowed the ring.
154 : /// Deployment knob (ANTARES_DELIVERY_WIDTH): what a width is worth is a
155 : /// property of the subscribers, not of this broker — a slot is held for as
156 : /// long as the endpoint takes to answer, so a fleet of local sinks and a
157 : /// fleet of remote endpoints sitting at their 30 s timeout (Table 5.2.15-1)
158 : /// are served by different numbers, and the one that fits is measured
159 : /// against a deployment rather than compiled in. Read once at first use.
160 66 : pub static DELIVERY_WIDTH: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
161 66 : count_from(
162 66 : std::env::var("ANTARES_DELIVERY_WIDTH").ok().as_deref(),
163 : 64,
164 : usize::MAX,
165 : )
166 66 : });
167 : /// Of that width, what one tenant may hold. A Subscription belongs to one
168 : /// tenant (5.2.12), and a delivery to an endpoint that accepts and never
169 : /// answers holds its slot for the endpoint's whole timeout — up to 30 s
170 : /// (Table 5.2.15-1). Sharing the width with no per-tenant bound, a single
171 : /// tenant with enough dead endpoints holds every slot and nothing leaves the
172 : /// broker for anyone else. The share is a fraction of the width and not a
173 : /// fair split of it: a tenant delivering alone still gets several slots, and
174 : /// eight of them is what a full width of 64 divides into before the
175 : /// per-tenant queue becomes the bottleneck for an ordinary fan-out.
176 : /// Deployment knob (ANTARES_DELIVERY_WIDTH_PER_TENANT), ceilinged at the
177 : /// width above: a share larger than the width is not a share, and
178 : /// publishing one would name a ceiling that can never fire. Read once at
179 : /// first use.
180 66 : pub static DELIVERY_WIDTH_PER_TENANT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
181 66 : count_from(
182 66 : std::env::var("ANTARES_DELIVERY_WIDTH_PER_TENANT")
183 66 : .ok()
184 66 : .as_deref(),
185 : 8,
186 66 : *DELIVERY_WIDTH,
187 : )
188 66 : });
189 :
190 : /// One configured count: the default stands in for anything unset,
191 : /// unparseable or zero — a zero would mint a semaphore that admits nobody
192 : /// and stop delivery altogether — and the ceiling bounds what a deployment
193 : /// can ask for where a larger number would be meaningless.
194 164 : fn count_from(raw: Option<&str>, default: usize, ceiling: usize) -> usize {
195 164 : raw.and_then(|v| v.parse().ok())
196 164 : .filter(|n| *n > 0)
197 164 : .unwrap_or(default)
198 164 : .min(ceiling)
199 164 : }
200 : /// Destinations and registrations the egress breaker tracks. Both maps are
201 : /// keyed by client-supplied strings, so they need a bound: at the ceiling the
202 : /// least recently recorded entry is dropped, which costs at most a forgotten
203 : /// failure count for a destination nobody has touched in a while.
204 : pub const MAX_TRACKED_DESTINATIONS: usize = 4096;
205 :
206 : /// Rejection counters, exported by /q/health.
207 : #[derive(Default)]
208 : pub struct LimitStats {
209 : pub uri_too_long: AtomicU64,
210 : pub body_too_large: AtomicU64,
211 : pub body_too_deep: AtomicU64,
212 : }
213 :
214 : impl LimitStats {
215 112 : pub fn snapshot(&self) -> serde_json::Value {
216 112 : serde_json::json!({
217 112 : "maxBodyBytes": *MAX_BODY_BYTES,
218 112 : "maxUriBytes": MAX_URI_BYTES,
219 112 : "maxJsonDepth": MAX_JSON_DEPTH,
220 112 : "maxGeoVertices": MAX_GEO_VERTICES,
221 112 : "maxBatchItems": *MAX_BATCH_ITEMS,
222 112 : "maxFoldDocs": *MAX_FOLD_DOCS,
223 112 : "maxFedResponseBytes": *MAX_FED_RESPONSE_BYTES,
224 112 : "maxFedFanout": *MAX_FED_FANOUT,
225 112 : "maxFedInflight": *MAX_FED_INFLIGHT,
226 112 : "maxJoinLevel": MAX_JOIN_LEVEL,
227 112 : "maxInProcessCallDepth": MAX_IN_PROCESS_CALL_DEPTH,
228 112 : "maxPeerWarnings": MAX_PEER_WARNINGS,
229 112 : "maxContextFetches": MAX_CONTEXT_FETCHES,
230 112 : "maxQNodes": MAX_Q_NODES,
231 112 : "maxQLinkLookups": MAX_Q_LINK_LOOKUPS,
232 112 : "maxRegexCache": MAX_REGEX_CACHE,
233 112 : "maxRegexCacheBytes": MAX_REGEX_CACHE_BYTES,
234 112 : "maxRegexProgramBytes": MAX_REGEX_PROGRAM_BYTES,
235 112 : "changeQueue": CHANGE_QUEUE,
236 112 : "deliveryWidth": *DELIVERY_WIDTH,
237 112 : "deliveryWidthPerTenant": *DELIVERY_WIDTH_PER_TENANT,
238 112 : "maxTrackedDestinations": MAX_TRACKED_DESTINATIONS,
239 112 : "rejectedUriTooLong": self.uri_too_long.load(Ordering::Relaxed),
240 112 : "rejectedBodyTooLarge": self.body_too_large.load(Ordering::Relaxed),
241 112 : "rejectedBodyTooDeep": self.body_too_deep.load(Ordering::Relaxed),
242 : })
243 112 : }
244 : }
245 :
246 : /// Maximum brace/bracket nesting of a JSON byte stream, string-aware.
247 : /// A scan, not a parse — depth is checked before serde ever runs.
248 22964 : pub(crate) fn json_depth(bytes: &[u8]) -> usize {
249 22964 : let (mut depth, mut max, mut in_str, mut esc) = (0usize, 0usize, false, false);
250 7816638 : for &b in bytes {
251 7816638 : if in_str {
252 4335602 : if esc {
253 566 : esc = false;
254 4335036 : } else if b == b'\\' {
255 566 : esc = true;
256 4334470 : } else if b == b'"' {
257 325998 : in_str = false;
258 4008472 : }
259 4335602 : continue;
260 3481036 : }
261 3481036 : match b {
262 325998 : b'"' => in_str = true,
263 893454 : b'{' | b'[' => {
264 893454 : depth += 1;
265 893454 : max = max.max(depth);
266 893454 : }
267 893474 : b'}' | b']' => depth = depth.saturating_sub(1),
268 1368110 : _ => {}
269 : }
270 : }
271 22964 : max
272 22964 : }
273 :
274 26786 : pub(crate) async fn bounds_layer(
275 26786 : axum::extract::State(st): axum::extract::State<crate::AppState>,
276 26786 : req: Request<Body>,
277 26786 : next: Next,
278 26786 : ) -> Response {
279 26786 : if req.uri().to_string().len() > MAX_URI_BYTES {
280 18 : st.limits.uri_too_long.fetch_add(1, Ordering::Relaxed);
281 18 : return StatusCode::URI_TOO_LONG.into_response(); // bare, like 6.3.4
282 26768 : }
283 : // 6.3.4: "For HTTP POST, PATCH and PUT HTTP requests implementations shall
284 : // check … Content-Length header shall include the length of the request
285 : // payload body", and its absence "shall result in just a 411 HTTP status
286 : // code (without any payload body)" — restated in 6.3.2. Scoped to HTTP/1.x:
287 : // HTTP/2 and later carry length in the framing layer and legitimately omit
288 : // the header, so demanding it there would reject conformant clients.
289 : // The clause grants NO exemption for `Transfer-Encoding: chunked` — a
290 : // chunked POST without Content-Length is exactly the case 411 covers, so it
291 : // is deliberately not carved out here.
292 : //
293 : // The ONE deviation, made explicit rather than implied: the check is scoped
294 : // to HTTP/1.x. 6.3.4 is written against RFC 7230/7231 and predates any h2
295 : // consideration; HTTP/2 carries length in its framing and conformant h2
296 : // clients routinely omit the header, so applying it there would reject
297 : // requests the spec never meant to describe. Recorded in docs/ics.yaml.
298 26768 : if matches!(req.method().as_str(), "POST" | "PATCH" | "PUT")
299 16434 : && req.version() <= axum::http::Version::HTTP_11
300 16434 : && !req
301 16434 : .headers()
302 16434 : .contains_key(axum::http::header::CONTENT_LENGTH)
303 : {
304 32 : return StatusCode::LENGTH_REQUIRED.into_response(); // bare 411
305 26736 : }
306 26736 : let has_body = matches!(req.method().as_str(), "POST" | "PATCH" | "PUT" | "DELETE");
307 26736 : if !has_body {
308 4062 : return next.run(req).await;
309 22674 : }
310 22674 : let (parts, body) = req.into_parts();
311 : // What the client says it is sending. The read below still decides — a
312 : // declared length is a claim, and answering on the claim alone would cut
313 : // the client off mid-body, where the RST that follows can take the
314 : // response with it.
315 22674 : let fits_its_claim = parts
316 22674 : .headers
317 22674 : .get(axum::http::header::CONTENT_LENGTH)
318 22674 : .and_then(|v| v.to_str().ok())
319 22674 : .and_then(|v| v.parse::<usize>().ok())
320 22674 : .is_some_and(|n| n <= *MAX_BODY_BYTES);
321 22674 : let bytes: Bytes = match axum::body::to_bytes(body, *MAX_BODY_BYTES).await {
322 22656 : Ok(b) => b,
323 : // `to_bytes` reports the length limit and any transport failure the
324 : // same way. A body whose declared length fits the cap cannot have
325 : // exceeded it, so what failed was the delivery: that is a bad request
326 : // and NOT a size rejection — counting it as one would make
327 : // `rejectedBodyTooLarge` read client aborts as clients hitting the
328 : // cap. A body that declared more than the cap, or declared nothing
329 : // (chunked), can only have hit the cap.
330 4 : Err(_) if fits_its_claim => {
331 4 : return crate::negotiate::ApiError::from(antares_model::NgsiError::InvalidRequest(
332 4 : "request body was not delivered completely".into(),
333 4 : ))
334 4 : .into_response();
335 : }
336 : Err(_) => {
337 14 : st.limits.body_too_large.fetch_add(1, Ordering::Relaxed);
338 14 : return StatusCode::PAYLOAD_TOO_LARGE.into_response(); // bare 413
339 : }
340 : };
341 : // An absent (or unreadable) Content-Type is parsed as JSON downstream —
342 : // 6.3.4 mandates Content-Length, not Content-Type — so it is scanned
343 : // here too, or the nesting cap has a hole exactly where the parser has
344 : // none. A header that names another media type keeps its 415.
345 22656 : let is_json = parts
346 22656 : .headers
347 22656 : .get(axum::http::header::CONTENT_TYPE)
348 22656 : .is_none_or(|v| match v.to_str() {
349 16260 : Ok(ct) => ct.contains("json"),
350 : // A header the parser cannot read is not a header naming another
351 : // media type: `negotiate::content_type` reports it as the empty
352 : // string, which every route that tolerates an absent
353 : // Content-Type reads as absent and parses. Scanning it is what
354 : // keeps the cap ahead of the parser on those routes.
355 6 : Err(_) => true,
356 16266 : });
357 22656 : if is_json && json_depth(&bytes) > MAX_JSON_DEPTH {
358 16 : st.limits.body_too_deep.fetch_add(1, Ordering::Relaxed);
359 16 : return crate::negotiate::ApiError::from(antares_model::NgsiError::BadRequestData(
360 16 : format!("JSON nesting exceeds the {MAX_JSON_DEPTH}-level limit"),
361 16 : ))
362 16 : .into_response();
363 22640 : }
364 22640 : next.run(Request::from_parts(parts, Body::from(bytes)))
365 22640 : .await
366 26786 : }
367 :
368 : #[cfg(test)]
369 : mod tests {
370 : use super::*;
371 :
372 : /// The scan is a bound, not a parser: unbalanced closers must not
373 : /// underflow, and the value it reports is the one the middleware compares
374 : /// against MAX_JSON_DEPTH, so the accept/reject boundary is exact.
375 : #[test]
376 4 : fn depth_scan_survives_unbalanced_and_boundary_input() {
377 4 : assert_eq!(json_depth(b"]]]]"), 0, "stray closers must not underflow");
378 4 : assert_eq!(json_depth(b"}}}{"), 1);
379 4 : assert_eq!(json_depth(b""), 0);
380 4 : assert_eq!(
381 4 : json_depth(br#""{{{{""#),
382 : 0,
383 : "a bare string carries no depth"
384 : );
385 4 : assert_eq!(
386 4 : json_depth(br#"{"a": "\\"}"#),
387 : 1,
388 : "an escaped backslash ends the escape"
389 : );
390 4 : let at_cap = "[".repeat(MAX_JSON_DEPTH) + &"]".repeat(MAX_JSON_DEPTH);
391 4 : assert_eq!(json_depth(at_cap.as_bytes()), MAX_JSON_DEPTH);
392 4 : assert!(
393 4 : json_depth(at_cap.as_bytes()) <= MAX_JSON_DEPTH,
394 : "exactly at the cap is accepted"
395 : );
396 4 : let over = "[".repeat(MAX_JSON_DEPTH + 1) + &"]".repeat(MAX_JSON_DEPTH + 1);
397 4 : assert!(
398 4 : json_depth(over.as_bytes()) > MAX_JSON_DEPTH,
399 : "one over is rejected"
400 : );
401 4 : }
402 :
403 : /// A cap `/q/health` publishes is the cap that fires: an OR chain of
404 : /// `MAX_Q_NODES - 1` terms (the chain node plus its terms) parses, one
405 : /// term more is 5.5.6 TooComplexQuery.
406 : #[test]
407 4 : fn the_published_q_node_cap_is_the_enforced_one() {
408 8 : let chain = |k: usize| vec!["a==1"; k].join("|");
409 4 : assert!(antares_ql::parse_q(&chain(MAX_Q_NODES - 1)).is_ok());
410 4 : assert!(matches!(
411 4 : antares_ql::parse_q(&chain(MAX_Q_NODES)),
412 : Err(antares_model::NgsiError::TooComplexQuery(_))
413 : ));
414 4 : }
415 :
416 : /// Every cap belongs in `/q/health` AND in the operator's chapter: the
417 : /// admin-API chapter prints the whole `limits` object as the answer a
418 : /// memory-store broker gives, and an operator reads a bound from there.
419 : /// The test above pins the key set in code; nothing pinned the chapter,
420 : /// and `maxFedInflight` was published for a release without ever being
421 : /// written down. A cap added here now fails until the chapter has it.
422 : #[test]
423 4 : fn the_documented_health_limits_are_the_published_ones() {
424 4 : let book = std::fs::read_to_string(concat!(
425 : env!("CARGO_MANIFEST_DIR"),
426 : "/../../docs/src/admin-api.md"
427 : ))
428 4 : .expect("the admin API chapter");
429 4 : let block = book
430 4 : .split_once("\"limits\": {")
431 4 : .expect("the limits object in the sample response")
432 4 : .1;
433 4 : let block = block.split_once('}').expect("the end of the object").0;
434 4 : let mut documented: Vec<&str> = block
435 4 : .lines()
436 108 : .filter_map(|l| l.trim().strip_prefix('"'))
437 100 : .filter_map(|l| l.split_once('"'))
438 4 : .map(|(k, _)| k)
439 4 : .collect();
440 4 : documented.sort_unstable();
441 4 : let snap = LimitStats::default().snapshot();
442 4 : let mut published: Vec<&str> = snap
443 4 : .as_object()
444 4 : .expect("object")
445 4 : .keys()
446 4 : .map(String::as_str)
447 4 : .collect();
448 4 : published.sort_unstable();
449 4 : assert_eq!(
450 : documented, published,
451 : "the admin API chapter's limits object is not the one /q/health answers"
452 : );
453 4 : }
454 :
455 : /// /q/health publishes the caps and the rejection counters — and nothing
456 : /// else: no configuration paths, no environment variable values, no
457 : /// internal error text.
458 : #[test]
459 4 : fn health_snapshot_reports_the_caps_and_nothing_internal() {
460 4 : let stats = LimitStats::default();
461 4 : stats.uri_too_long.fetch_add(3, Ordering::Relaxed);
462 4 : let snap = stats.snapshot();
463 4 : let obj = snap.as_object().expect("object");
464 4 : let mut keys: Vec<&str> = obj.keys().map(String::as_str).collect();
465 4 : keys.sort_unstable();
466 4 : assert_eq!(
467 : keys,
468 : [
469 : "changeQueue",
470 : "deliveryWidth",
471 : "deliveryWidthPerTenant",
472 : "maxBatchItems",
473 : "maxBodyBytes",
474 : "maxContextFetches",
475 : "maxFedFanout",
476 : "maxFedInflight",
477 : "maxFedResponseBytes",
478 : "maxFoldDocs",
479 : "maxGeoVertices",
480 : "maxInProcessCallDepth",
481 : "maxJoinLevel",
482 : "maxJsonDepth",
483 : "maxPeerWarnings",
484 : "maxQLinkLookups",
485 : "maxQNodes",
486 : "maxRegexCache",
487 : "maxRegexCacheBytes",
488 : "maxRegexProgramBytes",
489 : "maxTrackedDestinations",
490 : "maxUriBytes",
491 : "rejectedBodyTooDeep",
492 : "rejectedBodyTooLarge",
493 : "rejectedUriTooLong",
494 : ],
495 : "no member beyond the caps, the pipeline ceilings and the counters"
496 : );
497 4 : assert_eq!(snap["rejectedUriTooLong"], 3);
498 4 : assert!(
499 100 : obj.values().all(|v| v.is_number()),
500 : "every member is a number — no strings to leak paths through"
501 : );
502 4 : }
503 :
504 : /// The nesting cap is checked BEFORE any parse — including on the path
505 : /// that carries no Content-Type header at all, which the body parser
506 : /// accepts and parses as JSON (6.3.4 only mandates Content-Length).
507 : /// An unparseable Content-Type follows the same rule.
508 : #[tokio::test]
509 4 : async fn over_depth_body_without_content_type_is_still_rejected() {
510 : use tower::ServiceExt;
511 4 : let st = crate::AppState::new("http://localhost:0".into());
512 4 : let app = axum::Router::new()
513 4 : .route(
514 4 : "/x",
515 4 : axum::routing::post(|| async { StatusCode::NO_CONTENT }),
516 : )
517 4 : .layer(axum::middleware::from_fn_with_state(st, bounds_layer));
518 4 : let deep = "[".repeat(MAX_JSON_DEPTH + 5) + &"]".repeat(MAX_JSON_DEPTH + 5);
519 : // The third case is the one a header map can hold and `to_str`
520 : // cannot read: a Content-Type carrying a byte outside UTF-8. Every
521 : // route reads it as an absent Content-Type, so the scan must too.
522 4 : let unreadable = axum::http::HeaderValue::from_bytes(b"application/\xffjson")
523 4 : .expect("header value from raw bytes");
524 4 : assert!(
525 4 : unreadable.to_str().is_err(),
526 : "the case under test is a header value that cannot be read as text"
527 : );
528 12 : for ct in [
529 4 : None,
530 4 : Some(axum::http::HeaderValue::from_static("application/json")),
531 4 : Some(unreadable),
532 4 : ] {
533 12 : let mut req = Request::post("/x")
534 12 : .header(axum::http::header::CONTENT_LENGTH, deep.len().to_string());
535 12 : if let Some(ct) = ct.clone() {
536 8 : req = req.header(axum::http::header::CONTENT_TYPE, ct);
537 8 : }
538 12 : let resp = app
539 12 : .clone()
540 12 : .oneshot(req.body(Body::from(deep.clone())).expect("req"))
541 12 : .await
542 12 : .expect("resp");
543 12 : assert_eq!(
544 12 : resp.status(),
545 4 : StatusCode::BAD_REQUEST,
546 4 : "an over-deep body must not reach the handler (content-type {ct:?})"
547 4 : );
548 4 : }
549 4 : }
550 :
551 : /// 6.3.4's 413 is about size, and `/q/health` publishes how often it
552 : /// fired. A body the client abandoned mid-flight is a different event: it
553 : /// is not over the cap, it must not be counted as one, and a declared
554 : /// length over the cap must be refused before the broker buffers a byte
555 : /// of it.
556 : #[tokio::test]
557 4 : async fn a_broken_body_is_not_an_over_cap_body() {
558 : use tower::ServiceExt;
559 4 : let st = crate::AppState::new("http://localhost:0".into());
560 4 : let app = axum::Router::new()
561 4 : .route(
562 4 : "/x",
563 4 : axum::routing::post(|| async { StatusCode::NO_CONTENT }),
564 : )
565 4 : .layer(axum::middleware::from_fn_with_state(
566 4 : st.clone(),
567 : bounds_layer,
568 : ));
569 :
570 : // the transport gave up: a declared length the body never delivers
571 4 : let broken = Body::from_stream(futures_util::stream::once(async {
572 4 : Err::<axum::body::Bytes, std::io::Error>(std::io::Error::other("reset"))
573 4 : }));
574 4 : let resp = app
575 4 : .clone()
576 4 : .oneshot(
577 4 : Request::post("/x")
578 4 : .header(axum::http::header::CONTENT_LENGTH, "100")
579 4 : .body(broken)
580 4 : .expect("req"),
581 4 : )
582 4 : .await
583 4 : .expect("resp");
584 4 : assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
585 4 : assert_eq!(
586 4 : st.limits.body_too_large.load(Ordering::Relaxed),
587 : 0,
588 : "a client abort is not a size rejection"
589 : );
590 :
591 : // over the cap for real: 6.3.4's 413, counted
592 4 : let over = vec![b'x'; *MAX_BODY_BYTES + 1];
593 4 : let resp = app
594 4 : .oneshot(
595 4 : Request::post("/x")
596 4 : .header(axum::http::header::CONTENT_LENGTH, over.len().to_string())
597 4 : .body(Body::from(over))
598 4 : .expect("req"),
599 4 : )
600 4 : .await
601 4 : .expect("resp");
602 4 : assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
603 4 : assert_eq!(st.limits.body_too_large.load(Ordering::Relaxed), 1);
604 4 : }
605 :
606 : /// Every cap this module declares is in the payload `/q/health` serves.
607 : /// A cap an operator cannot read is one they cannot tell from a
608 : /// coincidence when a request is refused or a change is dropped, so the
609 : /// list is read out of this file's own source rather than kept by hand:
610 : /// a cap added below without a member above fails here.
611 : #[test]
612 4 : fn every_declared_cap_is_published() {
613 4 : let src = include_str!("bounds.rs");
614 4 : let published = LimitStats::default().snapshot();
615 4 : let published = published.as_object().expect("an object");
616 4 : let mut missing = Vec::new();
617 2872 : for line in src.lines() {
618 2872 : let line = line.trim_start();
619 : // `pub use` re-exports name their cap in another crate; the
620 : // declarations here are the ones this file owns.
621 5744 : for kw in ["pub const ", "pub static "] {
622 5744 : let Some(rest) = line.strip_prefix(kw) else {
623 5680 : continue;
624 : };
625 64 : let Some((name, ty)) = rest.split_once(':') else {
626 0 : continue;
627 : };
628 64 : let name = name.trim();
629 : // A cap is a count; the semaphore built from one is not a
630 : // second cap and has nothing of its own to publish.
631 64 : if !ty.contains("usize") {
632 4 : continue;
633 60 : }
634 : // MAX_URI_BYTES → maxUriBytes
635 60 : let mut camel = String::new();
636 188 : for (i, word) in name.split('_').enumerate() {
637 188 : let lower = word.to_lowercase();
638 188 : if i == 0 {
639 60 : camel.push_str(&lower);
640 60 : } else {
641 128 : let mut c = lower.chars();
642 128 : if let Some(f) = c.next() {
643 128 : camel.extend(f.to_uppercase());
644 128 : camel.push_str(c.as_str());
645 128 : }
646 : }
647 : }
648 60 : if !published.contains_key(&camel) {
649 0 : missing.push(format!("{name} (expected {camel:?})"));
650 60 : }
651 : }
652 : }
653 4 : assert!(
654 4 : missing.is_empty(),
655 : "caps declared here and absent from /q/health: {missing:?}"
656 : );
657 4 : }
658 :
659 : /// A configured count is read once and never re-read, so a value that
660 : /// would stop delivery has no second chance to be corrected: zero mints
661 : /// a semaphore that admits nobody, and a per-tenant share above the
662 : /// width names a ceiling that can never fire. Both fall back rather
663 : /// than take the number as given.
664 : #[test]
665 4 : fn a_configured_count_falls_back_rather_than_disabling_delivery() {
666 4 : assert_eq!(count_from(None, 64, usize::MAX), 64, "unset is the default");
667 4 : assert_eq!(count_from(Some("weeks"), 64, usize::MAX), 64);
668 4 : assert_eq!(count_from(Some(""), 64, usize::MAX), 64);
669 4 : assert_eq!(
670 4 : count_from(Some("0"), 64, usize::MAX),
671 : 64,
672 : "zero would admit nobody"
673 : );
674 4 : assert_eq!(count_from(Some("-8"), 64, usize::MAX), 64);
675 4 : assert_eq!(count_from(Some("256"), 64, usize::MAX), 256);
676 4 : assert_eq!(
677 4 : count_from(Some("512"), 8, 64),
678 : 64,
679 : "a share over the width is the width"
680 : );
681 4 : assert_eq!(
682 4 : count_from(None, 8, 4),
683 : 4,
684 : "the ceiling binds the default too"
685 : );
686 4 : }
687 :
688 : /// A guard on the compiled defaults rather than on the clamp: both
689 : /// counts are read once per process, so a test cannot set one without
690 : /// racing every other test in the binary for the read. What it holds is
691 : /// that the two numbers shipped stay a share and a total, and that
692 : /// neither default is ever lowered to zero — which would admit nobody.
693 : #[test]
694 4 : fn the_published_tenant_share_fits_inside_the_published_width() {
695 4 : let snap = LimitStats::default().snapshot();
696 4 : let width = snap["deliveryWidth"].as_u64().expect("a number");
697 4 : let share = snap["deliveryWidthPerTenant"].as_u64().expect("a number");
698 4 : assert!(share <= width, "share {share} exceeds width {width}");
699 4 : assert!(width > 0 && share > 0, "neither may be zero");
700 4 : }
701 :
702 : #[test]
703 4 : fn depth_scan_is_string_aware() {
704 4 : assert_eq!(json_depth(br#"{"a": [1, {"b": 2}]}"#), 3);
705 4 : assert_eq!(
706 4 : json_depth(br#"{"a": "}]}]}]{[{["}"#),
707 : 1,
708 : "braces in strings don't count"
709 : );
710 4 : assert_eq!(
711 4 : json_depth(br#"{"a": "\"}"}"#),
712 : 1,
713 : "escaped quotes stay in-string"
714 : );
715 4 : let deep = "[".repeat(100) + &"]".repeat(100);
716 4 : assert_eq!(json_depth(deep.as_bytes()), 100);
717 4 : }
718 : }
|
