Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! One EntityMap document (5.2.39) and the rules for using one: store it
3 : //! under its tenant with a lifetime, read it back only while it is alive,
4 : //! take the candidate ids of a page out of it, merge the registrations a
5 : //! distributed query reached into it, and serve a retrieve through the map
6 : //! a client presented. The /entityMaps resource itself is `entity_maps`,
7 : //! which composes this over the queries in `entities` and `temporal`.
8 :
9 : use crate::negotiate::*;
10 : use crate::state::AppState;
11 : use antares_model::{NgsiError, TenantId};
12 : use antares_store::CurrentStateDriverExt;
13 : use antares_store::Kind;
14 : use axum::http::{HeaderMap, StatusCode};
15 : use axum::response::Response;
16 : use serde_json::{json, Map, Value};
17 : use std::collections::HashMap;
18 :
19 : /// Per-tenant EntityMap cap (every buffer bounded); earliest-expiring evicted.
20 : pub(crate) const MAX_MAPS_PER_TENANT: usize = 512;
21 :
22 : /// Default lifetime when the client suggests none — 5.5.14: "the caching
23 : /// strategy and expiry time … depend on implementation specific
24 : /// configurations".
25 : pub(crate) const DEFAULT_LIFETIME_SECS: i64 = 3600;
26 :
27 : /// Ceiling on client-suggested lifetimes — 6.4.3.2-1: "the actual expiresAt
28 : /// time of the EntityMap shall be set by the Context Broker or Context
29 : /// Source, possibly overriding the requested duration".
30 : pub(crate) const MAX_LIFETIME_SECS: i64 = 86_400;
31 :
32 3853 : pub(crate) fn dt(s: &str) -> Option<chrono::DateTime<chrono::Utc>> {
33 3853 : chrono::DateTime::parse_from_rfc3339(s)
34 3853 : .ok()
35 3853 : .map(|d| d.with_timezone(&chrono::Utc))
36 3853 : }
37 :
38 : /// Record whose map this is, so 5.5.14's "cannot be accessed" covers a map
39 : /// built for someone else. A broker-internal member: 5.2.39 defines none, no
40 : /// served map carries it, and a client cannot supply one because every map
41 : /// the broker stores is one the broker built.
42 254 : fn stamp_subject(doc: &mut Value, tenant: &TenantId, headers: &HeaderMap) {
43 10 : if let (Some(o), Some(subject)) = (
44 254 : doc.as_object_mut(),
45 254 : crate::policy::subject_member(&crate::policy::subject_of(tenant, headers)),
46 10 : ) {
47 10 : o.insert(crate::policy::SUBJECT_MEMBER.into(), subject);
48 244 : }
49 254 : }
50 :
51 : /// Fetch a live EntityMap; an expired one "cannot be accessed" (5.5.14).
52 : /// Reading is a read: the row behind a map this refuses is freed by
53 : /// `sweep_expired_maps` on the sweep tick, not by the request that found
54 : /// it. Maps live in the store (Kind::EntityMap) so persistent modes survive
55 : /// restarts.
56 520 : pub(crate) async fn map_get(
57 520 : st: &AppState,
58 520 : tenant: &TenantId,
59 520 : id: &str,
60 520 : ) -> Result<Option<Value>, NgsiError> {
61 520 : let Some(doc) = st.store.get(tenant, Kind::EntityMap, id).await? else {
62 82 : return Ok(None);
63 : };
64 430 : Ok(map_live(&doc).then_some(doc))
65 520 : }
66 :
67 : /// 5.5.14 is a positive condition: a map is usable only while a READABLE
68 : /// expiry is still in the future. Judging "expired" instead lets a map whose
69 : /// `expiresAt` is missing or unparseable outlive every ceiling. One
70 : /// definition, so the sweep reaps exactly what a read refuses.
71 1455 : fn map_live(doc: &Value) -> bool {
72 1455 : doc.get("expiresAt")
73 1455 : .and_then(Value::as_str)
74 1455 : .and_then(dt)
75 1455 : .is_some_and(|e| e > chrono::Utc::now())
76 1455 : }
77 :
78 : /// 4.22 for EntityMaps: a map a read will not serve is a map nothing can
79 : /// reach, and this is what removes it.
80 145070 : pub(crate) async fn sweep_expired_maps(st: &AppState, tenant: &TenantId) -> usize {
81 145070 : let mut dead: Vec<String> = Vec::new();
82 145070 : if crate::csource::walk_docs(st, tenant, Kind::EntityMap, |doc| {
83 1025 : if !map_live(&doc) {
84 24 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
85 24 : dead.push(id.to_owned());
86 24 : }
87 1001 : }
88 1025 : Ok(())
89 1025 : })
90 145070 : .await
91 145070 : .is_err()
92 : {
93 12 : return 0;
94 145058 : }
95 145058 : let mut n: usize = 0;
96 145058 : for id in dead {
97 24 : if st
98 24 : .store
99 24 : .delete(tenant, Kind::EntityMap, &id)
100 24 : .await
101 24 : .unwrap_or(false)
102 24 : {
103 24 : n = n.saturating_add(1);
104 24 : }
105 : }
106 145058 : n
107 145070 : }
108 :
109 : /// The map a consumption request named, or nothing.
110 : ///
111 : /// 5.5.14: "If an EntityMap has expired, or cannot be accessed, no inference
112 : /// can be made as to which entities are held within the Context Sources and a
113 : /// new one shall be created." A store that refuses the read is one way a map
114 : /// cannot be accessed, so these paths recover the way they recover from an
115 : /// expiry — with a new map — instead of failing the request.
116 : /// A map built for a DIFFERENT subject is one this one cannot access, so the
117 : /// clause's own recovery applies: a new map is created for this request. It
118 : /// has to be that and not a refusal — the map id came from a header the
119 : /// client may well be replaying honestly, and an error would tell it that
120 : /// someone else's transaction exists (ADR-0020).
121 76 : pub(crate) async fn map_if_accessible(
122 76 : st: &AppState,
123 76 : tenant: &TenantId,
124 76 : headers: &HeaderMap,
125 76 : id: &str,
126 76 : ) -> Option<Value> {
127 76 : map_get(st, tenant, id)
128 76 : .await
129 76 : .ok()
130 76 : .flatten()
131 76 : .filter(|doc| crate::policy::belongs_to(doc, &crate::policy::subject_of(tenant, headers)))
132 76 : }
133 :
134 : /// The Entities of a map a given request may be answered from.
135 : ///
136 : /// 5.5.9.3: "the set of Entities considered for the result is fixed with the
137 : /// initial query creating the Entity map." The map is the CANDIDATE set; the
138 : /// request's own filters still apply on top of it, so a request that names
139 : /// `id=` is asking for the intersection and never for the map's whole set.
140 : /// Returned in the map's own order, which is the pagination order.
141 52 : pub(crate) fn candidate_ids(map: &Value, params: &HashMap<String, String>) -> Vec<String> {
142 52 : let named: Option<std::collections::HashSet<&str>> =
143 52 : params.get("id").map(|s| s.split(',').collect());
144 52 : map["entityMap"]
145 52 : .as_object()
146 52 : .map(|o| {
147 52 : o.keys()
148 228 : .filter(|k| named.as_ref().is_none_or(|n| n.contains(k.as_str())))
149 52 : .cloned()
150 52 : .collect()
151 52 : })
152 52 : .unwrap_or_default()
153 52 : }
154 :
155 : ///
156 : /// Every store failure here is the caller's: a map the broker could not count
157 : /// against its ceiling, or could not write, is not a map the client can be
158 : /// handed the id of (Table 6.3.2-1 InternalError).
159 2370 : pub(crate) async fn map_put(
160 2370 : st: &AppState,
161 2370 : tenant: &TenantId,
162 2370 : mut doc: Value,
163 2370 : ) -> Result<(), NgsiError> {
164 2370 : let Some(id) = doc.get("id").and_then(Value::as_str).map(str::to_owned) else {
165 0 : return Ok(());
166 : };
167 2370 : clamp_expiry(&mut doc);
168 : // The ceiling decides whether a NEW map fits; rewriting a map that is
169 : // already stored replaces a row that is already counted, so the paging
170 : // path neither lists the tenant's maps nor evicts one per page. A count
171 : // the store refuses leaves the ceiling unenforceable, and an unbounded
172 : // buffer is not the safer half of that choice: the write is refused.
173 2370 : if st.store.get(tenant, Kind::EntityMap, &id).await?.is_none() {
174 2366 : let existing = st.store.list(tenant, Kind::EntityMap).await?;
175 2364 : if existing.len() >= MAX_MAPS_PER_TENANT {
176 : // eviction order is a heuristic — earliest expiresAt string wins
177 32 : if let Some(victim) = existing
178 32 : .iter()
179 16352 : .min_by(|a, b| {
180 16352 : a["expiresAt"]
181 16352 : .as_str()
182 16352 : .unwrap_or("")
183 16352 : .cmp(b["expiresAt"].as_str().unwrap_or(""))
184 16352 : })
185 32 : .and_then(|d| d.get("id").and_then(Value::as_str))
186 : {
187 32 : st.store.delete(tenant, Kind::EntityMap, victim).await?;
188 0 : }
189 2332 : }
190 4 : }
191 2368 : let updated = st
192 2368 : .store
193 2368 : .mutate(tenant, Kind::EntityMap, &id, |d| {
194 4 : *d = doc.clone();
195 4 : Ok::<_, std::convert::Infallible>(())
196 4 : })
197 2368 : .await?
198 2368 : .is_some();
199 2368 : if !updated {
200 2364 : st.store.create(tenant, Kind::EntityMap, &id, doc).await?;
201 4 : }
202 2368 : Ok(())
203 2370 : }
204 :
205 : /// 6.4.3.2-1: "the actual expiresAt time of the EntityMap shall be set by
206 : /// the Context Broker or Context Source, possibly overriding the requested
207 : /// duration": every writer of a map's expiresAt goes through this. An
208 : /// absent or unreadable expiry is left alone: 5.5.14 keeps it unusable.
209 2394 : fn clamp_expiry(doc: &mut Value) {
210 2394 : let now = chrono::Utc::now();
211 2394 : let ceiling = now
212 2394 : .checked_add_signed(chrono::Duration::seconds(MAX_LIFETIME_SECS))
213 2394 : .unwrap_or(now);
214 2394 : if doc
215 2394 : .get("expiresAt")
216 2394 : .and_then(Value::as_str)
217 2394 : .and_then(dt)
218 2394 : .is_some_and(|e| e > ceiling)
219 : {
220 2092 : if let Some(o) = doc.as_object_mut() {
221 2092 : o.insert(
222 2092 : "expiresAt".into(),
223 2092 : json!(ceiling.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
224 2092 : );
225 2092 : }
226 302 : }
227 2394 : }
228 :
229 : /// 5.14.2.4: set the expiresAt of a map as it is stored now, clamped;
230 : /// `false` when the map is gone (a deleted map is not resurrected).
231 24 : pub(crate) async fn map_set_expiry(
232 24 : st: &AppState,
233 24 : tenant: &TenantId,
234 24 : id: &str,
235 24 : expires: &str,
236 24 : ) -> Result<bool, NgsiError> {
237 24 : let mut want = json!({"expiresAt": expires});
238 24 : clamp_expiry(&mut want);
239 24 : Ok(st
240 24 : .store
241 24 : .mutate(tenant, Kind::EntityMap, id, |d| {
242 20 : if let Some(o) = d.as_object_mut() {
243 20 : if let Some(e) = want.get("expiresAt") {
244 20 : o.insert("expiresAt".into(), e.clone());
245 20 : }
246 0 : }
247 20 : Ok::<_, std::convert::Infallible>(())
248 20 : })
249 24 : .await?
250 24 : .is_some())
251 24 : }
252 :
253 : /// 5.5.9.3 "Entities not or no longer fitting the query shall be removed
254 : /// from the Entity map during pagination": remove the `stale` ids a page
255 : /// visited, and nothing else, from the map as it is stored NOW. A request
256 : /// that read the map before an update or a delete writes back only its own
257 : /// removals, so a new expiresAt survives and a deleted map stays deleted.
258 58 : pub(crate) async fn map_prune(
259 58 : st: &AppState,
260 58 : tenant: &TenantId,
261 58 : id: &str,
262 58 : stale: &[String],
263 58 : ) -> Result<(), NgsiError> {
264 58 : if stale.is_empty() {
265 42 : return Ok(());
266 16 : }
267 16 : st.store
268 16 : .mutate(tenant, Kind::EntityMap, id, |d| {
269 12 : if let Some(emap) = d.get_mut("entityMap").and_then(Value::as_object_mut) {
270 12 : for k in stale {
271 12 : emap.remove(k);
272 12 : }
273 0 : }
274 12 : Ok::<_, std::convert::Infallible>(())
275 12 : })
276 16 : .await?;
277 16 : Ok(())
278 58 : }
279 :
280 : /// 5.14.3.4: "If the NGSI-LD endpoint does not know about a matching EntityMap
281 : /// for the EntityMap ID, then an error of type ResourceNotFound shall be
282 : /// raised." What the endpoint knows about is what [`map_get`] serves, so the
283 : /// delete reads through it: an expired map is beyond access (5.5.14) for the
284 : /// retrieve and for the delete alike; the expired row is left for
285 : /// `sweep_expired_maps` to collect.
286 116 : pub(crate) async fn map_delete(
287 116 : st: &AppState,
288 116 : tenant: &TenantId,
289 116 : id: &str,
290 116 : ) -> Result<bool, NgsiError> {
291 116 : if map_get(st, tenant, id).await?.is_none() {
292 4 : return Ok(false);
293 112 : }
294 112 : st.store.delete(tenant, Kind::EntityMap, id).await
295 116 : }
296 :
297 : /// Parse an ISO 8601 duration (entityMapLifetime, Table 6.4.3.2-1) to whole
298 : /// seconds; years/months are approximated (365/30 days), fractions rejected.
299 174 : pub(crate) fn iso8601_secs(s: &str) -> Option<i64> {
300 : // A lifetime is a span, so the calendar components are weighed at their
301 : // nominal length — a year 365 days, a month 30 — and a fractional or
302 : // absent component has no whole-second span to weigh.
303 174 : let d = antares_model::parse_iso_duration(s).filter(|d| d.whole && !d.empty)?;
304 68 : [
305 68 : (d.years, 31_536_000),
306 68 : (d.months, 2_592_000),
307 68 : (d.weeks, 604_800),
308 68 : (d.days, 86_400),
309 68 : (d.hours, 3_600),
310 68 : (d.minutes, 60),
311 68 : (d.seconds, 1),
312 68 : ]
313 68 : .into_iter()
314 452 : .try_fold(0i64, |acc, (n, per)| {
315 452 : acc.checked_add((n as i64).checked_mul(per)?)
316 452 : })
317 174 : }
318 :
319 : /// The expiresAt the broker assigns (5.2.39): now + suggested lifetime,
320 : /// bounded by the broker's ceiling; the default applies when none is given.
321 276 : pub(crate) fn expires_at(params: &HashMap<String, String>) -> Result<String, NgsiError> {
322 276 : let secs = match params.get("entityMapLifetime") {
323 22 : Some(d) => iso8601_secs(d)
324 22 : .ok_or_else(|| {
325 10 : NgsiError::BadRequestData(format!(
326 10 : "entityMapLifetime is not an ISO 8601 duration: {d:?}"
327 10 : ))
328 10 : })?
329 : // A zero or negative suggestion would answer 201 with a map that
330 : // 5.5.14 already forbids anyone from accessing, so the broker
331 : // floor applies as well as the ceiling.
332 12 : .clamp(1, MAX_LIFETIME_SECS),
333 254 : None => DEFAULT_LIFETIME_SECS,
334 : };
335 : // `secs` is clamped to MAX_LIFETIME_SECS above, far inside chrono's range.
336 266 : let now = chrono::Utc::now();
337 266 : Ok(now
338 266 : .checked_add_signed(chrono::Duration::seconds(secs))
339 266 : .unwrap_or(now)
340 266 : .to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
341 276 : }
342 :
343 : /// 5.14.1.4 / 5.14.2.4 / 5.14.3.4: every /entityMaps/{id} method opens the
344 : /// same way — the tenant it runs in, `local` (6.3.18) as the only parameter
345 : /// this resource takes, and an id that must be a valid URI before the store
346 : /// is touched at all.
347 284 : pub(crate) fn open_map(
348 284 : params: &HashMap<String, String>,
349 284 : headers: &HeaderMap,
350 284 : id: &str,
351 284 : ) -> ApiResult<TenantId> {
352 284 : let tenant = tenant_from(headers)?;
353 284 : check_params(params, &["local"])?;
354 284 : map_id_check(id)?;
355 272 : Ok(tenant)
356 284 : }
357 :
358 : /// 5.14.1.4 / 5.14.3.4: "If the EntityMap id is not present or it is not a
359 : /// valid URI, then an error of type BadRequestData shall be raised."
360 326 : pub(crate) fn map_id_check(id: &str) -> Result<(), NgsiError> {
361 326 : antares_model::EntityId::new(id)
362 326 : .map(|_| ())
363 326 : .map_err(|_| NgsiError::BadRequestData(format!("EntityMap id is not a valid URI: {id:?}")))
364 326 : }
365 :
366 : /// Table 5.2.39-2 on the way in: a returned EntityMap's `entityMap` is "a
367 : /// set of key-value pairs whose keys shall be strings representing Entity
368 : /// ids", so a key from a Context Source is checked before it becomes a key
369 : /// of the map this broker stores under its own id and serves from
370 : /// `/entityMaps/{id}`. Per key, not per peer: a source that names one
371 : /// unusable id still contributes its usable ones. `@none` fails the check
372 : /// with everything else, which is what it deserves here — it is the PEER's
373 : /// "held locally" marker (5.2.39) and stands for no Entity id on this side.
374 : /// `cap` is the ceiling the local half of the map already carries
375 : /// (`st.max_limit`), so no one Context Source is larger than the broker.
376 14 : fn peer_entity_ids(remote: &Value, cap: usize) -> Vec<&String> {
377 14 : remote
378 14 : .get("entityMap")
379 14 : .and_then(Value::as_object)
380 14 : .into_iter()
381 14 : .flat_map(serde_json::Map::keys)
382 2020 : .filter(|k| antares_model::EntityId::new(k).is_ok())
383 14 : .take(cap)
384 14 : .collect()
385 14 : }
386 :
387 : /// 5.14.4.4 and 5.14.5.4 end the same way: "The mapping between the Context
388 : /// Source Registration and the EntityMap Id is added to the linkedMaps
389 : /// element of the local EntityMap and for the Entity ids included in the
390 : /// returned Entity Maps a mapping to the Context Source Registration is added
391 : /// to the entityMap element of the local EntityMap. The local EntityMap is
392 : /// stored and made accessible based on its identifier." The two operations
393 : /// differ only in which registration operation they ask for and which peer
394 : /// resource carries it.
395 236 : pub(crate) async fn merge_and_store_map(
396 236 : st: &AppState,
397 236 : tenant: &TenantId,
398 236 : headers: &HeaderMap,
399 236 : ctx: &antares_jsonld::Context,
400 236 : params: &HashMap<String, String>,
401 236 : temporal: bool,
402 236 : mut emap: Map<String, Value>,
403 236 : ) -> ApiResult<Value> {
404 236 : let (op, path) = if temporal {
405 50 : ("createEntityMapQueryTemporal", "temporal/entityMaps")
406 : } else {
407 186 : ("createEntityMapQueryEntity", "entityMaps")
408 : };
409 236 : let mut linked = Map::new();
410 : // 5.5.13 local=true: no Context Source Registration is considered, so
411 : // nothing merges in and linkedMaps stays empty.
412 236 : if params.get("local").map(String::as_str) != Some("true") {
413 224 : let split = params.get("splitEntities").map(String::as_str) == Some("true");
414 14 : for (reg_id, remote) in
415 224 : crate::federation::fed_entity_maps(st, tenant, headers, ctx, params, split, op, path)
416 224 : .await?
417 : {
418 2012 : for eid in peer_entity_ids(&remote, st.max_limit) {
419 2012 : if let Some(a) = emap
420 2012 : .entry(eid.clone())
421 2012 : .or_insert_with(|| json!([]))
422 2012 : .as_array_mut()
423 2012 : {
424 2012 : a.push(json!(reg_id.clone()));
425 2012 : }
426 : }
427 : // Table 5.2.39-1 restricts an EntityMap id to a valid URI, and
428 : // 5.14.1.4 refuses one that is not from a client. The peer's id
429 : // travels back out as the `NGSILD-EntityMap` header of every
430 : // later forwarded page (`federation::map_gate`), so it is held
431 : // to the same rule; without a usable id the registration simply
432 : // carries no linked map and the peer re-runs its own query.
433 14 : match remote.get("id").and_then(Value::as_str) {
434 14 : Some(mid) if map_id_check(mid).is_ok() => {
435 12 : linked.insert(reg_id, json!(mid));
436 12 : }
437 2 : _ => {}
438 : }
439 : }
440 12 : }
441 236 : let mut doc = json!({
442 236 : "id": format!("urn:ngsi-ld:entitymap:{}", uuid::Uuid::new_v4()),
443 236 : "type": "EntityMap",
444 236 : "expiresAt": expires_at(params)?,
445 230 : "entityMap": Value::Object(emap),
446 230 : "linkedMaps": Value::Object(linked),
447 : });
448 230 : stamp_subject(&mut doc, tenant, headers);
449 230 : map_put(st, tenant, doc.clone()).await?;
450 228 : Ok(doc)
451 236 : }
452 :
453 : /// 5.7.1.4 / 5.7.3.4: the EntityMap created for a single-Entity retrieve —
454 : /// its one entry lists "@none" when Attribute data is held locally plus
455 : /// every matching Context Source Registration supporting the retrieve
456 : /// operation ("only the retrieved Entity Map shall be used to determine
457 : /// which Context Source Registrations match the Entity ID").
458 : #[allow(clippy::too_many_arguments)] // one param per 5.7.1.4 input
459 24 : pub(crate) async fn build_retrieve_map(
460 24 : st: &AppState,
461 24 : tenant: &TenantId,
462 24 : ctx: &antares_jsonld::Context,
463 24 : headers: &HeaderMap,
464 24 : id: &str,
465 24 : params: &HashMap<String, String>,
466 24 : temporal: bool,
467 24 : local_held: bool,
468 24 : ) -> Result<Value, NgsiError> {
469 24 : let mut srcs: Vec<Value> = Vec::new();
470 24 : if local_held {
471 24 : srcs.push(json!("@none"));
472 24 : }
473 24 : if crate::federation::active(params) {
474 22 : let spec = crate::registry::CsrSpec {
475 22 : ids: Some(vec![id.to_owned()]),
476 22 : ..Default::default()
477 22 : };
478 22 : for reg in crate::federation::matching_regs(st, tenant, &spec, ctx, headers).await? {
479 0 : let ok = if temporal {
480 0 : reg.supports("retrieveTemporal")
481 : } else {
482 0 : reg.read_op().is_some()
483 : };
484 0 : if ok {
485 0 : srcs.push(json!(reg.reg_id));
486 0 : }
487 : }
488 2 : }
489 24 : let mut emap = Map::new();
490 24 : if !srcs.is_empty() {
491 24 : emap.insert(id.to_owned(), Value::Array(srcs));
492 24 : }
493 24 : let mut doc = json!({
494 24 : "id": format!("urn:ngsi-ld:entitymap:{}", uuid::Uuid::new_v4()),
495 24 : "type": "EntityMap",
496 24 : "expiresAt": expires_at(params)?,
497 24 : "entityMap": Value::Object(emap),
498 24 : "linkedMaps": {},
499 : });
500 24 : stamp_subject(&mut doc, tenant, headers);
501 24 : map_put(st, tenant, doc.clone()).await?;
502 24 : Ok(doc)
503 24 : }
504 :
505 : /// The NGSILD-EntityMap response header: the resource URI of the map that
506 : /// determined the sources of this response (6.3.17).
507 30 : pub(crate) fn set_map_header(resp: &mut Response, mid: &str) {
508 30 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{mid}").parse() {
509 30 : resp.headers_mut().insert("NGSILD-EntityMap", v);
510 30 : }
511 30 : }
512 :
513 : /// 5.7.1.4 / 5.7.3.4 EntityMap usage on a single-Entity retrieve: a supplied
514 : /// NGSILD-EntityMap location is retrieved and, if live, is the only source
515 : /// used to determine which registrations match; an unknown or expired
516 : /// reference — or the `entityMap=true` flag — creates a new map, whose
517 : /// location is returned in the NGSILD-EntityMap response header. The two
518 : /// clauses word that rule identically and part company only over what
519 : /// "held locally" reads and which operation a registration must support,
520 : /// and both of those follow from `temporal` — so the rule is read once
521 : /// here and the retrieve itself is the caller's `inner`.
522 3264 : pub(crate) async fn retrieve_with_map<F, Fut>(
523 3264 : st: &AppState,
524 3264 : id: &str,
525 3264 : params: &HashMap<String, String>,
526 3264 : headers: &HeaderMap,
527 3264 : temporal: bool,
528 3264 : inner: F,
529 3264 : ) -> ApiResult<Response>
530 3264 : where
531 3264 : F: Fn(Option<Value>) -> Fut,
532 3264 : Fut: std::future::Future<Output = ApiResult<Response>>,
533 3264 : {
534 3264 : let tenant = tenant_from(headers)?;
535 3264 : let map_ref = single_header(headers, "NGSILD-EntityMap")?
536 3262 : .map(|r| r.rsplit('/').next().unwrap_or(&r).to_owned());
537 3262 : let existing = match map_ref.as_deref() {
538 12 : Some(mid) => map_if_accessible(st, &tenant, headers, mid).await,
539 3250 : None => None,
540 : };
541 3262 : if let Some(map) = existing {
542 6 : let mut resp = inner(Some(map)).await?;
543 6 : set_map_header(&mut resp, &map_ref.unwrap_or_default());
544 6 : return Ok(resp);
545 3256 : }
546 3256 : let want_map = map_ref.is_some() || params.get("entityMap").map(String::as_str) == Some("true");
547 3256 : let mut resp = inner(None).await?;
548 2940 : if want_map && resp.status().is_success() {
549 24 : let ctx = request_context(&st.loader, headers).await?;
550 24 : let local_held = if temporal {
551 6 : st.temporal
552 6 : .get_temporal(
553 6 : &tenant,
554 6 : id,
555 6 : &antares_store::filter::TemporalFilter::default(),
556 6 : )
557 6 : .await?
558 6 : .is_some()
559 : } else {
560 18 : st.store.get(&tenant, Kind::Entity, id).await?.is_some()
561 : };
562 24 : let map = build_retrieve_map(st, &tenant, &ctx, headers, id, params, temporal, local_held)
563 24 : .await?;
564 24 : if let Some(mid) = map.get("id").and_then(Value::as_str) {
565 24 : set_map_header(&mut resp, mid);
566 24 : }
567 2916 : }
568 2940 : Ok(resp)
569 3264 : }
570 :
571 : /// 201 + the EntityMap body + the NGSILD-EntityMap header carrying the
572 : /// resource URI of the created map (6.34.3.1 / 6.35.3.1).
573 170 : pub(crate) fn created_response(
574 170 : doc: Value,
575 170 : ctx: &antares_jsonld::Context,
576 170 : accept: Accept,
577 170 : tenant: &TenantId,
578 170 : ) -> Response {
579 170 : let uri = format!(
580 : "/ngsi-ld/v1/entityMaps/{}",
581 170 : doc.get("id").and_then(Value::as_str).unwrap_or_default()
582 : );
583 170 : let mut doc = doc;
584 170 : crate::policy::strip_internal(&mut doc);
585 170 : let mut resp = respond(StatusCode::CREATED, doc, ctx, accept, tenant);
586 170 : if let Ok(v) = uri.parse() {
587 170 : resp.headers_mut().insert("NGSILD-EntityMap", v);
588 170 : }
589 170 : resp
590 170 : }
591 :
592 : #[cfg(test)]
593 : mod tests {
594 : use super::*;
595 : use antares_store::Kind;
596 : use serde_json::json;
597 :
598 : /// `map_get` with its store failure unwrapped: these tests drive a
599 : /// working store, where a refusal would be the test's own bug.
600 56 : async fn map_read(st: &AppState, tenant: &TenantId, id: &str) -> Option<Value> {
601 56 : map_get(st, tenant, id).await.expect("the store answers")
602 56 : }
603 :
604 : /// 5.5.14 with RFC 9110 §9.2.1: reading a map the broker will not serve
605 : /// must not write. The row survives the read, and the sweep is what
606 : /// frees it.
607 : #[tokio::test]
608 4 : async fn an_expired_map_is_refused_by_a_read_that_writes_nothing() {
609 4 : let st = crate::wired_state("antares-map-sweep").await;
610 4 : let t = TenantId::default();
611 4 : let id = "urn:ngsi-ld:EntityMap:stale";
612 4 : st.store
613 4 : .create(
614 4 : &t,
615 4 : Kind::EntityMap,
616 4 : id,
617 4 : json!({"id": id, "expiresAt": "2000-01-01T00:00:00Z"}),
618 4 : )
619 4 : .await
620 4 : .expect("seed");
621 :
622 4 : assert!(map_read(&st, &t, id).await.is_none(), "5.5.14: not served");
623 4 : assert!(
624 4 : st.store
625 4 : .get(&t, Kind::EntityMap, id)
626 4 : .await
627 4 : .expect("store")
628 4 : .is_some(),
629 : "the read deleted the row: a GET must be safe (RFC 9110 9.2.1)"
630 : );
631 :
632 4 : assert_eq!(sweep_expired_maps(&st, &t).await, 1, "the sweep reaps it");
633 4 : assert!(
634 4 : st.store
635 4 : .get(&t, Kind::EntityMap, id)
636 4 : .await
637 4 : .expect("store")
638 4 : .is_none(),
639 4 : "the sweep left the row behind"
640 4 : );
641 4 : }
642 :
643 : /// The sweep removes exactly what a read refuses — never a map still in
644 : /// its lifetime.
645 : #[tokio::test]
646 4 : async fn the_sweep_keeps_every_map_a_read_would_still_serve() {
647 4 : let st = crate::wired_state("antares-map-sweep-live").await;
648 4 : let t = TenantId::default();
649 4 : let live = "urn:ngsi-ld:EntityMap:live";
650 4 : st.store
651 4 : .create(
652 4 : &t,
653 4 : Kind::EntityMap,
654 4 : live,
655 4 : json!({"id": live, "expiresAt": "2999-01-01T00:00:00Z"}),
656 4 : )
657 4 : .await
658 4 : .expect("seed");
659 4 : assert_eq!(
660 4 : sweep_expired_maps(&st, &t).await,
661 : 0,
662 : "a live map was reaped"
663 : );
664 4 : assert!(map_read(&st, &t, live).await.is_some(), "still served");
665 4 : }
666 :
667 : /// Table 6.4.3.2-1: entityMapLifetime is an ISO 8601 duration.
668 : #[test]
669 4 : fn clause_5_14_4_lifetime_parse() {
670 4 : assert_eq!(iso8601_secs("PT1H"), Some(3600));
671 4 : assert_eq!(iso8601_secs("PT90S"), Some(90));
672 4 : assert_eq!(iso8601_secs("P1DT2H3M4S"), Some(93784));
673 4 : assert_eq!(iso8601_secs("P2W"), Some(1_209_600));
674 : // invalid shapes are rejected (→ 400 at the handler)
675 28 : for bad in ["", "P", "PT", "1H", "PT1X", "PT1.5S", "PT1"] {
676 28 : assert_eq!(iso8601_secs(bad), None, "{bad:?}");
677 : }
678 4 : }
679 :
680 : /// Table 6.4.3.2-1: entityMapLifetime arrives on the query string, so
681 : /// the parser is attacker-facing — every hostile shape must return None
682 : /// (a 400) rather than panic, wrap or saturate.
683 : #[test]
684 4 : fn clause_5_14_4_lifetime_hostile_inputs() {
685 64 : for bad in [
686 4 : "P-1D", // negative component
687 4 : "-P1D", // negative duration
688 4 : "P+1D", // signed component
689 4 : "p1d", // lower case designators
690 4 : " PT1H", // leading whitespace
691 4 : "PT1H ", // trailing whitespace
692 4 : "P1DT", // empty time part
693 4 : "PT99999999999999999999S", // digit run past i64
694 4 : "P9999999999999Y", // multiplication overflow
695 4 : "P92233720368547758S", // addition overflow after scaling
696 4 : "PT1H1", // trailing digits, no designator
697 4 : "P١D", // non-ASCII digit
698 4 : "P1D\u{0}", // embedded NUL
699 4 : "PT,5S", // comma fraction
700 4 : "P1S", // time designator in the date part
701 4 : "PT1D", // date designator in the time part
702 4 : ] {
703 64 : assert_eq!(iso8601_secs(bad), None, "{bad:?} must not parse");
704 : }
705 : // the whole i64 range is walked without panicking
706 4 : assert_eq!(iso8601_secs(&format!("PT{}S", i64::MAX)), Some(i64::MAX));
707 4 : assert_eq!(iso8601_secs(&format!("PT{}S", u64::MAX)), None);
708 4 : assert_eq!(iso8601_secs("PT0S"), Some(0));
709 4 : }
710 :
711 16 : fn params(pairs: &[(&str, &str)]) -> HashMap<String, String> {
712 16 : pairs
713 16 : .iter()
714 16 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
715 16 : .collect()
716 16 : }
717 :
718 : /// 6.4.3.2-1: "the actual expiresAt time of the EntityMap shall be set by
719 : /// the Context Broker or Context Source, possibly overriding the
720 : /// requested duration" — the client suggestion is bounded above by the
721 : /// broker ceiling and below by a lifetime the map can actually be used
722 : /// for; an unparseable duration is BadRequestData.
723 : #[test]
724 4 : fn clause_5_14_4_expires_at_is_broker_bounded() {
725 4 : let now = chrono::Utc::now();
726 12 : let at = |p: &[(&str, &str)]| {
727 12 : let s = expires_at(¶ms(p)).expect("expiry");
728 12 : dt(&s).expect("RFC 3339 expiry")
729 12 : };
730 4 : let default = at(&[]);
731 4 : assert!(
732 4 : (default - now).num_seconds() >= DEFAULT_LIFETIME_SECS - 5
733 4 : && (default - now).num_seconds() <= DEFAULT_LIFETIME_SECS + 5,
734 : "no suggestion → the default lifetime"
735 : );
736 4 : let capped = at(&[("entityMapLifetime", "P30D")]);
737 4 : assert!(
738 4 : (capped - now).num_seconds() <= MAX_LIFETIME_SECS,
739 : "a client cannot exceed the broker ceiling"
740 : );
741 4 : let zero = at(&[("entityMapLifetime", "PT0S")]);
742 4 : assert!(
743 4 : zero > now,
744 : "a zero lifetime would return 201 for a map that is already \
745 : unusable (5.5.14): {zero}"
746 : );
747 4 : match expires_at(¶ms(&[("entityMapLifetime", "yesterday")])) {
748 4 : Err(NgsiError::BadRequestData(_)) => {}
749 0 : other => panic!("an invalid duration must be BadRequestData: {other:?}"),
750 : }
751 4 : }
752 :
753 : /// 5.14: EntityMaps are per-tenant resources — an EntityMap created
754 : /// under one tenant is invisible and undeletable from another (4.14
755 : /// multi-tenancy: "an NGSI-LD system shall behave as if the tenants were
756 : /// separate systems").
757 : #[tokio::test]
758 4 : async fn clause_5_14_maps_are_tenant_scoped() {
759 4 : let st = AppState::new("antares-em-unit".into());
760 4 : let a = TenantId::new("alpha").expect("tenant");
761 4 : let b = TenantId::new("beta").expect("tenant");
762 4 : let id = "urn:ngsi-ld:entitymap:t1";
763 4 : map_put(&st, &a, live_map(id)).await.expect("stored");
764 4 : assert!(map_read(&st, &a, id).await.is_some());
765 4 : assert!(
766 4 : map_read(&st, &b, id).await.is_none(),
767 : "another tenant must not read the map"
768 : );
769 4 : assert!(
770 4 : !map_delete(&st, &b, id).await.expect("delete"),
771 : "another tenant must not delete the map"
772 : );
773 4 : assert!(
774 4 : map_read(&st, &a, id).await.is_some(),
775 : "the owner still has its map"
776 : );
777 4 : assert!(map_delete(&st, &a, id).await.expect("delete"));
778 4 : assert!(map_read(&st, &a, id).await.is_none());
779 4 : }
780 :
781 : /// 5.5.9.3 + 5.14.2.4: a page's prune lands after an
782 : /// update and a delete that ran while the page was read. The update's
783 : /// expiresAt survives, only the visited stale id goes, and a map deleted
784 : /// mid-walk stays deleted.
785 : #[tokio::test]
786 4 : async fn clause_5_5_9_3_prune_keeps_a_concurrent_update_and_delete() {
787 4 : let st = AppState::new("antares-em-prune".into());
788 4 : let t = TenantId::default();
789 4 : let id = "urn:ngsi-ld:entitymap:p1";
790 4 : let mut doc = live_map(id);
791 4 : doc["entityMap"] = json!({"urn:a": ["@none"], "urn:b": ["@none"]});
792 4 : map_put(&st, &t, doc).await.expect("stored");
793 4 : let read_by_page = map_read(&st, &t, id).await.expect("map");
794 4 : let later = (chrono::Utc::now() + chrono::Duration::seconds(3000))
795 4 : .to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
796 4 : assert!(map_set_expiry(&st, &t, id, &later).await.expect("update"));
797 4 : assert_ne!(read_by_page["expiresAt"], json!(later));
798 4 : map_prune(&st, &t, id, &["urn:a".to_owned()])
799 4 : .await
800 4 : .expect("prune");
801 4 : let now = map_read(&st, &t, id).await.expect("map");
802 4 : assert_eq!(now["expiresAt"], json!(later), "the prune undid the update");
803 4 : assert_eq!(now["entityMap"], json!({"urn:b": ["@none"]}));
804 :
805 4 : assert!(map_delete(&st, &t, id).await.expect("delete"));
806 4 : map_prune(&st, &t, id, &["urn:b".to_owned()])
807 4 : .await
808 4 : .expect("prune");
809 4 : assert!(
810 4 : st.store
811 4 : .get(&t, Kind::EntityMap, id)
812 4 : .await
813 4 : .expect("store")
814 4 : .is_none(),
815 : "the prune resurrected a deleted map"
816 : );
817 4 : assert!(
818 4 : !map_set_expiry(&st, &t, id, &later).await.expect("update"),
819 : "an update resurrected a deleted map"
820 : );
821 4 : assert!(st
822 4 : .store
823 4 : .get(&t, Kind::EntityMap, id)
824 4 : .await
825 4 : .expect("store")
826 4 : .is_none());
827 4 : }
828 :
829 2116 : fn live_map(id: &str) -> Value {
830 2116 : json!({
831 2116 : "id": id,
832 2116 : "type": "EntityMap",
833 2116 : "expiresAt": (chrono::Utc::now() + chrono::Duration::seconds(600))
834 2116 : .to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
835 2116 : "entityMap": {},
836 2116 : "linkedMaps": {},
837 : })
838 2116 : }
839 :
840 : /// 5.5.14: an expired EntityMap "cannot be accessed" — it is never
841 : /// served, and a map whose expiry cannot be read is treated the same way
842 : /// rather than living forever.
843 : #[tokio::test]
844 4 : async fn clause_5_5_14_expired_maps_are_never_served() {
845 4 : let st = AppState::new("antares-em-exp".into());
846 4 : let t = TenantId::default();
847 4 : let mut past = live_map("urn:ngsi-ld:entitymap:past");
848 4 : past["expiresAt"] = json!("2020-01-01T00:00:00.000Z");
849 4 : map_put(&st, &t, past).await.expect("stored");
850 4 : assert!(map_read(&st, &t, "urn:ngsi-ld:entitymap:past")
851 4 : .await
852 4 : .is_none());
853 4 : assert!(
854 4 : st.store
855 4 : .get(&t, Kind::EntityMap, "urn:ngsi-ld:entitymap:past")
856 4 : .await
857 4 : .expect("store")
858 4 : .is_some(),
859 : "the read pruned the row: reading must not write"
860 : );
861 12 : for (id, expiry) in [
862 4 : ("urn:ngsi-ld:entitymap:none", None),
863 4 : ("urn:ngsi-ld:entitymap:junk", Some(json!("whenever"))),
864 4 : ("urn:ngsi-ld:entitymap:num", Some(json!(0))),
865 4 : ] {
866 12 : let mut doc = live_map(id);
867 12 : match expiry {
868 8 : Some(v) => doc["expiresAt"] = v,
869 4 : None => {
870 4 : doc.as_object_mut().expect("object").remove("expiresAt");
871 4 : }
872 4 : }
873 12 : map_put(&st, &t, doc).await.expect("stored");
874 12 : assert!(
875 12 : map_read(&st, &t, id).await.is_none(),
876 4 : "{id} has no readable expiry and must not be served"
877 4 : );
878 4 : }
879 4 : // What no read will serve, the sweep frees: the four seeded above.
880 4 : assert_eq!(sweep_expired_maps(&st, &t).await, 4);
881 4 : }
882 :
883 : /// 5.14.1.1 storage: every buffer is bounded — the per-tenant EntityMap
884 : /// registry has a ceiling, and filling it evicts rather than growing.
885 : #[tokio::test]
886 4 : async fn clause_5_14_1_map_registry_is_bounded() {
887 4 : let st = AppState::new("antares-em-cap".into());
888 4 : let t = TenantId::default();
889 2080 : for i in 0..MAX_MAPS_PER_TENANT + 8 {
890 2080 : let mut doc = live_map(&format!("urn:ngsi-ld:entitymap:{i:04}"));
891 : // earliest expiry first, so the eviction victim is deterministic
892 2080 : doc["expiresAt"] = json!(format!("2099-01-01T00:00:{:02}.000Z", i % 60));
893 2080 : map_put(&st, &t, doc).await.expect("stored");
894 2080 : assert!(
895 2080 : st.store
896 2080 : .list(&t, Kind::EntityMap)
897 2080 : .await
898 2080 : .expect("list")
899 2080 : .len()
900 : <= MAX_MAPS_PER_TENANT,
901 : "the registry exceeded its ceiling at {i}"
902 : );
903 : }
904 : // re-storing a known id is an update, never an eviction
905 4 : let before = st
906 4 : .store
907 4 : .list(&t, Kind::EntityMap)
908 4 : .await
909 4 : .expect("list")
910 4 : .len();
911 4 : let known = st.store.list(&t, Kind::EntityMap).await.expect("list")[0]["id"]
912 4 : .as_str()
913 4 : .expect("id")
914 4 : .to_owned();
915 4 : map_put(&st, &t, live_map(&known)).await.expect("stored");
916 4 : assert_eq!(
917 4 : st.store
918 4 : .list(&t, Kind::EntityMap)
919 4 : .await
920 4 : .expect("list")
921 4 : .len(),
922 4 : before
923 4 : );
924 4 : }
925 :
926 : /// 5.5.14 + Table 6.4.3.2-1: "the actual expiresAt time of the EntityMap
927 : /// shall be set by the Context Broker or Context Source, possibly
928 : /// overriding the requested duration" — the 5.14.2.4 update path writes a
929 : /// client-chosen instant, so the broker ceiling binds when the map is
930 : /// stored, not only when it is created.
931 : #[tokio::test]
932 4 : async fn clause_5_5_14_stored_expiry_never_exceeds_the_broker_ceiling() {
933 4 : let st = AppState::new("antares-em-clamp".into());
934 4 : let t = TenantId::default();
935 4 : let ceiling = chrono::Utc::now() + chrono::Duration::seconds(MAX_LIFETIME_SECS);
936 4 : let far = "urn:ngsi-ld:entitymap:far";
937 4 : let mut doc = live_map(far);
938 4 : doc["expiresAt"] = json!("2099-01-01T00:00:00.000Z");
939 4 : map_put(&st, &t, doc).await.expect("stored");
940 4 : let stored = map_read(&st, &t, far)
941 4 : .await
942 4 : .expect("a clamped map is still live");
943 4 : let at = dt(stored["expiresAt"].as_str().expect("expiresAt")).expect("RFC 3339 expiry");
944 4 : assert!(
945 4 : at <= ceiling + chrono::Duration::seconds(5),
946 : "a client cannot pin an EntityMap past the broker ceiling: {at}"
947 : );
948 4 : assert!(at > chrono::Utc::now(), "the map stays usable: {at}");
949 : // an expiry inside the ceiling is stored verbatim, not rewritten
950 4 : let near = "urn:ngsi-ld:entitymap:near";
951 4 : let doc = live_map(near);
952 4 : let want = doc["expiresAt"].clone();
953 4 : map_put(&st, &t, doc).await.expect("stored");
954 4 : assert_eq!(
955 4 : map_read(&st, &t, near).await.expect("live")["expiresAt"],
956 4 : want
957 4 : );
958 4 : }
959 : }
|
