Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Router assembly: the NGSI-LD route table, the ops-only router and the request layers (tenant, CORS, metrics, trailing slash, 405/501/OPTIONS).
3 :
4 : use super::*;
5 : use tower::layer::util::Stack;
6 : use tower_http::compression::predicate::{And, NotForContentType, Predicate, SizeAbove};
7 : use tower_http::compression::{CompressionLayer, CompressionLevel};
8 :
9 : /// 4.3.5 NGSI-LD API structure: Core API mandatory; Distributed API mandatory for
10 : /// distributed/federated deployments; Temporal API and Registry API integrated
11 : /// locally here (Table 4.3.5-2 row "integrated temporal + integrated Context
12 : /// Registry"); JSONLDContext API implemented; optional Snapshot API offered
13 : /// (5.16, resources 6.36-6.38, NGSILD-Snapshot scoping 6.3.22).
14 : /// Ops-only router: what a pod WITHOUT the api role serves —
15 : /// health, readiness and metrics, nothing else. The shipped antares-worker
16 : /// Deployment used to answer the full read/write NGSI-LD API on its pod IP
17 : /// (and a subscription created there was never KV-synced, because the sync
18 : /// hooks are wired `if roles.api`); a worker now 404s the API surface.
19 12 : pub fn ops_router(state: AppState) -> Router {
20 : // Admin only, deliberately: a worker widening its surface to whatever a
21 : // deployment registered is the same class of accident this router exists
22 : // to prevent.
23 12 : Router::new()
24 12 : .nest(Admin.prefix(), Admin.router(state.clone()))
25 12 : .with_state(state)
26 12 : }
27 :
28 : /// The NGSI-LD API, path by path: `router()` mounts exactly these under the
29 : /// API root, and the policy seam's own test walks the same list, asking the
30 : /// router (OPTIONS, the Allow set) which methods each path serves.
31 18846 : pub fn api_routes() -> Vec<(&'static str, MethodRouter<AppState>)> {
32 18846 : vec![
33 : // entities (6.4/6.5)
34 18846 : (
35 18846 : "/entities",
36 18846 : post(entities::create_entity)
37 18846 : .get(entities::query_entities)
38 18846 : .delete(entities::purge_entities)
39 18846 : .patch(missing_entity_id)
40 18846 : .put(missing_entity_id),
41 18846 : ),
42 18846 : (
43 18846 : "/entities/{id}",
44 18846 : get(entities::retrieve_entity)
45 18846 : .patch(entities::merge_entity)
46 18846 : .put(entities::replace_entity)
47 18846 : .delete(entities::delete_entity),
48 18846 : ),
49 : // attrs (6.6/6.7)
50 18846 : (
51 18846 : "/entities/{id}/attrs",
52 18846 : post(attrs::append_attrs).patch(attrs::update_attrs),
53 18846 : ),
54 18846 : (
55 18846 : "/entities/{id}/attrs/{attr}",
56 18846 : // GET is the 2.0 #14 pre-adoption
57 18846 : get(entities::retrieve_entity_attr)
58 18846 : .patch(attrs::partial_update_attr)
59 18846 : .put(attrs::replace_attr)
60 18846 : .delete(attrs::delete_attr),
61 18846 : ),
62 : // 2.0 #15 pre-adoption: the bare attribute value, and PUT is Set
63 : // Attribute Value (TS 104 176 7.6.3.2)
64 18846 : (
65 18846 : "/entities/{id}/attrs/{attr}/value",
66 18846 : get(entities::retrieve_entity_attr_value).put(attrs::set_attr_value),
67 18846 : ),
68 : // batch (6.14–6.17, 6.23, 6.31)
69 18846 : ("/entityOperations/create", post(batch::batch_create)),
70 18846 : ("/entityOperations/upsert", post(batch::batch_upsert)),
71 18846 : ("/entityOperations/update", post(batch::batch_update)),
72 18846 : ("/entityOperations/delete", post(batch::batch_delete)),
73 18846 : ("/entityOperations/merge", post(batch::batch_merge)),
74 18846 : ("/entityOperations/query", post(batch::batch_query)),
75 : // EntityMaps (5.14; 6.32/6.34/6.35)
76 18846 : (
77 18846 : "/entityMaps",
78 18846 : get(entity_maps::create_entity_map).post(entity_maps::create_entity_map_post),
79 18846 : ),
80 18846 : (
81 18846 : "/entityMaps/{id}",
82 18846 : get(entity_maps::retrieve_entity_map)
83 18846 : .patch(entity_maps::update_entity_map)
84 18846 : .delete(entity_maps::delete_entity_map),
85 18846 : ),
86 18846 : (
87 18846 : "/temporal/entityMaps",
88 18846 : get(entity_maps::create_temporal_entity_map)
89 18846 : .post(entity_maps::create_temporal_entity_map_post),
90 18846 : ),
91 : // subscriptions (6.10/6.11)
92 18846 : (
93 18846 : "/subscriptions",
94 18846 : post(subscriptions::create_subscription).get(subscriptions::query_subscriptions),
95 18846 : ),
96 18846 : (
97 18846 : "/subscriptions/{id}",
98 18846 : get(subscriptions::retrieve_subscription)
99 18846 : .patch(subscriptions::update_subscription)
100 18846 : .delete(subscriptions::delete_subscription),
101 18846 : ),
102 : // csourceRegistrations (6.8/6.9)
103 18846 : (
104 18846 : "/csourceRegistrations",
105 18846 : post(csource::create_registration).get(csource::query_registrations),
106 18846 : ),
107 18846 : (
108 18846 : "/csourceRegistrations/{id}",
109 18846 : get(csource::retrieve_registration)
110 18846 : .patch(csource::update_registration)
111 18846 : .delete(csource::delete_registration),
112 18846 : ),
113 : // csourceSubscriptions (6.12/6.13)
114 18846 : (
115 18846 : "/csourceSubscriptions",
116 18846 : post(subscriptions::create_csource_subscription)
117 18846 : .get(subscriptions::query_csource_subscriptions),
118 18846 : ),
119 18846 : (
120 18846 : "/csourceSubscriptions/{id}",
121 18846 : get(subscriptions::retrieve_csource_subscription)
122 18846 : .patch(subscriptions::update_csource_subscription)
123 18846 : .delete(subscriptions::delete_csource_subscription),
124 18846 : ),
125 : // temporal (6.18–6.22, 6.24)
126 18846 : (
127 18846 : "/temporal/entities",
128 18846 : post(temporal::upsert_temporal).get(temporal::query_temporal),
129 18846 : ),
130 18846 : (
131 18846 : "/temporal/entities/{id}",
132 18846 : get(temporal::retrieve_temporal).delete(temporal::delete_temporal),
133 18846 : ),
134 18846 : (
135 18846 : "/temporal/entities/{id}/attrs",
136 18846 : post(temporal::add_temporal_attrs),
137 18846 : ),
138 18846 : (
139 18846 : "/temporal/entities/{id}/attrs/{attr}",
140 18846 : delete(temporal::delete_temporal_attr),
141 18846 : ),
142 18846 : (
143 18846 : "/temporal/entities/{id}/attrs/{attr}/{instance}",
144 18846 : patch(temporal::modify_temporal_instance).delete(temporal::delete_temporal_instance),
145 18846 : ),
146 18846 : (
147 18846 : "/temporal/entityOperations/query",
148 18846 : post(temporal::batch_temporal_query),
149 18846 : ),
150 : // discovery (6.25–6.28)
151 18846 : ("/types", get(types_attrs::entity_types)),
152 18846 : ("/types/{type}", get(types_attrs::entity_type_info)),
153 18846 : ("/attributes", get(types_attrs::attributes)),
154 18846 : ("/attributes/{attr}", get(types_attrs::attribute_info)),
155 : // jsonldContexts (6.29/6.30)
156 18846 : (
157 18846 : "/jsonldContexts",
158 18846 : post(contexts::add_context).get(contexts::list_contexts),
159 18846 : ),
160 18846 : (
161 18846 : "/jsonldContexts/{id}",
162 18846 : get(contexts::serve_context).delete(contexts::delete_context),
163 18846 : ),
164 : // snapshots (5.16; 6.36-6.38, optional API group — offered)
165 18846 : (
166 18846 : "/snapshots",
167 18846 : post(snapshots::create_snapshot).delete(snapshots::purge_snapshots),
168 18846 : ),
169 18846 : (
170 18846 : "/snapshots/{id}",
171 18846 : get(snapshots::retrieve_snapshot)
172 18846 : .patch(snapshots::update_snapshot)
173 18846 : .delete(snapshots::delete_snapshot),
174 18846 : ),
175 18846 : ("/snapshots/{id}/clone", post(snapshots::clone_snapshot)),
176 : // info (6.33)
177 18846 : ("/info/sourceIdentity", get(source_identity)),
178 : ]
179 18846 : }
180 :
181 : /// The broker's whole HTTP surface, under trailing-slash normalization: every
182 : /// host of it (the binary, the browser build, an in-process forward, a test)
183 : /// serves `/entities/{id}/attrs/` as `/entities/{id}/attrs`, the form 5.6.2
184 : /// and 5.6.3 forwards are sent in.
185 18840 : pub fn router(state: AppState) -> Router {
186 18840 : let served = routes(state);
187 18840 : Router::new().fallback_service(tower::util::MapRequest::new(
188 18840 : served,
189 : trim_trailing_slash::<axum::body::Body>,
190 : ))
191 18840 : }
192 :
193 : /// gzip or br for a client whose Accept-Encoding asks for it, laid
194 : /// over [`router()`] by the binary that serves the network and by nothing else:
195 : /// an in-process call (`AppState::call`, a forward) parses the body it gets
196 : /// back and must read plain JSON, and each nested frame would grow the stack
197 : /// a façade chain spends. A body that already has a Content-Encoding or a
198 : /// Content-Range is never touched, and `Vary: Accept-Encoding` is appended
199 : /// once to whatever Vary the handler set. Request bodies stay as sent: no
200 : /// decompression, so the body bound holds.
201 : ///
202 : /// Levels (the best size per CPU on real answers): brotli quality 2
203 : /// and gzip level 1, each its own layer because tower-http gives one layer one
204 : /// quality. The br layer sits inside: it packs whenever the client accepts br,
205 : /// and the gzip layer outside leaves an encoded answer alone, so gzip serves
206 : /// only a client without br. [`CompressWhen`] leaves a body of known size
207 : /// under 1 KiB plain, and text/event-stream, images and gRPC.
208 86 : pub fn compression() -> Stack<CompressionLayer<CompressWhen>, CompressionLayer<CompressWhen>> {
209 86 : let br = CompressionLayer::new()
210 86 : .gzip(false)
211 86 : .quality(CompressionLevel::Precise(2))
212 86 : .compress_when(compress_when());
213 86 : let gzip = CompressionLayer::new()
214 86 : .br(false)
215 86 : .quality(CompressionLevel::Precise(1))
216 86 : .compress_when(compress_when());
217 86 : Stack::new(br, gzip)
218 86 : }
219 :
220 : /// The predicate of [`compression()`]: tower-http's default with its 32-byte
221 : /// minimum raised to 1 KiB, since a smaller body saves under 700 bytes.
222 : pub type CompressWhen =
223 : And<And<And<SizeAbove, NotForContentType>, NotForContentType>, NotForContentType>;
224 :
225 172 : fn compress_when() -> CompressWhen {
226 172 : SizeAbove::new(1024)
227 172 : .and(NotForContentType::GRPC)
228 172 : .and(NotForContentType::IMAGES)
229 172 : .and(NotForContentType::SSE)
230 172 : }
231 :
232 : /// tower-http's `NormalizePath::trim_trailing_slash`: trailing slashes and a
233 : /// leading `//` collapse to one path, `/` alone for a path of slashes, and
234 : /// the query is kept.
235 79212 : pub fn trim_trailing_slash<B>(mut req: axum::http::Request<B>) -> axum::http::Request<B> {
236 79212 : let path = req.uri().path();
237 79212 : if !path.ends_with('/') && !path.starts_with("//") {
238 73164 : return req;
239 6048 : }
240 6048 : let trimmed = format!("/{}", path.trim_matches('/'));
241 6048 : let path_and_query = match req.uri().query() {
242 1386 : Some(query) => format!("{trimmed}?{query}"),
243 4662 : None => trimmed,
244 : };
245 6048 : let mut parts = req.uri().clone().into_parts();
246 : // A path cut from a valid one stays valid; were it ever not, the request
247 : // goes on as sent rather than failing here.
248 6048 : if let Ok(pq) = path_and_query.parse() {
249 6048 : parts.path_and_query = Some(pq);
250 6048 : if let Ok(uri) = axum::http::Uri::from_parts(parts) {
251 6048 : *req.uri_mut() = uri;
252 6048 : }
253 0 : }
254 6048 : req
255 79212 : }
256 :
257 18840 : pub(super) fn routes(state: AppState) -> Router {
258 18840 : let api = api_routes()
259 18840 : .into_iter()
260 678240 : .fold(Router::new(), |api, (path, methods)| {
261 678240 : api.route(path, methods)
262 678240 : });
263 :
264 : // 5.8.1.4 consumer half: where forwarded subscription copies point their
265 : // notifications; remapped to the original subscriber. CIM 009 defines no
266 : // path for it — 5.2.15 makes a notification endpoint any URI, and 6.2
267 : // standardizes only what hangs under the API root — so it lives outside
268 : // the `/ngsi-ld` prefix ETSI owns, under this broker's own versioned
269 : // peer-facing root (ADR-0019). Being outside the API nest, it carries the
270 : // bounds wall and the body limit itself: a peer-facing write path must
271 : // not be the one route where the documented caps do not apply.
272 18840 : let remote_notify = Router::new()
273 18840 : .route("/ex/v1/remote-notify", post(distsub::remote_notify))
274 18840 : .layer(axum::extract::DefaultBodyLimit::max(
275 18840 : *bounds::MAX_BODY_BYTES,
276 : ))
277 18840 : .layer(axum::middleware::from_fn_with_state(
278 18840 : state.clone(),
279 : bounds::bounds_layer,
280 : ));
281 :
282 : // Every registered surface, each under its own reserved prefix — the
283 : // admin one by default. `with_surface` already refused a prefix outside
284 : // /q and /x and any overlap, so the merge here cannot shadow a route.
285 : // The caps are the broker's, not the API nest's: these routes carry the
286 : // bounds wall and the body limit for the same reason the peer-facing
287 : // notification endpoint above does — /q deletes Tenants and replays dead
288 : // letters, and a deployment's own /x routes are no less a way in.
289 18840 : let mut surfaces = Router::new();
290 18964 : for s in state.surfaces.iter() {
291 18964 : surfaces = surfaces.nest(s.prefix(), s.router(state.clone()));
292 18964 : }
293 18840 : let surfaces = surfaces
294 18840 : .layer(axum::extract::DefaultBodyLimit::max(
295 18840 : *bounds::MAX_BODY_BYTES,
296 : ))
297 18840 : .layer(axum::middleware::from_fn_with_state(
298 18840 : state.clone(),
299 : bounds::bounds_layer,
300 : ));
301 :
302 18840 : surfaces
303 18840 : .merge(remote_notify)
304 : // 6.3.6/6.3.21: Prefer: ngsi-ld=<version> → 4.3.6.8 amendment +
305 : // Preference-Applied (+203 when altered) on every API response.
306 : // OPTIONS (2.0 #59 pre-adoption): axum's MethodRouter already
307 : // computes the exact per-route Allow set for its 405s — the layer
308 : // turns an OPTIONS 405 into 204 + that same Allow. HEAD (#58) needs
309 : // nothing: axum's get() serves HEAD natively.
310 18840 : .nest(
311 18840 : API_ROOT,
312 18840 : api.layer(axum::middleware::from_fn(conformance::prefer_version_layer))
313 : // The temporal seam's drain: every write's history events
314 : // land in one driver call once the handler is done.
315 18840 : .layer(axum::middleware::from_fn_with_state(
316 18840 : state.clone(),
317 : history::layer,
318 : ))
319 : // 6.3.22 / 5.5.15: NGSILD-Snapshot resolves to the
320 : // snapshot's synthetic tenant, so the handlers below serve
321 : // the frozen copy.
322 18840 : .layer(axum::middleware::from_fn_with_state(
323 18840 : state.clone(),
324 : snapshots::snapshot_layer,
325 : ))
326 : // 5.5.10: non-create operations targeting a non-existing
327 : // Tenant answer NonexistentTenant 404; create operations
328 : // implicitly create the Tenant. Outside the snapshot layer,
329 : // so a client naming one of the broker's own internal
330 : // tenants is refused before that layer legitimately sets
331 : // one.
332 18840 : .layer(axum::middleware::from_fn_with_state(
333 18840 : state.clone(),
334 : tenant_exists_layer,
335 : ))
336 : // a request without a subject needs the admin
337 : // bearer, before the tenant lookup so a 404 tells such a
338 : // caller nothing about which Tenants exist.
339 18840 : .layer(axum::middleware::from_fn_with_state(
340 18840 : state.clone(),
341 : crate::admin_auth::subject_or_admin,
342 : ))
343 : // Bounds wall: URI length, body size, JSON depth — checked
344 : // before any parse (size-check-before-parse), and outside the
345 : // tenant and snapshot lookups so 6.3.4's bare 411/414 is not
346 : // spent on a store round-trip or masked by its 404.
347 18840 : .layer(axum::middleware::from_fn_with_state(
348 18840 : state.clone(),
349 : bounds::bounds_layer,
350 : ))
351 : // axum's built-in extractor limit defaults to 2 MiB and fires
352 : // BEFORE the bounds wall's documented cap — a 3 MiB body was
353 : // 413'd although /q/health advertises maxBodyBytes = 4 MiB
354 : // (found by mutation-testing the 413 wall). One
355 : // number governs both walls.
356 18840 : .layer(axum::extract::DefaultBodyLimit::max(
357 18840 : *bounds::MAX_BODY_BYTES,
358 : ))
359 : // 6.3.14, last so it sees every exit: outside the snapshot
360 : // layer, which rewrites the request header to the snapshot's
361 : // internal tenant, and outside the bounds wall, whose bare
362 : // 411/414 leave through no handler at all.
363 18840 : .layer(axum::middleware::from_fn(echo_tenant_layer)),
364 : )
365 18840 : .fallback(not_found)
366 : // a panicking handler answers 500 instead of unwinding the
367 : // connection task; inside the metrics layer so the 500 is counted.
368 18840 : .layer(tower_http::catch_panic::CatchPanicLayer::custom(panic_500))
369 : // outermost on purpose: axum attaches the 405 Allow header in the
370 : // Router itself, above any nested layer — only a layer wrapping the
371 : // WHOLE router sees it.
372 18840 : .layer(axum::middleware::from_fn(options_204))
373 : // TS 104 176 6.3.2: an unrecognised verb is 501 before any routing,
374 : // so the method router's 405 stays the answer for a recognised one.
375 18840 : .layer(axum::middleware::from_fn(unrecognised_method_501))
376 : // Outermost so the duration covers the full stack.
377 18840 : .layer(axum::middleware::from_fn(http_metrics_layer))
378 : // absent knob = no layer at all: a bare OPTIONS keeps its 204 + Allow
379 18840 : .layer(tower::util::option_layer(cors_layer()))
380 18840 : .with_state(state)
381 18840 : }
382 :
383 : /// A handler that panicked: 500 InternalError, and the broker goes on
384 : /// serving. The panic message stays in the log (the default hook prints it),
385 : /// never in the body, where it could carry a stored value.
386 4 : pub(super) fn panic_500(_: Box<dyn std::any::Any + Send + 'static>) -> Response {
387 4 : ApiError::from(NgsiError::InternalError("the request failed".into())).into_response()
388 4 : }
389 :
390 : /// Browser access is off unless ANTARES_CORS_ORIGINS names the origins
391 : /// (comma-separated, or `*` for any). NGSI-LD says nothing about CORS; the
392 : /// Link header and the NGSILD-Tenant / NGSILD-Results-Count headers are
393 : /// exposed so a browser client can read them.
394 18840 : pub(super) fn cors_layer() -> Option<tower_http::cors::CorsLayer> {
395 : use tower_http::cors::{AllowOrigin, CorsLayer};
396 18840 : let spec = std::env::var("ANTARES_CORS_ORIGINS")
397 18840 : .ok()
398 18840 : .filter(|s| !s.trim().is_empty())?;
399 0 : let origin = if spec.trim() == "*" {
400 0 : AllowOrigin::any()
401 : } else {
402 0 : AllowOrigin::list(
403 0 : spec.split(',')
404 0 : .filter_map(|o| o.trim().parse::<axum::http::HeaderValue>().ok()),
405 : )
406 : };
407 0 : let layer = CorsLayer::new()
408 0 : .allow_origin(origin)
409 0 : .allow_methods(tower_http::cors::Any)
410 0 : .allow_headers(tower_http::cors::Any)
411 0 : .expose_headers([
412 0 : axum::http::header::LINK,
413 0 : axum::http::HeaderName::from_static("ngsild-tenant"),
414 0 : axum::http::HeaderName::from_static("ngsild-results-count"),
415 0 : ]);
416 0 : Some(layer)
417 18840 : }
418 :
419 : /// Tenants the broker mints for its own bookkeeping: the snapshot module's
420 : /// internal and per-snapshot tenants share a prefix, the distributed
421 : /// subscription inbound index is a single fixed name. A client-supplied
422 : /// tenant may not name any of them — it would put request-shaped writes in
423 : /// the same keyspace the broker keeps its own state in. The names live on
424 : /// `TenantId`, where the constructor a client's name goes through refuses
425 : /// them; the admin surfaces below ask by name, before any parse.
426 2742 : pub(super) fn reserved_tenant(raw: &str) -> bool {
427 2742 : TenantId::is_reserved_str(raw)
428 2742 : }
429 :
430 : /// 5.5.10 Multi-Tenant Behaviour: Tenants are created implicitly by create
431 : /// operations (Create Entity 5.6.1, Batch Create/Upsert 5.6.7/5.6.8, Create
432 : /// Temporal 5.6.11, Create Subscription 5.8.1, Register Context Source
433 : /// 5.9.2, Create CSource Subscription 5.11.2); "all other NGSI-LD
434 : /// operations … that target a non-existing Tenant should raise an error of
435 : /// type NonexistentTenant". Malformed tenant headers pass through — the
436 : /// handler's own parse answers 400.
437 55158 : pub(super) async fn tenant_exists_layer(
438 55158 : axum::extract::State(st): axum::extract::State<AppState>,
439 55158 : req: axum::http::Request<axum::body::Body>,
440 55158 : next: axum::middleware::Next,
441 55158 : ) -> Response {
442 : // 6.3.14: the tenant namespace the broker mints for itself is not a
443 : // legal client tenant. Snapshots run on internal tenants named
444 : // "snap-index" (the synthetic-tenant reverse index) and "snap-<uuid>"
445 : // (one per snapshot); a client naming one would read and delete another
446 : // tenant's snapshot bookkeeping. Rejected here, ahead of the snapshot
447 : // layer that legitimately rewrites the header to such a tenant.
448 : // Read once, and refuse a request that names two Tenants (or one in a
449 : // form no header can carry): the guard below and the handler further in
450 : // have to be deciding about the same value, or the wall inspects one
451 : // name while the operation runs on another.
452 55158 : let named = match crate::negotiate::single_header(req.headers(), "NGSILD-Tenant") {
453 55154 : Ok(named) => named,
454 4 : Err(e) => return e.into_response(),
455 : };
456 55154 : if let Some(raw) = &named {
457 2530 : if reserved_tenant(raw) {
458 48 : return crate::negotiate::ApiError::from(NgsiError::BadRequestData(format!(
459 48 : "invalid NGSILD-Tenant value: {raw:?}"
460 48 : )))
461 48 : .into_response();
462 2482 : }
463 52624 : }
464 55106 : let path = req.uri().path().trim_start_matches(API_ROOT);
465 : // 7.8.2 lists the creating operations; Batch Upsert creates Entities too
466 : // and joins them unless ANTARES_IMPLICIT_TENANTS=listed.
467 55106 : let implicit_create = req.method() == axum::http::Method::POST
468 18600 : && (matches!(
469 20752 : path,
470 20752 : "/entities"
471 8070 : | "/entityOperations/create"
472 7860 : | "/temporal/entities"
473 6574 : | "/subscriptions"
474 5342 : | "/csourceRegistrations"
475 2414 : | "/csourceSubscriptions"
476 2152 : ) || (path == "/entityOperations/upsert" && !st.profile.implicit_tenants_listed_only));
477 : // tenant-independent resources: broker identity (6.33) answers for any
478 : // tenant (the per-tenant alias exists before any data does), and the
479 : // @context resources are keyed by local id across tenants (5.13) — a
480 : // Cached row belongs to no tenant, so the existence of the requesting
481 : // one decides nothing here. Ownership of a Hosted row is still enforced,
482 : // by `contexts::row_visible` on every serve, list and delete. A POST
483 : // stores a Hosted row its Tenant owns, so that Tenant must exist:
484 : // a row of a Tenant that never existed could not be
485 : // purged with it. DELETE stays free so such a row can still go.
486 55106 : let tenant_free = path.starts_with("/info/")
487 55044 : || (path.starts_with("/jsonldContexts") && req.method() != axum::http::Method::POST);
488 55106 : if !implicit_create && !tenant_free {
489 35254 : if let Some(t) = named
490 35254 : .as_deref()
491 35254 : .and_then(|s| antares_model::TenantId::new(s).ok())
492 : {
493 : // a Tenant the store said exists a moment ago is not
494 : // asked about again; only "exists" is remembered.
495 756 : let known = st.known_tenants.holds(&t);
496 756 : let exists = if known {
497 502 : Ok(true)
498 : } else {
499 254 : st.store.tenant_exists(&t).await
500 : };
501 700 : match exists {
502 : Ok(false) => {
503 56 : let mut resp = crate::negotiate::ApiError::from(NgsiError::NonexistentTenant(
504 56 : format!("tenant {} does not exist", t.as_str()),
505 56 : ))
506 56 : .into_response();
507 : // 6.3.14: a request-supplied NGSILD-Tenant SHALL be
508 : // present in the response — error responses included.
509 56 : crate::negotiate::echo_tenant(&t, &mut resp);
510 56 : return resp;
511 : }
512 0 : Err(e) => return crate::negotiate::ApiError::from(e).into_response(),
513 198 : Ok(true) if !known => st.known_tenants.remember(&t),
514 502 : Ok(true) => {}
515 : }
516 34498 : }
517 19852 : }
518 55050 : next.run(req).await
519 55158 : }
520 :
521 : /// 6.3.14: "If the HTTP header `NGSILD-Tenant` is present in the HTTP
522 : /// request, it shall also be present in HTTP response." The sentence admits
523 : /// no exception, and the responses that most need it are the ones no handler
524 : /// shapes: a rejection from the bounds wall, a 405, a body that never parsed.
525 : /// An echo the handlers own is an echo some handler forgets, so it is read
526 : /// once here, from the request the CALLER sent — the snapshot layer below
527 : /// rewrites that header to the snapshot's internal tenant, which must never
528 : /// reach the caller.
529 55230 : pub(super) async fn echo_tenant_layer(
530 55230 : req: axum::http::Request<axum::body::Body>,
531 55230 : next: axum::middleware::Next,
532 55230 : ) -> Response {
533 : // A repeated header names no Tenant, so there is none to echo; the
534 : // BadRequestData the wall answers says so on its own.
535 55230 : let tenant = crate::negotiate::single_header(req.headers(), "NGSILD-Tenant")
536 55230 : .ok()
537 55230 : .flatten()
538 55230 : .and_then(|s| TenantId::new(&s).ok());
539 55230 : let mut resp = next.run(req).await;
540 55230 : if let Some(t) = tenant {
541 2498 : if !resp.headers().contains_key("NGSILD-Tenant") {
542 118 : crate::negotiate::echo_tenant(&t, &mut resp);
543 2380 : }
544 52732 : }
545 55230 : resp
546 55230 : }
547 :
548 : /// Bounded method label for the request metrics: standard HTTP methods
549 : /// pass through, anything else (extension methods are minted by the
550 : /// client) collapses to "OTHER" so the label's cardinality stays bounded.
551 112016 : pub(super) fn metric_method(m: &str) -> &'static str {
552 112016 : match m {
553 112016 : "GET" => "GET",
554 92692 : "POST" => "POST",
555 50820 : "PATCH" => "PATCH",
556 21944 : "PUT" => "PUT",
557 21232 : "DELETE" => "DELETE",
558 828 : "OPTIONS" => "OPTIONS",
559 380 : "HEAD" => "HEAD",
560 48 : _ => "OTHER",
561 : }
562 112016 : }
563 :
564 : /// Request counter + duration histogram, labelled by method and status
565 : /// class only (bounded cardinality).
566 56000 : pub(super) async fn http_metrics_layer(
567 56000 : req: axum::http::Request<axum::body::Body>,
568 56000 : next: axum::middleware::Next,
569 56000 : ) -> Response {
570 56000 : let method = metric_method(req.method().as_str());
571 : // Clock rule: std Instant panics on wasm32.
572 : #[cfg(not(target_arch = "wasm32"))]
573 56000 : let start = std::time::Instant::now();
574 : #[cfg(target_arch = "wasm32")]
575 : let start = web_time::Instant::now();
576 56000 : let resp = next.run(req).await;
577 56000 : let class = match resp.status().as_u16() {
578 56000 : 100..=199 => "1xx",
579 56000 : 200..=299 => "2xx",
580 8432 : 300..=399 => "3xx",
581 8432 : 400..=499 => "4xx",
582 214 : _ => "5xx",
583 : };
584 56000 : metrics::counter!("antares_http_requests_total", "method" => method, "status" => class)
585 56000 : .increment(1);
586 56000 : metrics::histogram!("antares_http_request_duration_seconds", "method" => method)
587 56000 : .record(start.elapsed().as_secs_f64());
588 56000 : resp
589 56000 : }
590 :
591 : /// TS 104 176 6.3.2: "Not Implemented (501) ... when a client invokes an
592 : /// HTTP verb that is unrecognized or not implemented". The recognised set is
593 : /// the one [`metric_method`] names; CONNECT, TRACE and every extension
594 : /// method are 501. A bare status like 6.3.4's 411 and 415: the error table
595 : /// names no error type for it.
596 56000 : pub(super) async fn unrecognised_method_501(
597 56000 : req: axum::http::Request<axum::body::Body>,
598 56000 : next: axum::middleware::Next,
599 56000 : ) -> Response {
600 56000 : if metric_method(req.method().as_str()) == "OTHER" {
601 20 : return StatusCode::NOT_IMPLEMENTED.into_response();
602 55980 : }
603 55980 : next.run(req).await
604 56000 : }
605 :
606 : /// 2.0 #59 pre-adoption: OPTIONS → 204 No Content + the Allow set the
607 : /// method router computed. Non-OPTIONS traffic passes through untouched.
608 55980 : pub(super) async fn options_204(
609 55980 : req: axum::http::Request<axum::body::Body>,
610 55980 : next: axum::middleware::Next,
611 55980 : ) -> Response {
612 55980 : let is_options = req.method() == axum::http::Method::OPTIONS;
613 55980 : let resp = next.run(req).await;
614 55980 : if is_options && resp.status() == axum::http::StatusCode::METHOD_NOT_ALLOWED {
615 : // Keep the response PARTS: axum carries the computed Allow set as a
616 : // response extension and turns it into the header above all layers —
617 : // preserving the extensions preserves the Allow header.
618 224 : let (mut parts, _) = resp.into_parts();
619 224 : parts.status = axum::http::StatusCode::NO_CONTENT;
620 224 : return Response::from_parts(parts, axum::body::Body::empty());
621 55756 : }
622 55756 : resp
623 55980 : }
624 :
625 : /// PATCH/PUT on the entities collection: the entity id is missing — 400.
626 16 : pub(super) async fn missing_entity_id(headers: HeaderMap) -> Response {
627 16 : let tenant = tenant_from(&headers).unwrap_or_default();
628 16 : let mut resp = ApiError::from(NgsiError::BadRequestData(
629 16 : "entity id is required in the request path".into(),
630 16 : ))
631 16 : .into_response();
632 16 : echo_tenant(&tenant, &mut resp);
633 16 : resp
634 16 : }
635 :
636 32 : pub(super) async fn not_found(headers: HeaderMap, uri: axum::http::Uri) -> Response {
637 32 : let path = uri.path().to_owned();
638 32 : let tenant = tenant_from(&headers).unwrap_or_default();
639 : // A path with an empty segment (…/attrs//{x}) names a resource whose
640 : // methods don't apply — 405 per the suite (016_02_04/06).
641 32 : if path.starts_with(API_ROOT) && path.contains("//") {
642 8 : let mut resp = axum::http::StatusCode::METHOD_NOT_ALLOWED.into_response();
643 8 : echo_tenant(&tenant, &mut resp);
644 8 : return resp;
645 24 : }
646 24 : let mut resp =
647 24 : ApiError::from(NgsiError::ResourceNotFound(format!("unknown path {path}"))).into_response();
648 24 : echo_tenant(&tenant, &mut resp);
649 24 : resp
650 32 : }
651 :
652 : #[cfg(test)]
653 : mod trim_trailing_slash_tests {
654 : use super::trim_trailing_slash;
655 :
656 28 : fn trimmed(uri: &str) -> String {
657 28 : let req = axum::http::Request::builder()
658 28 : .uri(uri)
659 28 : .body(())
660 28 : .unwrap_or_default();
661 28 : trim_trailing_slash(req).uri().to_string()
662 28 : }
663 :
664 : /// The NGSI-LD API and every other surface: tower-http's trim, the query
665 : /// kept.
666 : #[test]
667 4 : fn trims_every_path() {
668 28 : for (sent, served) in [
669 4 : ("/ngsi-ld/v1/entities/", "/ngsi-ld/v1/entities"),
670 4 : (
671 4 : "/ngsi-ld/v1/entities//?type=Car",
672 4 : "/ngsi-ld/v1/entities?type=Car",
673 4 : ),
674 4 : ("//ngsi-ld/v1/entities", "/ngsi-ld/v1/entities"),
675 4 : ("/", "/"),
676 4 : ("///", "/"),
677 4 : ("/x/example/", "/x/example"),
678 4 : ("/q/tenants/", "/q/tenants"),
679 4 : ] {
680 28 : assert_eq!(trimmed(sent), served, "{sent}");
681 : }
682 4 : }
683 : }
|
