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