Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! /csourceRegistrations (5.9, 5.10; resources 6.8/6.9).
3 :
4 : use crate::negotiate::*;
5 : use crate::registry::{
6 : csf_matches, csr_matches, ei_types, present_registration, reg_expired,
7 : temporal_interval_matches, CsrSpec,
8 : };
9 : use crate::state::{now_iso, AppState};
10 : use antares_jsonld::{parse_datetime, Context};
11 : use antares_model::operations::{OPERATION_GROUPS, OPERATION_NAMES};
12 : use antares_model::{NgsiError, TenantId};
13 : use antares_store::CurrentStateDriverExt;
14 : use antares_store::Kind;
15 : use axum::body::Bytes;
16 : use axum::extract::{Path, State};
17 : use axum::http::{HeaderMap, StatusCode};
18 : use axum::response::{IntoResponse, Response};
19 : use serde_json::{Map, Value};
20 :
21 : use crate::negotiate::CleanParams;
22 :
23 : /// Cardinality caps on a CSourceRegistration. Generous against any real
24 : /// federation topology (a tenant is sized at 1000+ registrations, not one
25 : /// registration at 1000+ selectors) and small enough that the worst case is
26 : /// MAX_INFORMATION × MAX_INFO_MEMBERS² index rows, not 10^10.
27 : const MAX_INFORMATION: usize = 128;
28 : const MAX_INFO_MEMBERS: usize = 128;
29 : /// Table 5.2.10-1 RegistrationInfo: the `information` member of a
30 : /// Context Source Registration (5.2.9). Each entry names what the
31 : /// source holds — entities, property names, relationship names — and
32 : /// every name is expanded, because matching (5.11.2) compares IRIs.
33 4416 : fn norm_registration_info(v: &Value, ctx: &Context) -> Result<Value, NgsiError> {
34 4416 : let bad = NgsiError::BadRequestData;
35 4416 : let arr = v
36 4416 : .as_array()
37 4416 : .filter(|a| !a.is_empty())
38 4416 : .ok_or_else(|| bad("information must be a non-empty array (5.2.9)".into()))?;
39 : // The csource_index explosion is |entities| ×
40 : // (|propertyNames| + |relationshipNames|) PER information
41 : // element, materialised in memory before any SQL runs. Under
42 : // only the 4 MiB body cap that is ~10^10 objects — an OOM from
43 : // one request. Cardinality is capped at the validation
44 : // boundary, where the error is a 400 and not a dead pod:
45 : // there is no query here to be too complex, and 5.9.2.4 gives
46 : // BadRequestData for a registration whose content is refused.
47 4412 : if arr.len() > MAX_INFORMATION {
48 4 : return Err(bad(format!(
49 4 : "information has {} entries (limit {MAX_INFORMATION})",
50 4 : arr.len()
51 4 : )));
52 4408 : }
53 4408 : let mut infos = Vec::new();
54 5054 : for info in arr {
55 5054 : let io = info
56 5054 : .as_object()
57 5054 : .ok_or_else(|| bad("information entries must be objects".into()))?;
58 20188 : for key in [
59 5054 : "entities",
60 5054 : "attributeNames",
61 5054 : "propertyNames",
62 5054 : "relationshipNames",
63 5054 : ] {
64 20188 : if let Some(n) = io.get(key).and_then(Value::as_array).map(Vec::len) {
65 7814 : if n > MAX_INFO_MEMBERS {
66 16 : return Err(bad(format!(
67 16 : "information.{key} has {n} entries (limit {MAX_INFO_MEMBERS})"
68 16 : )));
69 7798 : }
70 12374 : }
71 : }
72 5038 : let mut ni = Map::new();
73 7786 : for (ik, iv) in io {
74 7786 : match ik.as_str() {
75 7786 : "entities" => {
76 5002 : let es = iv
77 5002 : .as_array()
78 5002 : .filter(|a| !a.is_empty())
79 5002 : .ok_or_else(|| bad("entities must be a non-empty array".into()))?;
80 4998 : let mut nes = Vec::new();
81 5674 : for e in es {
82 5674 : let eo = e
83 5674 : .as_object()
84 5674 : .ok_or_else(|| bad("entities entries must be objects".into()))?;
85 5674 : let mut ne = Map::new();
86 9504 : for (ek, ev) in eo {
87 9504 : match ek.as_str() {
88 : // 5.2.8: type is "String or String[]" — both forms legal.
89 9504 : "type" => {
90 5650 : let expand_one = |t: &Value| -> Result<Value, NgsiError> {
91 5650 : let t = t.as_str().filter(|t| !t.is_empty()).ok_or_else(|| {
92 0 : bad("EntityInfo type must be a non-empty string (5.2.8)".into())
93 0 : })?;
94 5650 : Ok(Value::String(ctx.expand_key(t)))
95 5650 : };
96 5642 : let expanded = match ev {
97 20 : Value::Array(ts) if !ts.is_empty() => Value::Array(
98 16 : ts.iter().map(expand_one).collect::<Result<_, _>>()?,
99 : ),
100 : Value::Array(_) => {
101 4 : return Err(bad(
102 4 : "EntityInfo type array must not be empty (5.2.8)"
103 4 : .into(),
104 4 : ))
105 : }
106 5626 : other => expand_one(other)?,
107 : };
108 5642 : ne.insert("type".into(), expanded);
109 : }
110 3858 : "id" => {
111 2568 : let id = ev
112 2568 : .as_str()
113 2568 : .ok_or_else(|| bad("EntityInfo id must be a URI".into()))?;
114 2568 : antares_model::EntityId::new(id)?;
115 2564 : ne.insert("id".into(), ev.clone());
116 : }
117 1290 : "idPattern" => {
118 1290 : let p = ev
119 1290 : .as_str()
120 1290 : .ok_or_else(|| bad("idPattern must be a string".into()))?;
121 1290 : antares_ql::regex::compile(p)
122 1290 : .map_err(|_| bad(format!("invalid idPattern {p:?}")))?;
123 1286 : ne.insert("idPattern".into(), ev.clone());
124 : }
125 0 : _ => {
126 0 : ne.insert(ek.clone(), ev.clone());
127 0 : }
128 : }
129 : }
130 : // type is optional in EntityInfo when an
131 : // id/idPattern identifies the entities
132 5662 : if !ne.contains_key("type")
133 20 : && !ne.contains_key("id")
134 20 : && !ne.contains_key("idPattern")
135 : {
136 0 : return Err(bad(
137 0 : "EntityInfo requires type, id or idPattern (5.2.8)".into()
138 0 : ));
139 5662 : }
140 5662 : nes.push(Value::Object(ne));
141 : }
142 4986 : ni.insert("entities".into(), Value::Array(nes));
143 : }
144 2784 : "attributeNames" | "propertyNames" | "relationshipNames" => {
145 : // 5.2.10, TS 104 175 Table 5.2.6.6.5-1: "Empty array is
146 : // not allowed"
147 2784 : let names = iv
148 2784 : .as_array()
149 2784 : .filter(|a| !a.is_empty())
150 2784 : .ok_or_else(|| bad(format!("{ik} must be a non-empty array (5.2.10)")))?;
151 2772 : let mut nn = Vec::new();
152 4150 : for n in names {
153 4150 : let s = n
154 4150 : .as_str()
155 4150 : .ok_or_else(|| bad(format!("{ik} entries must be strings")))?;
156 4148 : nn.push(Value::String(ctx.expand_key(s)));
157 : }
158 2770 : ni.insert(ik.clone(), Value::Array(nn));
159 : }
160 0 : _ => {
161 0 : ni.insert(ik.clone(), iv.clone());
162 0 : }
163 : }
164 : }
165 5008 : infos.push(Value::Object(ni));
166 : }
167 4362 : Ok(Value::Array(infos))
168 4416 : }
169 :
170 : /// 4.3.6.5 and Table 5.2.22-1: the `contextSourceInfo` pairs a broker
171 : /// sends when it contacts the source. Every pair becomes an HTTP header
172 : /// on a forward, and the four keys 4.3.6.6 processes have value spaces
173 : /// of their own.
174 246 : fn check_context_source_info(v: &Value) -> Result<(), NgsiError> {
175 246 : let bad = NgsiError::BadRequestData;
176 246 : let arr = v
177 246 : .as_array()
178 246 : .ok_or_else(|| bad("contextSourceInfo must be an array (5.2.9)".into()))?;
179 258 : for kv in arr {
180 258 : let Some(key) = kv.get("key").and_then(Value::as_str) else {
181 8 : return Err(bad(
182 8 : "contextSourceInfo entries must be {key, value} pairs (5.2.22)".into(),
183 8 : ));
184 : };
185 : // Table 5.2.22-1: value is a String, cardinality 1.
186 250 : let Some(value) = kv.get("value").filter(|v| v.is_string()) else {
187 28 : return Err(bad(
188 28 : "contextSourceInfo entries must be {key, value} pairs of Strings (5.2.22)".into(),
189 28 : ));
190 : };
191 : // 6.3.19: "Key and value members shall adhere to IETF
192 : // RFC 7230 definitions concerning HTTP headers". The pair
193 : // becomes a header on every forward, so the transport's
194 : // own RFC 7230 parsers are the judge — a name or a value
195 : // they refuse can only fail later, at a forward whose
196 : // error names no registration.
197 222 : if !crate::negotiate::is_field_name(key) {
198 40 : return Err(bad(format!(
199 40 : "contextSourceInfo key {key:?} is not an RFC 7230 header name (6.3.19)"
200 40 : )));
201 182 : }
202 182 : if !value.as_str().is_some_and(crate::negotiate::is_field_value) {
203 24 : return Err(bad(format!(
204 24 : "contextSourceInfo value for {key:?} is not an RFC 7230 header value \
205 24 : (6.3.19)"
206 24 : )));
207 158 : }
208 : // 4.3.6.6: the four processed keys have constrained
209 : // value spaces — reject bad ones at registration, not at
210 : // first forward
211 158 : let sval = value.as_str();
212 158 : match key.to_ascii_lowercase().as_str() {
213 : // TS 104 175 9.5.1: "As Tenant information, if applicable, is
214 : // directly specified in the CSourceRegistration, it shall not
215 : // be part of contextSourceInfo." A forward never sends it
216 : // (`federation::CSI_SKIP`); refusing it here tells the client
217 : // its routing would not happen, instead of forwarding silently
218 : // to the default tenant.
219 158 : "ngsild-tenant" => {
220 18 : return Err(bad(
221 18 : "contextSourceInfo must not carry NGSILD-Tenant: set the \
222 18 : registration's tenant member instead (9.5.1)"
223 18 : .into(),
224 18 : ));
225 : }
226 140 : "accept" | "contenttype" => {
227 24 : if !matches!(sval, Some("application/json" | "application/ld+json")) {
228 8 : return Err(bad(format!(
229 8 : "contextSourceInfo {key} must be application/json or \
230 8 : application/ld+json (4.3.6.6)"
231 8 : )));
232 16 : }
233 : }
234 116 : "jsonldcontext" => {
235 32 : if sval.is_none_or(|s| antares_model::EntityId::new(s).is_err()) {
236 4 : return Err(bad(
237 4 : "contextSourceInfo jsonldContext must be a URL (4.3.6.6)".into(),
238 4 : ));
239 28 : }
240 : }
241 84 : "ngsildconformance"
242 8 : if sval.is_none_or(|s| crate::conformance::parse_version(s).is_none()) =>
243 : {
244 4 : return Err(bad(
245 4 : "contextSourceInfo ngsildConformance must be major.minor \
246 4 : (4.3.6.6)"
247 4 : .into(),
248 4 : ));
249 : }
250 80 : _ => {}
251 : }
252 : }
253 112 : Ok(())
254 246 : }
255 :
256 : /// Table 5.2.9-1: `operations` entries "are limited to the named API
257 : /// operations and named operation groups (see clause 4.20)".
258 1418 : fn check_operations(v: &Value) -> Result<(), NgsiError> {
259 1418 : let bad = NgsiError::BadRequestData;
260 1418 : let arr = v
261 1418 : .as_array()
262 1418 : .filter(|a| !a.is_empty())
263 1418 : .ok_or_else(|| bad("operations must be a non-empty array (5.2.9)".into()))?;
264 2372 : for op in arr {
265 2372 : let name = op.as_str().unwrap_or_default();
266 2372 : if !OPERATION_NAMES.contains(&name) && !OPERATION_GROUPS.contains(&name) {
267 28 : return Err(bad(format!(
268 28 : "unknown operation {name:?} — entries are limited to the \
269 28 : 4.20 names and groups (5.2.9)"
270 28 : )));
271 2344 : }
272 : }
273 1382 : Ok(())
274 1418 : }
275 :
276 : /// Table 5.2.34-1 RegistrationManagementInfo: `cacheDuration` an ISO 8601
277 : /// duration, `cooldown` and `timeout` numbers greater than 0, `localOnly`
278 : /// a Boolean.
279 66 : fn check_registration_management(v: &Value) -> Result<(), NgsiError> {
280 66 : let bad = NgsiError::BadRequestData;
281 66 : let m = v.as_object().ok_or_else(|| {
282 8 : bad("management must be a RegistrationManagementInfo object (5.2.34)".into())
283 8 : })?;
284 58 : if let Some(d) = m.get("cacheDuration") {
285 20 : if !d.as_str().is_some_and(valid_iso8601_duration) {
286 12 : return Err(bad(
287 12 : "management cacheDuration must be an ISO 8601 duration (5.2.34)".into(),
288 12 : ));
289 8 : }
290 38 : }
291 84 : for key in ["cooldown", "timeout"] {
292 84 : if let Some(n) = m.get(key) {
293 46 : if !n.as_f64().is_some_and(|n| n > 0.0) {
294 20 : return Err(bad(format!(
295 20 : "management {key} must be a number greater than 0 (5.2.34)"
296 20 : )));
297 26 : }
298 38 : }
299 : }
300 26 : if let Some(l) = m.get("localOnly") {
301 16 : if !l.is_boolean() {
302 8 : return Err(bad("management localOnly must be a boolean (5.2.34)".into()));
303 8 : }
304 10 : }
305 18 : Ok(())
306 66 : }
307 :
308 : /// One member of Table 5.2.9-1, validated and normalized into `out`.
309 : /// A member the table does not name is kept verbatim: 5.5.9 asks a
310 : /// receiver to tolerate what it does not know.
311 23734 : fn norm_reg_member(
312 23734 : k: &str,
313 23734 : v: &Value,
314 23734 : ctx: &Context,
315 23734 : out: &mut Map<String, Value>,
316 23734 : ) -> Result<(), NgsiError> {
317 23734 : let bad = NgsiError::BadRequestData;
318 21802 : match k {
319 : // Table 5.2.9-2's members are read-only and "shall be automatically
320 : // generated by NGSI-LD implementations. In the event that they are
321 : // provided (in update or create operations) NGSI-LD implementations
322 : // shall ignore them" — dropped here rather than refused, which is
323 : // what "ignore" says, alongside the 4.8 system attributes.
324 23734 : "@context" | "createdAt" | "modifiedAt" | "status" | "timesSent" | "timesFailed"
325 21834 : | "lastSuccess" | "lastFailure" => return Ok(()),
326 : // The `__` prefix is the broker's (`policy::SUBJECT_MEMBER`): this
327 : // only sees a create body or a 5.9.3 fragment, so dropping it here
328 : // stops a client from seeding the subject a Registration was
329 : // created by and from replacing it later.
330 21802 : k if k.starts_with("__") => return Ok(()),
331 21798 : "id" => {
332 4346 : let id = v
333 4346 : .as_str()
334 4346 : .ok_or_else(|| bad("registration id must be a string URI".into()))?;
335 4346 : antares_model::EntityId::new(id)?;
336 4346 : out.insert("id".into(), v.clone());
337 : }
338 17452 : "type" => {
339 4242 : if v.as_str() != Some("ContextSourceRegistration") {
340 4 : return Err(bad(
341 4 : "type must be \"ContextSourceRegistration\" (5.2.9)".into()
342 4 : ));
343 4238 : }
344 4238 : out.insert("type".into(), v.clone());
345 : }
346 13210 : "information" => {
347 4416 : out.insert("information".into(), norm_registration_info(v, ctx)?);
348 : }
349 8794 : "mode" => {
350 2228 : let m = v
351 2228 : .as_str()
352 2228 : .filter(|m| ["inclusive", "auxiliary", "exclusive", "redirect"].contains(m))
353 2228 : .ok_or_else(|| {
354 4 : bad("mode must be inclusive, auxiliary, exclusive or redirect (5.2.9)".into())
355 4 : })?;
356 2224 : out.insert("mode".into(), Value::String(m.to_owned()));
357 : }
358 6566 : "endpoint" => {
359 4470 : let uri = v
360 4470 : .as_str()
361 4470 : .ok_or_else(|| bad("endpoint must be a URI string".into()))?;
362 4470 : antares_model::EntityId::new(uri)
363 4470 : .map_err(|_| bad(format!("endpoint is not a valid URI: {uri:?}")))?;
364 4470 : out.insert("endpoint".into(), v.clone());
365 : }
366 2096 : "expiresAt" => {
367 70 : let s = v
368 70 : .as_str()
369 70 : .filter(|s| parse_datetime(s))
370 70 : .ok_or_else(|| bad("expiresAt must be an ISO 8601 DateTime".into()))?;
371 : // the instant decides, not the spelling: now_iso always
372 : // carries 3 fraction digits, a client's expiresAt 0 to 6
373 62 : if antares_model::dt_key(s) < antares_model::dt_key(&now_iso()) {
374 36 : return Err(bad("expiresAt is in the past".into()));
375 26 : }
376 26 : out.insert("expiresAt".into(), v.clone());
377 : }
378 : // 5.2.9 `tenant`: the Tenant to use in all requests to this
379 : // source — validated with the same rules as the header (4.14).
380 2026 : "tenant" => {
381 12 : let t = v
382 12 : .as_str()
383 12 : .ok_or_else(|| bad("tenant must be a string (5.2.9)".into()))?;
384 12 : antares_model::TenantId::new(t)?;
385 12 : out.insert("tenant".into(), v.clone());
386 : }
387 : // 4.3.6.5: KeyValuePair[] conveyed when contacting the source.
388 2014 : "contextSourceInfo" => {
389 246 : check_context_source_info(v)?;
390 112 : out.insert("contextSourceInfo".into(), v.clone());
391 : }
392 : // Table 5.2.9-1: operations entries "are limited to the named
393 : // API operations and named operation groups (see clause 4.20)".
394 1768 : "operations" => {
395 1418 : check_operations(v)?;
396 1382 : out.insert("operations".into(), v.clone());
397 : }
398 : // 4.3.6.4 / 5.2.9: localOnly is a Boolean.
399 350 : "localOnly" => {
400 16 : if !v.is_boolean() {
401 4 : return Err(bad("localOnly must be a boolean (5.2.9)".into()));
402 12 : }
403 12 : out.insert("localOnly".into(), v.clone());
404 : }
405 : // Table 5.2.9-1: a non-empty RFC 7230 pseudonym token.
406 334 : "contextSourceAlias" => {
407 18 : let a = v
408 18 : .as_str()
409 18 : .filter(|a| crate::negotiate::is_field_name(a))
410 18 : .ok_or_else(|| {
411 8 : bad(
412 8 : "contextSourceAlias must be a non-empty RFC 7230 pseudonym token \
413 8 : (5.2.9)"
414 8 : .into(),
415 8 : )
416 8 : })?;
417 10 : out.insert("contextSourceAlias".into(), Value::String(a.to_owned()));
418 : }
419 : // Table 5.2.9-1: non-empty strings.
420 316 : "description" | "registrationName" => {
421 20 : if v.as_str().is_none_or(str::is_empty) {
422 8 : return Err(bad(format!("{k} must be a non-empty string (5.2.9)")));
423 12 : }
424 12 : out.insert(k.to_owned(), v.clone());
425 : }
426 : // Table 5.2.9-1: valid URIs, "@none" for the default instances.
427 296 : "datasetId" => {
428 10 : let arr = v
429 10 : .as_array()
430 10 : .ok_or_else(|| bad("datasetId must be an array of URIs (5.2.9)".into()))?;
431 14 : for d in arr {
432 14 : let d = d.as_str().unwrap_or_default();
433 14 : if d != "@none" && antares_model::EntityId::new(d).is_err() {
434 4 : return Err(bad(format!("datasetId entry {d:?} is not a URI (5.2.9)")));
435 10 : }
436 : }
437 6 : out.insert("datasetId".into(), v.clone());
438 : }
439 : // Table 5.2.9-1: scope(s) per the 4.18 grammar.
440 286 : "scope" => {
441 28 : let all_valid = match v {
442 20 : Value::String(s) => antares_jsonld::valid_scope_value(s),
443 8 : Value::Array(a) => a
444 8 : .iter()
445 16 : .all(|s| s.as_str().is_some_and(antares_jsonld::valid_scope_value)),
446 0 : _ => false,
447 : };
448 28 : if !all_valid {
449 4 : return Err(bad("scope violates the 4.18 grammar (5.2.9)".into()));
450 24 : }
451 24 : out.insert("scope".into(), v.clone());
452 : }
453 : // Table 5.2.9-1: GeoJSON geometries per 4.7.
454 : // Table 5.2.9-1: each is a GeoJSON geometry (4.7).
455 258 : "location" | "observationSpace" | "operationSpace" => {
456 52 : let ok = v
457 52 : .as_object()
458 52 : .and_then(|o| Some((o.get("type")?.as_str()?, o.get("coordinates")?)))
459 52 : .is_some_and(|(t, c)| antares_ql::geo::parse_ref_geometry(t, c).is_ok());
460 52 : if !ok {
461 4 : return Err(bad(format!("{k} must be a 4.7 GeoJSON geometry (5.2.9)")));
462 48 : }
463 48 : out.insert(k.to_owned(), v.clone());
464 : }
465 : // Table 5.2.34-1 (RegistrationManagementInfo): cacheDuration an
466 : // ISO 8601 duration, cooldown/timeout numbers greater than 0,
467 : // localOnly a boolean.
468 206 : "management" => {
469 66 : check_registration_management(v)?;
470 18 : out.insert("management".into(), v.clone());
471 : }
472 : // Table 5.2.9-1: an ISO 8601 duration.
473 140 : "refreshRate" => {
474 8 : let ok = v.as_str().is_some_and(valid_iso8601_duration);
475 8 : if !ok {
476 4 : return Err(bad(
477 4 : "refreshRate must be an ISO 8601 duration (5.2.9)".into()
478 4 : ));
479 4 : }
480 4 : out.insert("refreshRate".into(), v.clone());
481 : }
482 132 : "observationInterval" | "managementInterval" => {
483 48 : let o = v
484 48 : .as_object()
485 48 : .ok_or_else(|| bad(format!("{k} must be a TimeInterval object")))?;
486 48 : let start = o
487 48 : .get("startAt")
488 48 : .and_then(Value::as_str)
489 48 : .filter(|s| parse_datetime(s));
490 48 : if start.is_none() {
491 8 : return Err(bad(format!("{k}.startAt must be an ISO 8601 DateTime")));
492 40 : }
493 40 : if let Some(e) = o.get("endAt") {
494 36 : e.as_str()
495 36 : .filter(|s| parse_datetime(s))
496 36 : .ok_or_else(|| bad(format!("{k}.endAt must be an ISO 8601 DateTime")))?;
497 4 : }
498 36 : out.insert(k.to_owned(), v.clone());
499 : }
500 84 : _ => {
501 84 : // tolerant reader: keep unknown members
502 84 : out.insert(k.to_owned(), v.clone());
503 84 : }
504 : }
505 21426 : Ok(())
506 23734 : }
507 :
508 : /// The Table 5.2.9-1 rules that hold between members rather than over
509 : /// one: what a Context Source Registration must carry, and the modes
510 : /// that exclude each other.
511 4698 : fn check_registration_members(out: &Map<String, Value>, is_patch: bool) -> Result<(), NgsiError> {
512 4698 : let bad = NgsiError::BadRequestData;
513 4698 : if !is_patch {
514 4234 : if !out.contains_key("type") {
515 0 : return Err(bad(
516 0 : "type must be \"ContextSourceRegistration\" (5.2.9)".into()
517 0 : ));
518 4234 : }
519 4234 : if !out.contains_key("endpoint") {
520 4 : return Err(bad("endpoint is required (5.2.9)".into()));
521 4230 : }
522 4230 : if !out.contains_key("information") {
523 0 : return Err(bad("information is required (5.2.9)".into()));
524 4230 : }
525 4230 : validate_exclusive(out)?;
526 464 : }
527 4670 : Ok(())
528 4698 : }
529 :
530 : /// Validate + normalize a CSourceRegistration (5.2.9): types and attribute
531 : /// names inside `information` expand to IRIs.
532 5078 : pub fn normalize_registration(
533 5078 : doc: &Map<String, Value>,
534 5078 : ctx: &Context,
535 5078 : is_patch: bool,
536 5078 : ) -> Result<Map<String, Value>, NgsiError> {
537 5078 : let bad = NgsiError::BadRequestData;
538 : // 5.5.4: first-level member nulls are only legal in fragments (patch)
539 5078 : if !is_patch {
540 4600 : antares_jsonld::reject_first_level_nulls(doc)?;
541 478 : }
542 5074 : let mut out = Map::new();
543 23746 : for (k, v) in doc {
544 : // NGSI-LD Fragment member removal (5.4): null / NGSI-LD Null
545 23746 : if is_patch && k != "id" && (v.is_null() || v.as_str() == Some("urn:ngsi-ld:null")) {
546 12 : if ["type", "information", "endpoint"].contains(&k.as_str()) {
547 4 : return Err(bad(format!("cannot remove mandatory member {k} (5.9.3)")));
548 8 : }
549 8 : out.insert(k.clone(), Value::Null);
550 8 : continue;
551 23734 : }
552 23734 : norm_reg_member(k, v, ctx, &mut out)?;
553 : }
554 4698 : check_registration_members(&out, is_patch)?;
555 4670 : Ok(out)
556 5078 : }
557 :
558 : /// ISO 8601 duration (5.2.9 refreshRate): `P[nY][nM][nW][nD][T[nH][nM][nS]]`,
559 : /// at least one component, digits (fraction allowed in seconds).
560 148 : fn valid_iso8601_duration(s: &str) -> bool {
561 148 : antares_model::parse_iso_duration(s).is_some_and(|d| !d.empty)
562 148 : }
563 :
564 : /// 5.9.2.4: an auxiliary registration may only offer "retrieveOps",
565 : /// "retrieveEntity" or "queryEntity" (or a combination thereof) — enforced
566 : /// when the operations member is present (absent = deployment default).
567 4312 : fn validate_auxiliary_ops(doc: &Map<String, Value>) -> Result<(), NgsiError> {
568 4312 : if doc.get("mode").and_then(Value::as_str) != Some("auxiliary") {
569 4276 : return Ok(());
570 36 : }
571 36 : let Some(ops) = doc.get("operations").and_then(Value::as_array) else {
572 8 : return Ok(());
573 : };
574 28 : let allowed = ["retrieveOps", "retrieveEntity", "queryEntity"];
575 28 : if let Some(bad_op) = ops
576 28 : .iter()
577 28 : .filter_map(Value::as_str)
578 32 : .find(|o| !allowed.contains(o))
579 : {
580 12 : return Err(NgsiError::BadRequestData(format!(
581 12 : "auxiliary registration operations are limited to \
582 12 : retrieveOps/retrieveEntity/queryEntity — {bad_op:?} is not allowed (5.9.2.4)"
583 12 : )));
584 16 : }
585 16 : Ok(())
586 4312 : }
587 :
588 : /// 5.9.2.4 registration-vs-entity conflicts. Exclusive: "If an Entity
589 : /// already exists for the supplied Entity ID (URI) and the existing Entity
590 : /// contains any of the Attributes defined in the registration, an error of
591 : /// type Conflict shall be raised." Redirect: "If an existing Entity already
592 : /// matches the Context Source Registration, an error of type Conflict shall
593 : /// be raised."
594 : ///
595 : /// Read shape, not read volume: this runs under the process-wide
596 : /// registration write lock, so a fold of the tenant here stalls every other
597 : /// registration write on the broker, and a whole-tenant query above the
598 : /// store's row ceiling (5.5.6) would refuse a registration create that has
599 : /// no TooManyResults to raise. An EntityInfo that names a concrete id is
600 : /// answered by reading that Entity — the only shape an exclusive
601 : /// registration has, since 4.3.6.3 requires an entity id. Everything else
602 : /// (an `idPattern`, or a type alone) is answered by walking the tenant a
603 : /// page at a time, asking every EntityInfo about each Entity as it arrives
604 : /// rather than re-reading the tenant per EntityInfo.
605 4348 : async fn check_entity_conflict(
606 4348 : st: &AppState,
607 4348 : tenant: &antares_model::TenantId,
608 4348 : doc: &Map<String, Value>,
609 4348 : ) -> Result<(), NgsiError> {
610 4348 : let mode = doc
611 4348 : .get("mode")
612 4348 : .and_then(Value::as_str)
613 4348 : .unwrap_or("inclusive");
614 4348 : if mode != "exclusive" && mode != "redirect" {
615 2558 : return Ok(());
616 1790 : }
617 1790 : let infos = doc
618 1790 : .get("information")
619 1790 : .and_then(Value::as_array)
620 1790 : .map(Vec::as_slice)
621 1790 : .unwrap_or_default();
622 1794 : for info in infos {
623 1794 : let attrs: Vec<String> = antares_model::registration_attribute_names(info)
624 1794 : .unwrap_or_default()
625 1794 : .into_iter()
626 1794 : .map(str::to_owned)
627 1794 : .collect();
628 1794 : let ents = info
629 1794 : .get("entities")
630 1794 : .and_then(Value::as_array)
631 1794 : .map(Vec::as_slice)
632 1794 : .unwrap_or_default();
633 1794 : if ents.is_empty() {
634 0 : continue;
635 1794 : }
636 : // Every selector of this RegistrationInfo, its idPattern compiled
637 : // once: an Entity is read at most once and asked about all of them.
638 1794 : let wants: Vec<_> = ents
639 1794 : .iter()
640 1830 : .map(|e| {
641 : (
642 1830 : e.get("id").and_then(Value::as_str),
643 : // 5.2.8: an EntityInfo type is a String or a String[]
644 1830 : ei_types(e),
645 1830 : e.get("idPattern")
646 1830 : .and_then(Value::as_str)
647 1830 : .and_then(|p| antares_ql::regex::compile(p).ok()),
648 : )
649 1830 : })
650 1794 : .collect();
651 4180 : let hit = |existing: &Value| -> Option<NgsiError> {
652 4180 : let eid = existing.get("id").and_then(Value::as_str).unwrap_or("");
653 4216 : for (want_id, want_types, pattern) in &wants {
654 4216 : let id_hit = match (want_id, pattern) {
655 164 : (Some(w), _) => *w == eid,
656 4036 : (None, Some(re)) => re.is_match(eid),
657 16 : (None, None) => true,
658 : };
659 4216 : if !id_hit {
660 4052 : continue;
661 164 : }
662 164 : if !want_types.is_empty() {
663 152 : let matches_type =
664 152 : existing
665 152 : .get("type")
666 152 : .and_then(Value::as_array)
667 152 : .is_some_and(|ts| {
668 152 : ts.iter()
669 152 : .filter_map(Value::as_str)
670 152 : .any(|x| want_types.contains(&x))
671 152 : });
672 152 : if !matches_type {
673 12 : continue;
674 140 : }
675 12 : }
676 152 : let conflict = match mode {
677 : // exclusive names concrete Attributes (4.3.6.3) — only
678 : // an entity already carrying one of them conflicts
679 152 : "exclusive" => attrs.iter().any(|a| existing.get(a).is_some()),
680 : // redirect: any matching entity conflicts
681 32 : _ => true,
682 : };
683 152 : if conflict {
684 44 : return Some(NgsiError::Conflict(format!(
685 44 : "existing entity {eid} conflicts with the {mode} registration (5.9.2.4)"
686 44 : )));
687 108 : }
688 : }
689 4136 : None
690 4180 : };
691 :
692 1794 : let ids: Vec<&str> = ents
693 1794 : .iter()
694 1830 : .filter_map(|e| e.get("id").and_then(Value::as_str))
695 1794 : .collect();
696 1794 : if ids.len() == ents.len() && ents.iter().all(|e| e.get("idPattern").is_none()) {
697 : // Every selector names one Entity, so the read is those Entities
698 : // and nothing else — bounded by the registration, not by the
699 : // tenant, whatever the tenant holds.
700 1676 : for id in &ids {
701 1676 : if let Some(existing) = st.store.get(tenant, Kind::Entity, id).await? {
702 132 : if let Some(conflict) = hit(&existing) {
703 24 : return Err(conflict);
704 108 : }
705 1544 : }
706 : }
707 : } else {
708 : // A pattern (or a type alone) can only be answered by the
709 : // Entities of the tenant. The walk stops at the first conflict.
710 : // ponytail: O(tenant) reads under the registration write lock for
711 : // a pattern selector; narrow the walk to the id range of
712 : // `filter::id_pattern_literal` when every selector of the
713 : // RegistrationInfo carries an anchored literal.
714 4048 : walk_docs(st, tenant, Kind::Entity, |existing| match hit(&existing) {
715 20 : Some(conflict) => Err(conflict),
716 4028 : None => Ok(()),
717 4048 : })
718 146 : .await?;
719 : }
720 : }
721 1746 : Ok(())
722 4348 : }
723 :
724 : /// The stored registration for `id`, with 5.9.2.4's deletion applied: "If
725 : /// expiresAt is a date and time in the future, implementations shall delete
726 : /// the Registration when this point in time is reached." The sweep is lazy
727 : /// — "final deletion will always lag the expiresAt timestamp" — so the write
728 : /// that names an expired registration performs the deletion and then sees
729 : /// what every later operation sees, which is what lets a create take the id
730 : /// back: 5.9.2.4 raises AlreadyExists only for a registration that exists,
731 : /// and the store's create is the atomic check-and-insert, so the row has to
732 : /// be gone before it runs. Only writes call this. The read handlers filter
733 : /// on `reg_expired` and touch nothing; the row they hide is freed by
734 : /// `sweep_expired_registrations` on the sweep tick. Unlike a Subscription
735 : /// (5.8.6), a Registration has no `status` member that keeps an expired one
736 : /// visible.
737 5334 : async fn take_live_registration(
738 5334 : st: &AppState,
739 5334 : tenant: &TenantId,
740 5334 : id: &str,
741 5334 : ) -> Result<Option<Value>, NgsiError> {
742 5334 : match st.store.get(tenant, Kind::Registration, id).await? {
743 1444 : Some(doc) if reg_expired(&doc) => {
744 6 : if st.store.delete(tenant, Kind::Registration, id).await? {
745 6 : st.reg_changed(tenant, id, None);
746 6 : }
747 6 : Ok(None)
748 : }
749 5328 : live => Ok(live),
750 : }
751 5334 : }
752 :
753 : /// 4.22 for registrations: the read paths refuse an expired registration
754 : /// (`reg_expired`), and this is what removes it. Without a collector the
755 : /// rows a read already hides stay for the life of the broker, so the
756 : /// predicate that hides one is the predicate that reaps it.
757 : ///
758 : /// A tenant the driver refuses to enumerate reaps nothing rather than
759 : /// failing the tick: the sweep is opportunistic, and every read still
760 : /// hides what it leaves behind.
761 145058 : pub(crate) async fn sweep_expired_registrations(st: &AppState, tenant: &TenantId) -> usize {
762 145058 : let mut dead: Vec<String> = Vec::new();
763 10400602 : if walk_docs(st, tenant, Kind::Registration, |doc| {
764 10393006 : if reg_expired(&doc) {
765 8 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
766 8 : dead.push(id.to_owned());
767 8 : }
768 10392998 : }
769 10393006 : Ok(())
770 10393006 : })
771 145058 : .await
772 145058 : .is_err()
773 : {
774 12 : return 0;
775 145046 : }
776 145046 : let mut n: usize = 0;
777 145046 : for id in dead {
778 8 : if st
779 8 : .store
780 8 : .delete(tenant, Kind::Registration, &id)
781 8 : .await
782 8 : .unwrap_or(false)
783 8 : {
784 8 : st.reg_changed(tenant, &id, None);
785 8 : n = n.saturating_add(1);
786 8 : }
787 : }
788 145046 : n
789 145058 : }
790 :
791 : /// 4.3.6.3 Proxied Registrations: "An exclusive registration shall always
792 : /// relate to specific Attributes found on a single Entity. Thus, the
793 : /// registration shall define both: an entity id (i.e. an id pattern or Entity
794 : /// type defining a group of entities is not supported for exclusive
795 : /// registrations) `[and]` Attributes."
796 4674 : pub fn validate_exclusive(doc: &Map<String, Value>) -> Result<(), NgsiError> {
797 4674 : if doc.get("mode").and_then(Value::as_str) != Some("exclusive") {
798 3252 : return Ok(());
799 1422 : }
800 1422 : let bad = |m: &str| NgsiError::BadRequestData(format!("{m} (4.3.6.3)"));
801 1422 : let infos = doc
802 1422 : .get("information")
803 1422 : .and_then(Value::as_array)
804 1422 : .map(Vec::as_slice)
805 1422 : .unwrap_or_default();
806 1422 : for info in infos {
807 1422 : let has_attrs = antares_model::registration_attribute_names(info)
808 1422 : .is_some_and(|names| !names.is_empty());
809 1422 : if !has_attrs {
810 8 : return Err(bad("an exclusive registration shall define Attributes"));
811 1414 : }
812 1414 : let ids_only = info
813 1414 : .get("entities")
814 1414 : .and_then(Value::as_array)
815 1414 : .is_some_and(|es| {
816 1442 : es.iter().all(|e| {
817 1442 : e.get("id").and_then(Value::as_str).is_some() && e.get("idPattern").is_none()
818 1442 : })
819 1414 : });
820 1414 : if !ids_only {
821 16 : return Err(bad(
822 16 : "an exclusive registration shall name an entity id — an id pattern or \
823 16 : Entity type defining a group of entities is not supported",
824 16 : ));
825 1398 : }
826 : }
827 1398 : Ok(())
828 4674 : }
829 :
830 : /// 4.3.6.3: "Once an exclusive Context Source Registration has been created,
831 : /// no further exclusive or redirect Context Source Registrations can be
832 : /// created for that same combination of Entity ID and Attributes" — and per
833 : /// 5.9.2, registering an exclusive Context Source when "an exclusive or
834 : /// redirect Context Source Registration already matches against the Entity ID
835 : /// (URI) and any of the Attributes defined in the registration" raises a
836 : /// Conflict (409; Table 6.3.2-1 defines no Conflict type, so it travels as
837 : /// AlreadyExists — the project's standing 409 mapping). Redirect overlapping
838 : /// redirect stays legal: "operations are distributed to all registered
839 : /// Context Sources".
840 4300 : pub async fn check_proxied_overlap(
841 4300 : st: &AppState,
842 4300 : tenant: &antares_model::TenantId,
843 4300 : doc: &Map<String, Value>,
844 4300 : self_id: Option<&str>,
845 4300 : ctx: &Context,
846 4300 : ) -> Result<(), NgsiError> {
847 4300 : let mode = doc
848 4300 : .get("mode")
849 4300 : .and_then(Value::as_str)
850 4300 : .unwrap_or("inclusive");
851 4300 : if mode != "exclusive" && mode != "redirect" {
852 2558 : return Ok(());
853 1742 : }
854 1742 : let infos = doc
855 1742 : .get("information")
856 1742 : .and_then(Value::as_array)
857 1742 : .map(Vec::as_slice)
858 1742 : .unwrap_or_default();
859 : // Fail closed: treating a lookup failure as "no conflicts" would admit a
860 : // second exclusive registration for the same scope.
861 2243 : walk_docs(st, tenant, Kind::Registration, |other| {
862 2243 : let other = &other;
863 2243 : if other.get("id").and_then(Value::as_str) == self_id {
864 331 : return Ok(());
865 1912 : }
866 1912 : if reg_expired(other) {
867 0 : return Ok(());
868 1912 : }
869 1912 : let omode = other
870 1912 : .get("mode")
871 1912 : .and_then(Value::as_str)
872 1912 : .unwrap_or("inclusive");
873 : // new exclusive × existing proxied; new redirect × existing exclusive
874 1912 : let guarded = match mode {
875 1912 : "exclusive" => omode == "exclusive" || omode == "redirect",
876 116 : _ => omode == "exclusive",
877 : };
878 1912 : if !guarded {
879 882 : return Ok(());
880 1030 : }
881 1030 : for info in infos {
882 1030 : let attrs: Vec<String> = antares_model::registration_attribute_names(info)
883 1030 : .unwrap_or_default()
884 1030 : .into_iter()
885 1030 : .map(str::to_owned)
886 1030 : .collect();
887 1030 : let empty = Vec::new();
888 1030 : let entities = info
889 1030 : .get("entities")
890 1030 : .and_then(Value::as_array)
891 1030 : .unwrap_or(&empty);
892 1030 : let ids: Vec<String> = entities
893 1030 : .iter()
894 1030 : .filter_map(|e| e.get("id").and_then(Value::as_str))
895 1030 : .map(str::to_owned)
896 1030 : .collect();
897 1030 : let types: Vec<String> = entities
898 1030 : .iter()
899 1030 : .flat_map(ei_types)
900 1030 : .map(str::to_owned)
901 1030 : .collect();
902 1030 : let spec = CsrSpec {
903 1030 : types: (!types.is_empty()).then_some(types),
904 1030 : ids: (!ids.is_empty()).then_some(ids),
905 1030 : id_pattern: entities
906 1030 : .iter()
907 1030 : .find_map(|e| e.get("idPattern").and_then(Value::as_str))
908 1030 : .map(str::to_owned),
909 1030 : attrs: (!attrs.is_empty()).then_some(attrs),
910 1030 : dataset_ids: None,
911 1030 : csf: None,
912 1030 : geo: None,
913 1030 : temporal: None,
914 : };
915 1030 : if csr_matches(&spec, other, ctx) {
916 1018 : let oid = other.get("id").and_then(Value::as_str).unwrap_or("?");
917 1018 : return Err(NgsiError::AlreadyExists(format!(
918 1018 : "proxied registration overlaps {oid} for the same combination of \
919 1018 : Entity ID and Attributes (4.3.6.3)"
920 1018 : )));
921 12 : }
922 : }
923 12 : Ok(())
924 2243 : })
925 1742 : .await
926 4300 : }
927 :
928 : /// The 5.9.2.4 conflict rules ("if an exclusive or redirect Context Source
929 : /// Registration already matches … an error of type Conflict shall be
930 : /// raised") are decided by reading the registration set, and the create or
931 : /// update that follows writes it. Read and write are separate store
932 : /// operations, so without this lock two requests can each observe a
933 : /// conflict-free set and both land, leaving the two exclusive registrations
934 : /// for one Entity ID and Attribute the clause forbids. Every registration
935 : /// write holds it for the whole check-then-write sequence.
936 : ///
937 : /// One lock per tenant: the clause decides a conflict against the
938 : /// registrations of the tenant being written, and two tenants' sets are
939 : /// disjoint, so the overlap scan of one tenant's idPattern-only
940 : /// registration never stalls another tenant's write. The map only grows,
941 : /// one entry per tenant that ever wrote a registration.
942 : ///
943 : /// A `tokio::sync::Mutex`, not a `std` one: the guarded section awaits the
944 : /// store between the check and the write, and a `std` guard cannot be held
945 : /// across an await.
946 : ///
947 : /// ponytail: process-local, so two broker pods sharing one database do not
948 : /// exclude each other and the check-then-write races between them (the gap
949 : /// the 5.9.2.4 ledger entry names). A database lock held across the section
950 : /// is the upgrade, and it pins a connection for the whole overlap scan —
951 : /// holders that each still need a connection to run that scan exhaust the
952 : /// pool and wait on each other, so the upgrade owes a bound on how many may
953 : /// hold at once.
954 : static REGISTRATION_WRITE: std::sync::Mutex<
955 : std::collections::BTreeMap<String, std::sync::Arc<tokio::sync::Mutex<()>>>,
956 : > = std::sync::Mutex::new(std::collections::BTreeMap::new());
957 :
958 4336 : async fn registration_write_lock(tenant: &TenantId) -> tokio::sync::OwnedMutexGuard<()> {
959 4336 : let lock = REGISTRATION_WRITE
960 4336 : .lock()
961 4336 : .unwrap_or_else(std::sync::PoisonError::into_inner)
962 4336 : .entry(tenant.as_str().to_owned())
963 4336 : .or_default()
964 4336 : .clone();
965 4336 : lock.lock_owned().await
966 4332 : }
967 :
968 : /// The expanded types and Attribute names a registration's information
969 : /// names, for the policy gate (ADR-0020), each once.
970 3870 : fn info_members(norm: &Map<String, Value>) -> (Vec<String>, Vec<String>) {
971 3870 : let (mut types, mut attrs): (Vec<String>, Vec<String>) = (Vec::new(), Vec::new());
972 16134 : let add = |out: &mut Vec<String>, v: Option<&Value>| {
973 16134 : let names: Vec<&str> = match v {
974 4130 : Some(Value::String(s)) => vec![s.as_str()],
975 2158 : Some(Value::Array(a)) => a.iter().filter_map(Value::as_str).collect(),
976 9846 : _ => Vec::new(),
977 : };
978 16134 : for n in names {
979 6638 : if !out.iter().any(|o| o == n) {
980 6486 : out.push(n.to_owned());
981 6486 : }
982 : }
983 16134 : };
984 4000 : for info in norm
985 3870 : .get("information")
986 3870 : .and_then(Value::as_array)
987 3870 : .into_iter()
988 3870 : .flatten()
989 : {
990 4134 : for e in info
991 4000 : .get("entities")
992 4000 : .and_then(Value::as_array)
993 4000 : .into_iter()
994 4000 : .flatten()
995 4134 : {
996 4134 : add(&mut types, e.get("type"));
997 4134 : }
998 : // every name the request carries, the deprecated pair beside
999 : // attributeNames included: the engine judges what was sent
1000 12000 : for k in antares_model::REGISTRATION_ATTRIBUTE_MEMBERS {
1001 12000 : add(&mut attrs, info.get(k));
1002 12000 : }
1003 : }
1004 3870 : (types, attrs)
1005 3870 : }
1006 :
1007 : // ---------- handlers ----------
1008 :
1009 4128 : pub async fn create_registration(
1010 4128 : State(st): State<AppState>,
1011 4128 : CleanParams(params): CleanParams,
1012 4128 : headers: HeaderMap,
1013 4128 : body: Bytes,
1014 4128 : ) -> Response {
1015 4128 : let go = async {
1016 4128 : let tenant = tenant_from(&headers)?;
1017 4128 : check_params(¶ms, &["local"])?;
1018 4128 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
1019 4108 : let obj = parsed.value.as_object().ok_or_else(|| {
1020 0 : NgsiError::BadRequestData("registration must be a JSON object".into())
1021 0 : })?;
1022 : // ADR-0020: 5.9.2 lets the client choose the registration id, and
1023 : // it is in hand here — see the same note on subscription create.
1024 4108 : let named = obj.get("id").and_then(Value::as_str);
1025 4108 : let mut norm = normalize_registration(obj, &parsed.ctx, false)?;
1026 3870 : let (types, attrs) = info_members(&norm);
1027 3870 : gate!(
1028 : st, &tenant, &headers, "5.9.2",
1029 : ids: named.as_slice(),
1030 : types: &types,
1031 : attrs: &attrs,
1032 : )
1033 3870 : .await?;
1034 3868 : let id = match norm.get("id").and_then(Value::as_str) {
1035 3834 : Some(id) => id.to_owned(),
1036 : None => {
1037 34 : let id = format!(
1038 : "urn:ngsi-ld:ContextSourceRegistration:{}",
1039 34 : uuid::Uuid::new_v4()
1040 : );
1041 34 : norm.insert("id".into(), Value::String(id.clone()));
1042 34 : id
1043 : }
1044 : };
1045 3868 : validate_auxiliary_ops(&norm)?;
1046 3860 : crate::federation::admit_registration(&st, &tenant, &norm).await?;
1047 : // The creating subject, kept as a Subscription keeps it: an engine
1048 : // narrows Query Registrations (5.10.2) by the subject the broker
1049 : // recorded, never by a member the client wrote.
1050 20 : if let Some(subject) =
1051 3860 : crate::policy::subject_member(&crate::policy::subject_of(&tenant, &headers))
1052 20 : {
1053 20 : norm.insert(crate::policy::SUBJECT_MEMBER.into(), subject);
1054 3840 : }
1055 3130 : let doc = {
1056 3860 : let _serialized = registration_write_lock(&tenant).await;
1057 3860 : take_live_registration(&st, &tenant, &id).await?;
1058 3860 : check_entity_conflict(&st, &tenant, &norm).await?;
1059 3844 : check_proxied_overlap(&st, &tenant, &norm, None, &parsed.ctx).await?;
1060 3134 : let ts = now_iso();
1061 3134 : norm.insert("createdAt".into(), Value::String(ts.clone()));
1062 3134 : norm.insert("modifiedAt".into(), Value::String(ts));
1063 3134 : let doc = Value::Object(norm);
1064 3134 : if !st
1065 3134 : .store
1066 3134 : .create(&tenant, Kind::Registration, &id, doc.clone())
1067 3134 : .await?
1068 : {
1069 4 : return Err(
1070 4 : NgsiError::AlreadyExists(format!("registration {id} already exists")).into(),
1071 4 : );
1072 3130 : }
1073 3130 : doc
1074 : };
1075 3130 : st.reg_changed(&tenant, &id, Some(&doc));
1076 3130 : crate::notify::csource_fanout(&st, &tenant, None, Some(doc)).await;
1077 3130 : Ok::<_, ApiError>(created(
1078 3130 : format!("/ngsi-ld/v1/csourceRegistrations/{id}"),
1079 3130 : &tenant,
1080 3130 : ))
1081 4128 : };
1082 4128 : go.await.unwrap_or_else(|e| e.into_response())
1083 4128 : }
1084 :
1085 178 : pub async fn retrieve_registration(
1086 178 : State(st): State<AppState>,
1087 178 : Path(id): Path<String>,
1088 178 : CleanParams(params): CleanParams,
1089 178 : headers: HeaderMap,
1090 178 : ) -> Response {
1091 178 : let go = async {
1092 178 : let tenant = tenant_from(&headers)?;
1093 178 : antares_model::EntityId::new(&id)
1094 178 : .map_err(|_| NgsiError::BadRequestData(format!("invalid registration id {id:?}")))?;
1095 174 : check_params(¶ms, &["options", "format", "local"])?;
1096 174 : let accept = parse_accept(&headers)?;
1097 174 : let ctx = request_context(&st.loader, &headers).await?;
1098 174 : gate!(st, &tenant, &headers, "5.10.1", ids: &[&id]).await?;
1099 174 : let doc = st
1100 174 : .store
1101 174 : .get(&tenant, Kind::Registration, &id)
1102 174 : .await?
1103 174 : .filter(|d| !reg_expired(d))
1104 174 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("registration {id} not found")))?;
1105 132 : let sys = sys_attrs_asked(¶ms);
1106 132 : Ok::<_, ApiError>(respond(
1107 132 : StatusCode::OK,
1108 132 : present_registration(&doc, &ctx, sys),
1109 132 : &ctx,
1110 132 : accept,
1111 132 : &tenant,
1112 132 : ))
1113 178 : };
1114 178 : go.await.unwrap_or_else(|e| e.into_response())
1115 178 : }
1116 :
1117 298 : pub async fn query_registrations(
1118 298 : State(st): State<AppState>,
1119 298 : CleanParams(params): CleanParams,
1120 298 : headers: HeaderMap,
1121 298 : ) -> Response {
1122 298 : let go = async {
1123 298 : let tenant = tenant_from(&headers)?;
1124 298 : check_params(
1125 298 : ¶ms,
1126 298 : &[
1127 298 : "id",
1128 298 : "idPattern",
1129 298 : "type",
1130 298 : "attrs",
1131 298 : "q",
1132 298 : "georel",
1133 298 : "geometry",
1134 298 : "coordinates",
1135 298 : "geoproperty",
1136 298 : "timeproperty",
1137 298 : "timerel",
1138 298 : "timeAt",
1139 298 : "endTimeAt",
1140 298 : "csf",
1141 298 : "limit",
1142 298 : "offset",
1143 298 : "count",
1144 298 : "options",
1145 298 : "format",
1146 298 : "local",
1147 298 : "scopeQ",
1148 298 : "pick",
1149 298 : "omit",
1150 298 : ],
1151 10 : )?;
1152 : // Table 6.8.3.2-1: pick and omit, refused together as for Entities
1153 : // (repr::check_projection_exclusive; no clause states it)
1154 288 : if params.contains_key("pick") && params.contains_key("omit") {
1155 2 : return Err(NgsiError::BadRequestData(
1156 2 : "pick and omit cannot be combined in one request".into(),
1157 2 : )
1158 2 : .into());
1159 286 : }
1160 286 : let accept = parse_accept(&headers)?;
1161 278 : let ctx = request_context(&st.loader, &headers).await?;
1162 278 : let narrowing = gate!(
1163 : st, &tenant, &headers, "5.10.2",
1164 : scope_q: params.get("scopeQ").map(String::as_str),
1165 : )
1166 278 : .await?;
1167 274 : let cond = crate::policy::Condition::of(&narrowing);
1168 274 : let bad = NgsiError::BadRequestData;
1169 274 : let mut spec = CsrSpec::default();
1170 274 : if let Some(s) = params.get("id") {
1171 16 : let mut ids = Vec::new();
1172 20 : for i in s.split(',') {
1173 20 : antares_model::EntityId::new(i)
1174 20 : .map_err(|_| bad(format!("invalid id in list: {i:?}")))?;
1175 16 : ids.push(i.to_owned());
1176 : }
1177 12 : spec.ids = Some(ids);
1178 258 : }
1179 270 : spec.id_pattern = params
1180 270 : .get("idPattern")
1181 270 : .map(|p| {
1182 8 : antares_ql::regex::compile(p)
1183 8 : .map(|_| p.clone())
1184 8 : .map_err(|_| bad(format!("invalid idPattern {p:?}")))
1185 8 : })
1186 270 : .transpose()?;
1187 : // Table 6.8.3.2-1: `type` is a "Selection of Entity Types as per
1188 : // clause 4.17", i.e. ONE expression — splitting it on ',' and
1189 : // expanding the fragments mangles every selector that uses ';' or
1190 : // parentheses. entity_info_matches evaluates it whole.
1191 270 : spec.types = params.get("type").cloned().map(|s| vec![s]);
1192 270 : let mut attrs: Vec<String> = params
1193 270 : .get("attrs")
1194 270 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect())
1195 270 : .unwrap_or_default();
1196 : // attributes referenced in q / geoQ count as query projection
1197 : // attributes for matching (5.10.2.4)
1198 : // `CleanParams` percent-decodes every value once, at the extractor
1199 : // (6.3.1). Decoding again here reads escapes that are part of the
1200 : // value, so a `q` legitimately containing `%22` became one carrying a
1201 : // bare quote and 4.9's parser refused a legal query. `csf` below
1202 : // takes the extractor's form, and so does this.
1203 270 : if let Some(q) = params.get("q") {
1204 10 : let ast = antares_ql::parse_q(q)?;
1205 6 : let mut roots = Vec::new();
1206 6 : q_attr_roots(&ast, &mut roots);
1207 6 : attrs.extend(roots.into_iter().map(|r| ctx.expand_key(&r)));
1208 260 : }
1209 266 : let geo = antares_ql::geo::GeoQuery::from_params(¶ms)?;
1210 262 : if let Some(g) = &geo {
1211 16 : attrs.push(ctx.expand_key(&g.geoproperty));
1212 246 : }
1213 262 : if !attrs.is_empty() {
1214 42 : spec.attrs = Some(attrs);
1215 220 : }
1216 : // 5.10.2.4: a discriminating input is required, else too wide
1217 : // (the suite additionally accepts id-only queries — 037_10_01)
1218 262 : if spec.types.is_none()
1219 68 : && spec.attrs.is_none()
1220 60 : && spec.ids.is_none()
1221 56 : && spec.id_pattern.is_none()
1222 : {
1223 56 : return Err(bad(
1224 56 : "query too wide: one of type, attrs, q or geo query is required (5.10.2.4)".into(),
1225 56 : )
1226 56 : .into());
1227 206 : }
1228 : // 5.10.2.4: csf is a 4.9 query over Context Source Properties
1229 206 : let csf = params
1230 206 : .get("csf")
1231 206 : .map(|c| antares_ql::parse_q(c))
1232 206 : .transpose()?;
1233 206 : let scope_q = params.get("scopeQ").cloned();
1234 : // temporal query: validate + interval presence rules (5.10.2.4)
1235 206 : let temporal = crate::temporalq::TemporalQ::from_params(¶ms, false)?
1236 206 : .filter(|t| t.timerel != "any");
1237 : // 5.10.2.4 fixes the order: run the query returning the
1238 : // registrations that "meet all the applicable conditions", THEN
1239 : // "Pagination logic shall be in place as mandated by clause 5.5.9".
1240 : // Filter first, page second — so the window 5.8.4 pushes into the
1241 : // store cannot be pushed here, where it would cut the page out of
1242 : // the stored rows and serve registrations the query never matched.
1243 : //
1244 : // What CAN be bounded is the read. Walking the tenant in pages keeps
1245 : // the peak at one page plus the matches, and moves the ceiling from
1246 : // what the tenant STORES to what the query MATCHES; 5.5.6 licenses
1247 : // TooManyResults for "a query operation ... producing so many
1248 : // results that can potentially exhaust client or server resources",
1249 : // which is a statement about the result and not about the store.
1250 543 : let keep = |doc: &Value| {
1251 543 : if reg_expired(doc) {
1252 5 : return false;
1253 538 : }
1254 538 : let has_interval =
1255 538 : doc.get("observationInterval").is_some() || doc.get("managementInterval").is_some();
1256 522 : match &temporal {
1257 0 : None if has_interval => return false,
1258 16 : Some(tq) if !temporal_interval_matches(doc, tq) => return false,
1259 538 : _ => {}
1260 : }
1261 : // 5.10.2.4: csf vs Context Source Properties, Scope query vs
1262 : // the registration scope, geoquery vs its location
1263 538 : if let Some(csf) = &csf {
1264 20 : if !csf_matches(csf, doc, &ctx) {
1265 12 : return false;
1266 8 : }
1267 518 : }
1268 526 : if let Some(sq) = &scope_q {
1269 16 : if !crate::scope_matches(sq, doc) {
1270 8 : return false;
1271 8 : }
1272 510 : }
1273 518 : if let Some(g) = &geo {
1274 24 : match doc.get("location") {
1275 16 : Some(geom) if g.matches_geometry(geom) => {}
1276 8 : _ => return false,
1277 : }
1278 494 : }
1279 510 : csr_matches(&spec, doc, &ctx) && cond.as_ref().is_none_or(|c| c.holds_for_record(doc))
1280 543 : };
1281 206 : let matches = collect_matching(&st, &tenant, keep, *crate::bounds::MAX_FOLD_DOCS).await?;
1282 206 : let (page, count_hdr, links) = crate::paging::paginate_accept(
1283 206 : &st,
1284 206 : ¶ms,
1285 206 : matches,
1286 206 : "/ngsi-ld/v1/csourceRegistrations",
1287 206 : accept,
1288 0 : )?;
1289 206 : let sys = sys_attrs_asked(¶ms);
1290 : // Pick/omit name top-level registration members by their
1291 : // compacted name; the table's remarks are the Entity ones
1292 206 : let names =
1293 412 : |k: &str| -> Option<Vec<&str>> { params.get(k).map(|v| v.split(',').collect()) };
1294 206 : let (pick, omit) = (names("pick"), names("omit"));
1295 206 : let payload: Vec<Value> = page
1296 206 : .iter()
1297 208 : .map(|d| {
1298 204 : let mut r = present_registration(d, &ctx, sys);
1299 204 : if let Some(o) = r.as_object_mut() {
1300 932 : o.retain(|k, _| {
1301 932 : pick.as_ref().is_none_or(|p| p.contains(&k.as_str()))
1302 928 : && omit.as_ref().is_none_or(|p| !p.contains(&k.as_str()))
1303 932 : });
1304 0 : }
1305 204 : r
1306 204 : })
1307 206 : .collect();
1308 206 : let mut resp =
1309 206 : crate::negotiate::respond_list(StatusCode::OK, payload, &ctx, accept, &tenant);
1310 206 : attach_paging(&mut resp, count_hdr, &links);
1311 206 : Ok::<_, ApiError>(resp)
1312 298 : };
1313 298 : go.await.unwrap_or_else(|e| e.into_response())
1314 298 : }
1315 :
1316 : /// Registrations per page of the 5.10.2.4 walk: the peak transient
1317 : /// allocation on top of the match set.
1318 : const SCAN_PAGE: usize = 1_000;
1319 :
1320 : /// Visit every document of one kind in one tenant, a page at a time. An
1321 : /// `Err` from `visit` ends the walk and is the walk's own result, which is
1322 : /// how the 5.9.2.4 conflict checks stop at the document they conflict with.
1323 : ///
1324 : /// The whole-tenant `list` and `query_entities` carry the row ceiling meant
1325 : /// for client queries (5.5.6). Every caller below must see EVERY document —
1326 : /// to answer a query over the registrations, to refuse a second exclusive
1327 : /// registration for the same scope, or to find the Entity a redirect
1328 : /// registration would shadow — so that ceiling refused them outright once a
1329 : /// tenant held more than it, whatever the read narrowed to and however few
1330 : /// conflicts existed. A page bounds the allocation by construction and
1331 : /// carries no ceiling, so a large tenant costs time here rather than a
1332 : /// permanent 403.
1333 604445 : pub(crate) async fn walk_docs(
1334 604445 : st: &AppState,
1335 604445 : tenant: &antares_model::TenantId,
1336 604445 : kind: Kind,
1337 604445 : mut visit: impl FnMut(Value) -> Result<(), NgsiError>,
1338 604445 : ) -> Result<(), NgsiError> {
1339 604445 : let mut after: Option<String> = None;
1340 : loop {
1341 614429 : let page = st
1342 614429 : .store
1343 614429 : .list_page(tenant, kind, after.as_deref(), SCAN_PAGE)
1344 614429 : .await?;
1345 614380 : let short = page.len() < SCAN_PAGE;
1346 614380 : let before = after.clone();
1347 10506942 : for doc in page {
1348 10474824 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
1349 10474792 : after = Some(id.to_owned());
1350 10474792 : }
1351 10474824 : visit(doc)?;
1352 : }
1353 : // A short page ends the walk, and so does a cursor that did not
1354 : // move: only a document carrying an `id` advances it, so a full page
1355 : // without one would otherwise be re-read forever.
1356 613338 : if short || after == before {
1357 603354 : break;
1358 9984 : }
1359 : }
1360 603354 : Ok(())
1361 604445 : }
1362 :
1363 : /// Every registration of one tenant that `keep` accepts.
1364 : /// Every registration of one tenant that `keep` accepts, up to `ceiling`.
1365 : ///
1366 : /// 5.10.2.4 filters before it pages, so the page cannot be pushed into the
1367 : /// store and the whole match set is held at once. A broker is built for
1368 : /// 100 000+ registrations per tenant, so "the whole match set" is a number a
1369 : /// client picks with one `type=` — and 5.5.6 gives the answer for "a query
1370 : /// operation … producing so many results that can potentially exhaust client
1371 : /// or server resources": TooManyResults, rather than the memory.
1372 218 : async fn collect_matching(
1373 218 : st: &AppState,
1374 218 : tenant: &antares_model::TenantId,
1375 218 : keep: impl Fn(&Value) -> bool,
1376 218 : ceiling: usize,
1377 218 : ) -> Result<Vec<Value>, NgsiError> {
1378 218 : let mut matches = Vec::new();
1379 595 : walk_docs(st, tenant, Kind::Registration, |doc| {
1380 595 : if keep(&doc) {
1381 356 : if matches.len() == ceiling {
1382 4 : return Err(NgsiError::TooManyResults(format!(
1383 4 : "the query matches more than {ceiling} registrations — narrow it (5.5.6)"
1384 4 : )));
1385 352 : }
1386 352 : matches.push(doc);
1387 239 : }
1388 591 : Ok(())
1389 595 : })
1390 218 : .await?;
1391 214 : Ok(matches)
1392 218 : }
1393 :
1394 : /// Root attribute names referenced by a q= expression (5.10.2.4: they count
1395 : /// as query projection attributes for RegistrationInfo matching).
1396 6 : fn q_attr_roots(node: &antares_ql::QNode, out: &mut Vec<String>) {
1397 : use antares_ql::QNode::*;
1398 6 : match node {
1399 0 : And(v) | Or(v) => v.iter().for_each(|n| q_attr_roots(n, out)),
1400 6 : Cmp { path, .. } | Exists { path, .. } => {
1401 6 : if let Some(r) = path.top() {
1402 6 : out.push(r.to_owned());
1403 6 : }
1404 : }
1405 : }
1406 6 : }
1407 :
1408 : /// 5.9.3 Update Context Source Registration: invalid URI 400, unknown 404,
1409 : /// 5.2.9 fragment merge per 5.5.8 with every mode rule re-checked on the
1410 : /// post-merge document (4.3.6.3 exclusive shape, auxiliary ops limit,
1411 : /// entity/registration conflicts).
1412 482 : pub async fn update_registration(
1413 482 : State(st): State<AppState>,
1414 482 : Path(id): Path<String>,
1415 482 : CleanParams(params): CleanParams,
1416 482 : headers: HeaderMap,
1417 482 : body: Bytes,
1418 482 : ) -> Response {
1419 482 : let go = async {
1420 482 : let tenant = tenant_from(&headers)?;
1421 482 : antares_model::EntityId::new(&id)
1422 482 : .map_err(|_| NgsiError::BadRequestData(format!("invalid registration id {id:?}")))?;
1423 478 : check_params(¶ms, &["local"])?;
1424 478 : let parsed = parse_body(&st, &headers, &body, BodyKind::MergePatch).await?;
1425 474 : let obj = parsed.object(NgsiError::BadRequestData(
1426 474 : "fragment must be a JSON object".into(),
1427 474 : ))?;
1428 474 : gate!(st, &tenant, &headers, "5.9.3", ids: &[&id]).await?;
1429 474 : let norm = normalize_registration(obj, &parsed.ctx, true)?;
1430 460 : let ts = now_iso();
1431 : // The 5.9.3.4 re-checks below read the registration set that the
1432 : // mutate then writes — the pair is atomic or a concurrent write can
1433 : // invalidate the checks between them.
1434 152 : let (before, res) = {
1435 460 : let _serialized = registration_write_lock(&tenant).await;
1436 460 : let before = take_live_registration(&st, &tenant, &id).await?;
1437 460 : if let Some(prev) = before.as_ref().and_then(Value::as_object) {
1438 : // validate the post-merge document (4.3.6.3) BEFORE mutating:
1439 : // a patch may flip the mode or rewrite information
1440 444 : let mut merged = prev.clone();
1441 444 : for (k, v) in &norm {
1442 440 : if k == "id" {
1443 0 : continue;
1444 440 : }
1445 440 : if v.is_null() {
1446 4 : merged.remove(k);
1447 436 : } else {
1448 436 : merged.insert(k.clone(), v.clone());
1449 436 : }
1450 : }
1451 444 : validate_exclusive(&merged)?;
1452 : // 5.9.3.4: the mode-specific rules apply to the merged document
1453 444 : validate_auxiliary_ops(&merged)?;
1454 440 : crate::federation::admit_registration(&st, &tenant, &merged).await?;
1455 440 : check_entity_conflict(&st, &tenant, &merged).await?;
1456 436 : check_proxied_overlap(&st, &tenant, &merged, Some(&id), &parsed.ctx).await?;
1457 16 : }
1458 152 : let res = st
1459 152 : .store
1460 152 : .mutate(&tenant, Kind::Registration, &id, |doc| {
1461 136 : let Some(target) = doc.as_object_mut() else {
1462 0 : return Err(NgsiError::InternalError(
1463 0 : "stored registration is not a JSON object".into(),
1464 0 : ));
1465 : };
1466 136 : crate::apply_doc_fragment(target, &norm, &ts);
1467 136 : Ok::<(), NgsiError>(())
1468 136 : })
1469 152 : .await?;
1470 152 : (before, res)
1471 : };
1472 136 : match res {
1473 16 : None => Err(NgsiError::ResourceNotFound(format!("registration {id} not found")).into()),
1474 0 : Some(Err(e)) => Err(ApiError::from(e)),
1475 : Some(Ok(())) => {
1476 136 : let after = st.store.get(&tenant, Kind::Registration, &id).await?;
1477 136 : st.reg_changed(&tenant, &id, after.as_ref());
1478 136 : crate::notify::csource_fanout(&st, &tenant, before, after).await;
1479 136 : Ok(no_content(&tenant))
1480 : }
1481 : }
1482 482 : };
1483 482 : go.await.unwrap_or_else(|e| e.into_response())
1484 482 : }
1485 :
1486 : /// 5.9.4 Delete Context Source Registration: invalid URI 400, unknown id
1487 : /// 404, 204 on removal (registry mirror + csource subscriptions refresh).
1488 1018 : pub async fn delete_registration(
1489 1018 : State(st): State<AppState>,
1490 1018 : Path(id): Path<String>,
1491 1018 : CleanParams(params): CleanParams,
1492 1018 : headers: HeaderMap,
1493 1018 : ) -> Response {
1494 1018 : let go = async {
1495 1018 : let tenant = tenant_from(&headers)?;
1496 1018 : antares_model::EntityId::new(&id)
1497 1018 : .map_err(|_| NgsiError::BadRequestData(format!("invalid registration id {id:?}")))?;
1498 1014 : check_params(¶ms, &["local"])?;
1499 1014 : gate!(st, &tenant, &headers, "5.9.4", ids: &[&id]).await?;
1500 1014 : let before = take_live_registration(&st, &tenant, &id).await?;
1501 1014 : if st.store.delete(&tenant, Kind::Registration, &id).await? {
1502 990 : st.reg_changed(&tenant, &id, None);
1503 990 : crate::notify::csource_fanout(&st, &tenant, before, None).await;
1504 990 : Ok(no_content(&tenant))
1505 : } else {
1506 24 : Err::<Response, ApiError>(
1507 24 : NgsiError::ResourceNotFound(format!("registration {id} not found")).into(),
1508 24 : )
1509 : }
1510 1018 : };
1511 1018 : go.await.unwrap_or_else(|e| e.into_response())
1512 1018 : }
1513 :
1514 : #[cfg(test)]
1515 : mod clause_4_20 {
1516 : use super::*;
1517 : use antares_jsonld::Loader;
1518 : use serde_json::json;
1519 :
1520 228 : fn reg_with(ops: Value) -> Result<Map<String, Value>, NgsiError> {
1521 228 : let ctx = Loader::new().core();
1522 228 : let doc = json!({
1523 228 : "id": "urn:ngsi-ld:ContextSourceRegistration:ops",
1524 228 : "type": "ContextSourceRegistration",
1525 228 : "endpoint": "http://peer:9090",
1526 228 : "information": [{"entities": [{"type": "Building"}]}],
1527 228 : "operations": ops
1528 : });
1529 228 : normalize_registration(doc.as_object().expect("object"), &ctx, false)
1530 228 : }
1531 :
1532 : /// Table 5.2.9-1: `operations` entries "are limited to the named API
1533 : /// operations and named operation groups (see clause 4.20)". Every name
1534 : /// of Table 4.20-1 and every group of Table 4.20-2 is accepted, and
1535 : /// nothing else is — including a name that only looks like one.
1536 : #[test]
1537 4 : fn only_the_4_20_vocabulary_is_accepted() {
1538 192 : for op in OPERATION_NAMES.iter().chain(OPERATION_GROUPS) {
1539 192 : assert!(reg_with(json!([op])).is_ok(), "{op} is a 4.20 name");
1540 : }
1541 4 : assert!(
1542 4 : reg_with(json!(OPERATION_NAMES.to_vec())).is_ok(),
1543 : "the whole vocabulary at once is legal"
1544 : );
1545 24 : for bad in [
1546 4 : "notARealOp",
1547 4 : "createentity",
1548 4 : "createEntity ",
1549 4 : "federationops",
1550 4 : "queryEntities",
1551 4 : "",
1552 4 : ] {
1553 24 : let e = reg_with(json!([bad])).expect_err("outside the vocabulary");
1554 24 : assert!(
1555 24 : matches!(e, NgsiError::BadRequestData(_)),
1556 : "{bad} must be BadRequestData, got {e:?}"
1557 : );
1558 : }
1559 : // 5.2.9: the member is present-or-absent, never an empty list
1560 4 : assert!(reg_with(json!([])).is_err(), "empty operations");
1561 4 : assert!(reg_with(json!("federationOps")).is_err(), "not an array");
1562 4 : }
1563 : }
1564 :
1565 : #[cfg(test)]
1566 : mod csi_tests {
1567 : use super::*;
1568 : use antares_jsonld::Loader;
1569 : use serde_json::json;
1570 : use std::collections::HashMap;
1571 :
1572 : /// 5.5.4: "urn:ngsi-ld:null" as a first-level member value is
1573 : /// BadRequestData on create; on patch it is the Fragment removal form.
1574 : #[test]
1575 4 : fn clause_5_5_4_first_level_null_in_registration() {
1576 4 : let ctx = Loader::new().core();
1577 4 : let doc = json!({
1578 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:n1",
1579 4 : "type": "ContextSourceRegistration",
1580 4 : "endpoint": "http://peer:9090",
1581 4 : "description": "urn:ngsi-ld:null",
1582 4 : "information": [{"entities": [{"type": "Building"}]}]
1583 : });
1584 4 : assert!(
1585 4 : normalize_registration(doc.as_object().unwrap(), &ctx, false).is_err(),
1586 : "create with a first-level null URN must be rejected"
1587 : );
1588 : // patch: the same member is a removal fragment (stored as Null)
1589 4 : let patch = json!({"description": "urn:ngsi-ld:null"});
1590 4 : let out = normalize_registration(patch.as_object().unwrap(), &ctx, true)
1591 4 : .expect("patch fragment null is legal");
1592 4 : assert!(out["description"].is_null());
1593 4 : }
1594 :
1595 : /// 4.3.6.3: "the registration shall define both: an entity id (i.e. an id
1596 : /// pattern or Entity type defining a group of entities is not supported
1597 : /// for exclusive registrations) [and] Attributes."
1598 : #[tokio::test]
1599 4 : async fn exclusive_registration_requires_entity_id_and_attributes() {
1600 4 : let ctx = Loader::new().core();
1601 24 : let mk = |mode: &str, info: Value| {
1602 24 : json!({
1603 24 : "id": "urn:ngsi-ld:ContextSourceRegistration:x1",
1604 24 : "type": "ContextSourceRegistration",
1605 24 : "endpoint": "http://peer:9090",
1606 24 : "mode": mode,
1607 24 : "information": [info]
1608 : })
1609 24 : };
1610 24 : let norm = |mode: &str, info: Value| {
1611 24 : normalize_registration(mk(mode, info).as_object().unwrap(), &ctx, false)
1612 24 : };
1613 4 : let full = json!({
1614 4 : "entities": [{"id": "urn:ngsi-ld:Vehicle:v1", "type": "Vehicle"}],
1615 4 : "propertyNames": ["speed"]
1616 : });
1617 4 : assert!(norm("exclusive", full.clone()).is_ok());
1618 4 : assert!(
1619 4 : norm(
1620 4 : "exclusive",
1621 4 : json!({"entities": [{"id": "urn:ngsi-ld:Vehicle:v1", "type": "Vehicle"}]})
1622 4 : )
1623 4 : .is_err(),
1624 : "exclusive without Attributes"
1625 : );
1626 4 : assert!(
1627 4 : norm(
1628 4 : "exclusive",
1629 4 : json!({"entities": [{"type": "Vehicle"}], "propertyNames": ["speed"]})
1630 4 : )
1631 4 : .is_err(),
1632 : "exclusive with a type-only entity group"
1633 : );
1634 4 : assert!(
1635 4 : norm(
1636 4 : "exclusive",
1637 4 : json!({"entities": [{"idPattern": ".*", "type": "Vehicle"}],
1638 4 : "propertyNames": ["speed"]})
1639 4 : )
1640 4 : .is_err(),
1641 : "exclusive with an id pattern"
1642 : );
1643 4 : assert!(
1644 4 : norm("redirect", json!({"entities": [{"type": "Vehicle"}]})).is_ok(),
1645 : "redirect may register a whole type without attributes (4.3.6.3)"
1646 : );
1647 4 : assert!(
1648 4 : norm("sideways", full).is_err(),
1649 4 : "mode outside the 5.2.9 enum"
1650 4 : );
1651 4 : }
1652 :
1653 : /// 4.3.6.3: "Once an exclusive Context Source Registration has been
1654 : /// created, no further exclusive or redirect Context Source Registrations
1655 : /// can be created for that same combination of Entity ID and Attributes"
1656 : /// — while redirect × redirect overlap stays legal.
1657 : #[tokio::test]
1658 4 : async fn proxied_overlap_with_an_exclusive_registration_conflicts() {
1659 4 : let st = crate::state::AppState::new("me".into());
1660 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1661 4 : let ctx = st.loader.core();
1662 28 : let mk = |id: &str, mode: &str, attr: &str| {
1663 28 : let doc = json!({
1664 28 : "id": id,
1665 28 : "type": "ContextSourceRegistration",
1666 28 : "endpoint": "http://peer:9090",
1667 28 : "mode": mode,
1668 28 : "information": [{
1669 28 : "entities": [{"id": "urn:ngsi-ld:Vehicle:v1", "type": "Vehicle"}],
1670 28 : "propertyNames": [attr]
1671 : }]
1672 : });
1673 28 : normalize_registration(doc.as_object().unwrap(), &ctx, false).expect("valid reg")
1674 28 : };
1675 4 : let seeded = mk(
1676 4 : "urn:ngsi-ld:ContextSourceRegistration:e1",
1677 4 : "exclusive",
1678 4 : "speed",
1679 4 : );
1680 4 : st.store
1681 4 : .create(
1682 4 : &tenant,
1683 4 : Kind::Registration,
1684 4 : "urn:ngsi-ld:ContextSourceRegistration:e1",
1685 4 : Value::Object(seeded),
1686 4 : )
1687 4 : .await
1688 4 : .expect("seed");
1689 4 : let overlap_exc = mk(
1690 4 : "urn:ngsi-ld:ContextSourceRegistration:e2",
1691 4 : "exclusive",
1692 4 : "speed",
1693 4 : );
1694 4 : assert!(
1695 4 : check_proxied_overlap(&st, &tenant, &overlap_exc, None, &ctx)
1696 4 : .await
1697 4 : .is_err(),
1698 : "second exclusive for the same (id, attr)"
1699 : );
1700 4 : let overlap_red = mk(
1701 4 : "urn:ngsi-ld:ContextSourceRegistration:r1",
1702 4 : "redirect",
1703 4 : "speed",
1704 4 : );
1705 4 : assert!(
1706 4 : check_proxied_overlap(&st, &tenant, &overlap_red, None, &ctx)
1707 4 : .await
1708 4 : .is_err(),
1709 : "redirect after an exclusive for the same combination"
1710 : );
1711 4 : let other_attr = mk(
1712 4 : "urn:ngsi-ld:ContextSourceRegistration:e3",
1713 4 : "exclusive",
1714 4 : "color",
1715 4 : );
1716 4 : assert!(
1717 4 : check_proxied_overlap(&st, &tenant, &other_attr, None, &ctx)
1718 4 : .await
1719 4 : .is_ok(),
1720 : "disjoint attribute is a different combination"
1721 : );
1722 : // the registration itself is not its own conflict (update path)
1723 4 : let self_doc = mk(
1724 4 : "urn:ngsi-ld:ContextSourceRegistration:e1",
1725 4 : "exclusive",
1726 4 : "speed",
1727 4 : );
1728 4 : assert!(check_proxied_overlap(
1729 4 : &st,
1730 4 : &tenant,
1731 4 : &self_doc,
1732 4 : Some("urn:ngsi-ld:ContextSourceRegistration:e1"),
1733 4 : &ctx
1734 : )
1735 4 : .await
1736 4 : .is_ok());
1737 : // redirect × redirect overlap is explicitly legal
1738 4 : let r2 = mk(
1739 4 : "urn:ngsi-ld:ContextSourceRegistration:r2",
1740 4 : "redirect",
1741 4 : "color",
1742 4 : );
1743 4 : st.store
1744 4 : .create(
1745 4 : &tenant,
1746 4 : Kind::Registration,
1747 4 : "urn:ngsi-ld:ContextSourceRegistration:r2",
1748 4 : Value::Object(r2),
1749 4 : )
1750 4 : .await
1751 4 : .expect("seed redirect");
1752 4 : let r3 = mk(
1753 4 : "urn:ngsi-ld:ContextSourceRegistration:r3",
1754 4 : "redirect",
1755 4 : "color",
1756 4 : );
1757 4 : assert!(check_proxied_overlap(&st, &tenant, &r3, None, &ctx)
1758 4 : .await
1759 4 : .is_ok());
1760 4 : }
1761 :
1762 : /// 5.2.8 Table 5.2.8-1 — an EntityInfo `type` is "String or String[]" —
1763 : /// applied to the 5.9.2.4 redirect rule: only an entity that matches the
1764 : /// registered Entity type conflicts, whichever spelling was registered.
1765 : #[tokio::test]
1766 4 : async fn clause_5_9_2_4_redirect_conflict_honours_the_array_form_entity_type() {
1767 4 : let st = crate::state::AppState::new("me".into());
1768 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1769 4 : let ctx = st.loader.core();
1770 4 : let mut building = Map::new();
1771 4 : building.insert("id".into(), json!("urn:ngsi-ld:Building:b1"));
1772 4 : building.insert("type".into(), json!([ctx.expand_key("Building")]));
1773 4 : st.store
1774 4 : .create(
1775 4 : &tenant,
1776 4 : Kind::Entity,
1777 4 : "urn:ngsi-ld:Building:b1",
1778 4 : Value::Object(building),
1779 4 : )
1780 4 : .await
1781 4 : .expect("seed building");
1782 16 : let reg = |ty: Value| {
1783 16 : let doc = json!({
1784 16 : "id": "urn:ngsi-ld:ContextSourceRegistration:red1",
1785 16 : "type": "ContextSourceRegistration",
1786 16 : "endpoint": "http://peer:9090",
1787 16 : "mode": "redirect",
1788 16 : "information": [{"entities": [{"type": ty}]}]
1789 : });
1790 16 : normalize_registration(doc.as_object().expect("object"), &ctx, false).expect("valid")
1791 16 : };
1792 16 : async fn err(st: &AppState, tenant: &TenantId, doc: Map<String, Value>) -> Option<String> {
1793 16 : match check_entity_conflict(st, tenant, &doc).await {
1794 8 : Err(NgsiError::Conflict(m)) => Some(m),
1795 0 : Err(other) => panic!("unexpected error {other:?}"),
1796 8 : Ok(()) => None,
1797 : }
1798 16 : }
1799 4 : assert_eq!(
1800 4 : err(&st, &tenant, reg(json!(["Vehicle"]))).await,
1801 : None,
1802 : "a redirect for Vehicle must not conflict with a Building"
1803 : );
1804 4 : assert_eq!(
1805 4 : err(&st, &tenant, reg(json!("Vehicle"))).await,
1806 : None,
1807 : "same, in the string spelling"
1808 : );
1809 4 : let hit = err(&st, &tenant, reg(json!(["Building"])))
1810 4 : .await
1811 4 : .expect("the Building conflicts");
1812 4 : assert!(
1813 4 : hit.contains("urn:ngsi-ld:Building:b1"),
1814 : "the conflict names the existing entity: {hit}"
1815 : );
1816 4 : assert!(
1817 4 : err(&st, &tenant, reg(json!("Building"))).await.is_some(),
1818 4 : "same, string spelling"
1819 4 : );
1820 4 : }
1821 :
1822 : /// 5.9.2.4 redirect: "If an existing Entity already matches the
1823 : /// `Context Source Registration`, an error of type Conflict shall be
1824 : /// raised." An EntityInfo may identify its Entities by `idPattern`
1825 : /// alone (5.2.8) — a predicate no store can decide — and a
1826 : /// RegistrationInfo may carry several EntityInfo entries, so each
1827 : /// Entity read is asked about all of them.
1828 : #[tokio::test]
1829 4 : async fn clause_5_9_2_4_redirect_conflict_matches_an_id_pattern() {
1830 4 : let st = crate::state::AppState::new("me".into());
1831 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1832 4 : let ctx = st.loader.core();
1833 8 : for (id, ty) in [
1834 4 : ("urn:ngsi-ld:Vehicle:v1", "Vehicle"),
1835 4 : ("urn:ngsi-ld:Device:d1", "Device"),
1836 4 : ] {
1837 8 : let mut e = Map::new();
1838 8 : e.insert("id".into(), json!(id));
1839 8 : e.insert("type".into(), json!([ctx.expand_key(ty)]));
1840 8 : st.store
1841 8 : .create(&tenant, Kind::Entity, id, Value::Object(e))
1842 8 : .await
1843 8 : .expect("seed");
1844 : }
1845 16 : let pat = |ents: Value| {
1846 16 : let doc = json!({
1847 16 : "id": "urn:ngsi-ld:ContextSourceRegistration:pat1",
1848 16 : "type": "ContextSourceRegistration",
1849 16 : "endpoint": "http://peer:9090",
1850 16 : "mode": "redirect",
1851 16 : "information": [{"entities": ents}]
1852 : });
1853 16 : normalize_registration(doc.as_object().expect("object"), &ctx, false).expect("valid")
1854 16 : };
1855 16 : async fn err(st: &AppState, tenant: &TenantId, doc: Map<String, Value>) -> Option<String> {
1856 16 : match check_entity_conflict(st, tenant, &doc).await {
1857 8 : Err(NgsiError::Conflict(m)) => Some(m),
1858 0 : Err(other) => panic!("unexpected error {other:?}"),
1859 8 : Ok(()) => None,
1860 : }
1861 16 : }
1862 4 : assert_eq!(
1863 4 : err(
1864 4 : &st,
1865 4 : &tenant,
1866 4 : pat(json!([{"idPattern": "^urn:ngsi-ld:Nothing:.*"}]))
1867 4 : )
1868 4 : .await,
1869 : None,
1870 : "a pattern no entity matches does not conflict"
1871 : );
1872 4 : let hit = err(
1873 4 : &st,
1874 4 : &tenant,
1875 4 : pat(json!([{"idPattern": "^urn:ngsi-ld:Vehicle:.*"}])),
1876 4 : )
1877 4 : .await
1878 4 : .expect("the Vehicle matches the pattern");
1879 4 : assert!(hit.contains("urn:ngsi-ld:Vehicle:v1"), "{hit}");
1880 4 : assert_eq!(
1881 4 : err(
1882 4 : &st,
1883 4 : &tenant,
1884 4 : pat(json!([{"idPattern": "^urn:ngsi-ld:Vehicle:.*", "type": "Device"}]))
1885 4 : )
1886 4 : .await,
1887 : None,
1888 : "the pattern and the type must both hold: the Vehicle is not a Device"
1889 : );
1890 : // The second EntityInfo is the one that matches: every selector of
1891 : // the RegistrationInfo is asked about every Entity, not only the first.
1892 4 : let hit = err(
1893 4 : &st,
1894 4 : &tenant,
1895 4 : pat(json!([
1896 4 : {"idPattern": "^urn:ngsi-ld:Nothing:.*"},
1897 4 : {"idPattern": "^urn:ngsi-ld:Device:.*"}
1898 4 : ])),
1899 4 : )
1900 4 : .await
1901 4 : .expect("the Device matches the second EntityInfo");
1902 4 : assert!(hit.contains("urn:ngsi-ld:Device:d1"), "{hit}");
1903 4 : }
1904 :
1905 : /// The Entities of the tenant are read a page at a time (a whole-tenant
1906 : /// read is refused above the store's row ceiling, 5.5.6, and this check
1907 : /// has no TooManyResults to raise), so a conflict that sits beyond the
1908 : /// first page must still be found.
1909 : #[tokio::test]
1910 4 : async fn clause_5_9_2_4_a_conflict_past_the_first_page_is_found() {
1911 4 : let st = crate::state::AppState::new("me".into());
1912 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1913 4 : let ctx = st.loader.core();
1914 4 : let device = ctx.expand_key("Device");
1915 4000 : for i in 0..SCAN_PAGE {
1916 4000 : let id = format!("urn:ngsi-ld:Device:{i:06}");
1917 4000 : let mut e = Map::new();
1918 4000 : e.insert("id".into(), json!(id));
1919 4000 : e.insert("type".into(), json!([device]));
1920 4000 : st.store
1921 4000 : .create(&tenant, Kind::Entity, &id, Value::Object(e))
1922 4000 : .await
1923 4000 : .expect("seed");
1924 : }
1925 : // sorts after every Device above, so it is only reached on page two
1926 4 : let mut zebra = Map::new();
1927 4 : zebra.insert("id".into(), json!("urn:ngsi-ld:Zebra:z1"));
1928 4 : zebra.insert("type".into(), json!([ctx.expand_key("Zebra")]));
1929 4 : st.store
1930 4 : .create(
1931 4 : &tenant,
1932 4 : Kind::Entity,
1933 4 : "urn:ngsi-ld:Zebra:z1",
1934 4 : Value::Object(zebra),
1935 4 : )
1936 4 : .await
1937 4 : .expect("seed");
1938 4 : let doc = json!({
1939 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:pat2",
1940 4 : "type": "ContextSourceRegistration",
1941 4 : "endpoint": "http://peer:9090",
1942 4 : "mode": "redirect",
1943 4 : "information": [{"entities": [{"idPattern": "^urn:ngsi-ld:Zebra:.*"}]}]
1944 : });
1945 4 : let norm =
1946 4 : normalize_registration(doc.as_object().expect("object"), &ctx, false).expect("valid");
1947 4 : match check_entity_conflict(&st, &tenant, &norm).await {
1948 4 : Err(NgsiError::Conflict(m)) => assert!(m.contains("urn:ngsi-ld:Zebra:z1"), "{m}"),
1949 4 : other => panic!("the conflict on page two was missed: {other:?}"),
1950 4 : }
1951 4 : }
1952 :
1953 : /// 5.9.2.4: an exclusive registration conflicts with an existing Entity
1954 : /// only when "the existing Entity contains any of the Attributes defined
1955 : /// in the registration".
1956 : #[tokio::test]
1957 4 : async fn clause_5_9_2_4_exclusive_conflict_needs_a_registered_attribute() {
1958 4 : let st = crate::state::AppState::new("me".into());
1959 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1960 4 : let ctx = st.loader.core();
1961 4 : let mut vehicle = Map::new();
1962 4 : vehicle.insert("id".into(), json!("urn:ngsi-ld:Vehicle:v1"));
1963 4 : vehicle.insert("type".into(), json!([ctx.expand_key("Vehicle")]));
1964 4 : vehicle.insert(ctx.expand_key("color"), json!([{"type": "Property"}]));
1965 4 : st.store
1966 4 : .create(
1967 4 : &tenant,
1968 4 : Kind::Entity,
1969 4 : "urn:ngsi-ld:Vehicle:v1",
1970 4 : Value::Object(vehicle),
1971 4 : )
1972 4 : .await
1973 4 : .expect("seed vehicle");
1974 12 : let reg = |attr: &str, id: &str| {
1975 12 : let doc = json!({
1976 12 : "id": "urn:ngsi-ld:ContextSourceRegistration:exc1",
1977 12 : "type": "ContextSourceRegistration",
1978 12 : "endpoint": "http://peer:9090",
1979 12 : "mode": "exclusive",
1980 12 : "information": [{
1981 12 : "entities": [{"id": id, "type": "Vehicle"}],
1982 12 : "propertyNames": [attr]
1983 : }]
1984 : });
1985 12 : normalize_registration(doc.as_object().expect("object"), &ctx, false).expect("valid")
1986 12 : };
1987 4 : assert!(
1988 4 : check_entity_conflict(&st, &tenant, ®("speed", "urn:ngsi-ld:Vehicle:v1"))
1989 4 : .await
1990 4 : .is_ok(),
1991 : "the entity carries no speed Attribute"
1992 : );
1993 4 : assert!(
1994 4 : check_entity_conflict(&st, &tenant, ®("color", "urn:ngsi-ld:Vehicle:v2"))
1995 4 : .await
1996 4 : .is_ok(),
1997 : "another entity id is not this entity"
1998 : );
1999 4 : assert!(
2000 4 : check_entity_conflict(&st, &tenant, ®("color", "urn:ngsi-ld:Vehicle:v1"))
2001 4 : .await
2002 4 : .is_err(),
2003 4 : "the entity carries the registered color Attribute"
2004 4 : );
2005 4 : }
2006 :
2007 : /// 6.8.3.2 Table 6.8.3.2-1: the `type` parameter of Query Context Source
2008 : /// Registrations is "Selection of Entity Types as per clause 4.17" — a
2009 : /// selection expression, not a comma-separated list of terms.
2010 : #[tokio::test]
2011 4 : async fn clause_4_17_type_selection_queries_registrations() {
2012 4 : let st = crate::state::AppState::new("me".into());
2013 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2014 4 : let ctx = st.loader.core();
2015 8 : let reg = |id: &str, ty: Value| {
2016 8 : let doc = json!({
2017 8 : "id": id,
2018 8 : "type": "ContextSourceRegistration",
2019 8 : "endpoint": "http://peer:9090",
2020 8 : "information": [{"entities": [{"type": ty}]}]
2021 : });
2022 8 : normalize_registration(doc.as_object().expect("object"), &ctx, false).expect("ok")
2023 8 : };
2024 8 : async fn seed(st: &AppState, tenant: &TenantId, id: &str, doc: Map<String, Value>) {
2025 8 : st.store
2026 8 : .create(tenant, Kind::Registration, id, Value::Object(doc))
2027 8 : .await
2028 8 : .expect("seed");
2029 8 : }
2030 4 : let both = "urn:ngsi-ld:ContextSourceRegistration:both";
2031 4 : let home = "urn:ngsi-ld:ContextSourceRegistration:home";
2032 4 : seed(&st, &tenant, both, reg(both, json!(["Home", "Vehicle"]))).await;
2033 4 : seed(&st, &tenant, home, reg(home, json!("Home"))).await;
2034 12 : let ids = |sel: &str| {
2035 12 : let st = st.clone();
2036 12 : let sel = sel.to_owned();
2037 12 : async move {
2038 12 : let params = HashMap::from([("type".to_owned(), sel)]);
2039 12 : let resp =
2040 12 : query_registrations(State(st), CleanParams(params), HeaderMap::new()).await;
2041 12 : assert_eq!(resp.status(), StatusCode::OK);
2042 12 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
2043 12 : .await
2044 12 : .expect("body");
2045 12 : let body: Value = serde_json::from_slice(&bytes).expect("json list");
2046 12 : body.as_array()
2047 12 : .expect("array")
2048 12 : .iter()
2049 12 : .filter_map(|r| r.get("id").and_then(Value::as_str).map(str::to_owned))
2050 12 : .collect::<Vec<String>>()
2051 12 : }
2052 12 : };
2053 4 : let conj = ids("(Home;Vehicle)").await;
2054 4 : assert!(
2055 4 : conj.contains(&"urn:ngsi-ld:ContextSourceRegistration:both".to_owned()),
2056 : "a registration declaring both types matches the conjunction: {conj:?}"
2057 : );
2058 4 : assert!(
2059 4 : !conj.contains(&"urn:ngsi-ld:ContextSourceRegistration:home".to_owned()),
2060 : "a registration declaring only Home must not match: {conj:?}"
2061 : );
2062 4 : let alt = ids("Vehicle,Home").await;
2063 4 : assert_eq!(alt.len(), 2, "a comma list is a disjunction: {alt:?}");
2064 4 : let none = ids("Parking").await;
2065 4 : assert!(
2066 4 : none.is_empty(),
2067 4 : "no registration declares Parking: {none:?}"
2068 4 : );
2069 4 : }
2070 :
2071 : /// 5.9.2.4: "If expiresAt is a date and time in the past, an error of
2072 : /// type BadRequestData shall be raised" — the comparison is over the
2073 : /// instant, not over the spelling, so an expiresAt written with fewer
2074 : /// sub-second digits than the server's own timestamp is still past.
2075 : #[test]
2076 4 : fn clause_5_9_2_4_past_expires_at_whatever_the_fraction_spelling() {
2077 4 : let ctx = Loader::new().core();
2078 4 : for _ in 0..8 {
2079 32 : let now = now_iso();
2080 32 : if now.len() < 24 || &now[20..23] == "000" {
2081 0 : continue; // no sub-second gap to compare against
2082 32 : }
2083 32 : let past = format!("{}Z", &now[..19]); // same second, no fraction
2084 32 : let doc = json!({
2085 32 : "id": "urn:ngsi-ld:ContextSourceRegistration:exp1",
2086 32 : "type": "ContextSourceRegistration",
2087 32 : "endpoint": "http://peer:9090",
2088 32 : "expiresAt": past,
2089 32 : "information": [{"entities": [{"type": "Vehicle"}]}]
2090 : });
2091 32 : assert!(
2092 32 : normalize_registration(doc.as_object().expect("object"), &ctx, false).is_err(),
2093 : "expiresAt {past} is in the past of {now}"
2094 : );
2095 : }
2096 4 : }
2097 :
2098 : /// The registration body's cardinality caps are the only bound on the
2099 : /// index rows one create materialises: the last accepted size and the
2100 : /// first rejected one both have to hold.
2101 : #[test]
2102 4 : fn registration_cardinality_caps_hold_at_the_edge() {
2103 4 : let ctx = Loader::new().core();
2104 1036 : let info = |n: usize| {
2105 1036 : json!({"entities": [{"type": "Vehicle"}], "propertyNames":
2106 2056 : (0..n).map(|i| Value::String(format!("p{i}"))).collect::<Vec<_>>()})
2107 1036 : };
2108 32 : let mk = |information: Value| {
2109 32 : let doc = json!({
2110 32 : "id": "urn:ngsi-ld:ContextSourceRegistration:cap1",
2111 32 : "type": "ContextSourceRegistration",
2112 32 : "endpoint": "http://peer:9090",
2113 32 : "information": information
2114 : });
2115 32 : normalize_registration(doc.as_object().expect("object"), &ctx, false)
2116 32 : };
2117 1028 : let many = |n: usize| Value::Array((0..n).map(|_| info(1)).collect());
2118 4 : assert!(mk(many(MAX_INFORMATION)).is_ok(), "the cap itself is legal");
2119 4 : let over = mk(many(MAX_INFORMATION + 1)).expect_err("over the cap");
2120 4 : assert!(matches!(over, NgsiError::BadRequestData(_)), "{over:?}");
2121 : // the status is what the client sees, and 403 would tell it to
2122 : // narrow a query it never sent
2123 4 : assert_eq!(over.status(), 400);
2124 4 : assert!(mk(json!([info(MAX_INFO_MEMBERS)])).is_ok());
2125 4 : assert!(matches!(
2126 4 : mk(json!([info(MAX_INFO_MEMBERS + 1)])),
2127 : Err(NgsiError::BadRequestData(_))
2128 : ));
2129 8 : let entities = |n: usize| {
2130 8 : json!([{"entities": (0..n)
2131 1028 : .map(|i| json!({"id": format!("urn:ngsi-ld:Vehicle:v{i}"), "type": "Vehicle"}))
2132 8 : .collect::<Vec<_>>()}])
2133 8 : };
2134 4 : assert!(mk(entities(MAX_INFO_MEMBERS)).is_ok());
2135 4 : assert!(matches!(
2136 4 : mk(entities(MAX_INFO_MEMBERS + 1)),
2137 : Err(NgsiError::BadRequestData(_))
2138 : ));
2139 8 : let rels = |n: usize| {
2140 8 : json!([{"entities": [{"type": "Vehicle"}], "relationshipNames":
2141 1028 : (0..n).map(|i| Value::String(format!("r{i}"))).collect::<Vec<_>>()}])
2142 8 : };
2143 4 : assert!(mk(rels(MAX_INFO_MEMBERS)).is_ok());
2144 4 : assert!(matches!(
2145 4 : mk(rels(MAX_INFO_MEMBERS + 1)),
2146 : Err(NgsiError::BadRequestData(_))
2147 : ));
2148 4 : }
2149 :
2150 : /// 4.3.6.6: the four processed contextSourceInfo keys have
2151 : /// constrained value spaces, checked at registration time.
2152 : #[test]
2153 4 : fn context_source_info_reserved_keys_are_validated() {
2154 4 : let ctx = Loader::new().core();
2155 36 : let mk = |key: &str, value: &str| {
2156 36 : json!({
2157 36 : "id": "urn:ngsi-ld:ContextSourceRegistration:csi1",
2158 36 : "type": "ContextSourceRegistration",
2159 36 : "endpoint": "http://peer:9090",
2160 36 : "information": [{"entities": [{"type": "Building"}]}],
2161 36 : "contextSourceInfo": [{"key": key, "value": value}]
2162 : })
2163 36 : };
2164 36 : let ok = |key: &str, value: &str| {
2165 36 : normalize_registration(mk(key, value).as_object().unwrap(), &ctx, false).is_ok()
2166 36 : };
2167 4 : assert!(ok("accept", "application/json"));
2168 4 : assert!(ok("contentType", "application/ld+json"));
2169 4 : assert!(!ok("accept", "text/html"), "MIME outside 4.3.6.6's list");
2170 4 : assert!(!ok("contentType", "application/geo+json"));
2171 4 : assert!(ok("jsonldContext", "https://example.org/ctx.jsonld"));
2172 4 : assert!(!ok("jsonldContext", "not a url"));
2173 4 : assert!(ok("ngsildConformance", "1.6"));
2174 4 : assert!(!ok("ngsildConformance", "latest"));
2175 : // ordinary custom keys stay free-form
2176 4 : assert!(ok("Authorization", "Bearer abc"));
2177 4 : }
2178 :
2179 : /// Table 5.2.9-1 refreshRate and Table 5.2.34-1 cacheDuration are ISO 8601
2180 : /// durations. The grammar is `P[nY][nM][nW][nD][T[nH][nM][nS]]`: every
2181 : /// component is a number, the components keep their order, and the time
2182 : /// units live after the `T` — a value that is none of those is not a
2183 : /// duration, whatever it looks like.
2184 : #[test]
2185 4 : fn table_5_2_9_1_refresh_rate_takes_an_iso_8601_duration() {
2186 48 : for good in [
2187 4 : "P1Y",
2188 4 : "P1M",
2189 4 : "P1W",
2190 4 : "P1D",
2191 4 : "PT1H",
2192 4 : "PT1M",
2193 4 : "PT30S",
2194 4 : "P1Y2M3DT4H5M6S",
2195 4 : "PT0.5S",
2196 4 : "PT0,5S",
2197 4 : "P1Y2M",
2198 4 : "P1DT12H",
2199 4 : ] {
2200 48 : assert!(valid_iso8601_duration(good), "{good} is a duration");
2201 : }
2202 76 : for bad in [
2203 4 : "", "P", "PT", "1Y", "P1", "P1X", "p1d", // the designators are upper case
2204 4 : "P1H", // an hour is a time component and belongs after the T
2205 4 : "PT1D", // a day is a date component and belongs before it
2206 4 : "P1D2M", // out of order: months precede days
2207 4 : "P,D", "P.D", "P..D", // separators are not a number
2208 4 : "P-1D", "P+1D", "P 1D", "PT1S1S", "P1DT", "1PD",
2209 4 : ] {
2210 76 : assert!(!valid_iso8601_duration(bad), "{bad:?} is not a duration");
2211 : }
2212 4 : }
2213 :
2214 : /// 5.5.6 + 5.10.2.4: the registration query filters before it pages, so
2215 : /// every match is held at once. A tenant at the 100 000-registration
2216 : /// target answers one `type=` selector with every one of them, which is
2217 : /// the "so many results that can potentially exhaust … server resources"
2218 : /// 5.5.6 names — the query is refused at the ceiling instead of building
2219 : /// the answer.
2220 : #[tokio::test]
2221 4 : async fn clause_5_5_6_a_registration_query_stops_at_the_fold_ceiling() {
2222 4 : let st = AppState::new("antares-csr-ceiling".into());
2223 4 : let tenant = TenantId::default();
2224 20 : for i in 0..5 {
2225 20 : let id = format!("urn:ngsi-ld:ContextSourceRegistration:ceil{i}");
2226 20 : st.store
2227 20 : .create(
2228 20 : &tenant,
2229 20 : Kind::Registration,
2230 20 : &id,
2231 20 : json!({
2232 20 : "id": id,
2233 20 : "type": "ContextSourceRegistration",
2234 20 : "endpoint": "http://peer:9090",
2235 20 : "information": [{"entities": [{"type": "Building"}]}],
2236 20 : }),
2237 20 : )
2238 20 : .await
2239 20 : .expect("store the registration");
2240 : }
2241 4 : let all = collect_matching(&st, &tenant, |_| true, 100)
2242 4 : .await
2243 4 : .expect("under the ceiling");
2244 4 : assert_eq!(all.len(), 5, "every registration matches");
2245 4 : let err = collect_matching(&st, &tenant, |_| true, 2)
2246 4 : .await
2247 4 : .expect_err("a match set over the ceiling is refused");
2248 4 : assert!(
2249 4 : matches!(err, NgsiError::TooManyResults(_)),
2250 : "5.5.6 names TooManyResults, got {err:?}"
2251 : );
2252 : // the ceiling counts MATCHES, not documents walked: a narrow query
2253 : // over the same tenant still answers
2254 4 : let narrow = collect_matching(
2255 4 : &st,
2256 4 : &tenant,
2257 20 : |d| d["id"] == json!("urn:ngsi-ld:ContextSourceRegistration:ceil3"),
2258 : 2,
2259 : )
2260 4 : .await
2261 4 : .expect("one match is under any ceiling");
2262 4 : assert_eq!(narrow.len(), 1);
2263 4 : }
2264 :
2265 : /// 6.3.19: "Key and value members shall adhere to IETF RFC 7230 …
2266 : /// definitions concerning HTTP headers". A pair that is not a header is
2267 : /// refused where 5.9.2.4 refuses registration content, not carried until
2268 : /// the first forward — where the request cannot be built at all, and the
2269 : /// registration that caused it is nowhere in the failure.
2270 : #[test]
2271 4 : fn clause_6_3_19_a_pair_that_is_not_a_header_is_refused() {
2272 4 : let ctx = Loader::new().core();
2273 52 : let mk = |key: &str, value: &str| {
2274 52 : json!({
2275 52 : "id": "urn:ngsi-ld:ContextSourceRegistration:csi2",
2276 52 : "type": "ContextSourceRegistration",
2277 52 : "endpoint": "http://peer:9090",
2278 52 : "information": [{"entities": [{"type": "Building"}]}],
2279 52 : "contextSourceInfo": [{"key": key, "value": value}]
2280 : })
2281 52 : };
2282 52 : let ok = |key: &str, value: &str| {
2283 52 : normalize_registration(mk(key, value).as_object().expect("object"), &ctx, false).is_ok()
2284 52 : };
2285 : // RFC 7230 field-name is a token: no separators, no space, no CTL
2286 24 : for key in [
2287 4 : "X-Injected\r\nX-Second",
2288 4 : "X Injected",
2289 4 : "X:Injected",
2290 4 : "",
2291 4 : "X-Inj\u{0000}ected",
2292 4 : "Über-Header",
2293 4 : ] {
2294 24 : assert!(!ok(key, "value"), "{key:?} is not an RFC 7230 field-name");
2295 : }
2296 : // RFC 7230 field-value carries no CR, LF or NUL
2297 16 : for value in ["a\r\nX-Injected: 1", "a\nb", "a\rb", "a\u{0000}b"] {
2298 16 : assert!(
2299 16 : !ok("X-Custom", value),
2300 : "{value:?} is not an RFC 7230 field-value"
2301 : );
2302 : }
2303 : // and the shapes a real registration uses stay accepted
2304 4 : assert!(ok("X-Custom", "urn:ngsi-ld:request"));
2305 4 : assert!(ok("X-Api-Key", "a b c"));
2306 4 : assert!(ok("Authorization", "Bearer abc.def-ghi_jkl"));
2307 4 : }
2308 : }
2309 :
2310 : #[cfg(test)]
2311 : mod concurrent_create_5_9_2_4 {
2312 : use super::*;
2313 : use std::collections::HashMap;
2314 :
2315 : /// 5.9.2.4 decides a conflict against the registrations "already"
2316 : /// present in one tenant, and two tenants' sets are disjoint: a write
2317 : /// in one tenant never waits for the check-then-write section of
2318 : /// another, while a second write in the same tenant does.
2319 : #[tokio::test]
2320 4 : async fn the_write_section_of_one_tenant_does_not_hold_another() {
2321 : use std::time::Duration;
2322 : use tokio::time::timeout;
2323 4 : let a = TenantId::new("locka").expect("tenant");
2324 4 : let b = TenantId::new("lockb").expect("tenant");
2325 4 : let held = registration_write_lock(&a).await;
2326 4 : let other = timeout(Duration::from_millis(200), registration_write_lock(&b)).await;
2327 4 : assert!(
2328 4 : other.is_ok(),
2329 : "tenant b waited behind tenant a's registration write"
2330 : );
2331 4 : let same = timeout(Duration::from_millis(200), registration_write_lock(&a)).await;
2332 4 : assert!(same.is_err(), "a second write in the same tenant must wait");
2333 4 : drop(held);
2334 4 : let again = timeout(Duration::from_millis(200), registration_write_lock(&a)).await;
2335 4 : assert!(again.is_ok(), "the section is free once its holder is done");
2336 4 : }
2337 :
2338 1200 : fn exclusive_body(n: usize) -> Bytes {
2339 1200 : Bytes::from(format!(
2340 : r#"{{"id":"urn:ngsi-ld:ContextSourceRegistration:race{n}",
2341 : "type":"ContextSourceRegistration",
2342 : "endpoint":"http://peer:9090",
2343 : "mode":"exclusive",
2344 : "information":[{{"entities":[{{"id":"urn:ngsi-ld:Vehicle:race",
2345 : "type":"Vehicle"}}],
2346 : "propertyNames":["speed"]}}]}}"#
2347 : ))
2348 1200 : }
2349 :
2350 1600 : fn json_headers() -> HeaderMap {
2351 1600 : let mut h = HeaderMap::new();
2352 1600 : h.insert(
2353 1600 : axum::http::header::CONTENT_TYPE,
2354 1600 : "application/json".parse().expect("header value"),
2355 : );
2356 1600 : h
2357 1600 : }
2358 :
2359 : /// 5.9.2.4: "If an exclusive or redirect Context Source Registration
2360 : /// already matches against the Entity ID (URI) and any of the Attributes
2361 : /// defined in the registration, an error of type Conflict shall be
2362 : /// raised." Requests racing each other must not both slip past that
2363 : /// check: whatever the interleaving, one registration is stored and
2364 : /// every other request gets the Conflict.
2365 : #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
2366 4 : async fn concurrent_exclusive_creates_store_exactly_one_registration() {
2367 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2368 100 : for round in 0..25 {
2369 100 : let st = crate::state::AppState::new("me".into());
2370 100 : let gate = std::sync::Arc::new(tokio::sync::Barrier::new(8));
2371 100 : let mut tasks = Vec::new();
2372 800 : for n in 0..8 {
2373 800 : let (st, gate) = (st.clone(), gate.clone());
2374 800 : tasks.push(tokio::spawn(async move {
2375 800 : gate.wait().await;
2376 800 : create_registration(
2377 800 : State(st),
2378 800 : CleanParams(HashMap::new()),
2379 800 : json_headers(),
2380 800 : exclusive_body(n),
2381 4 : )
2382 800 : .await
2383 800 : .status()
2384 800 : }));
2385 4 : }
2386 100 : let mut created = 0;
2387 800 : for t in tasks {
2388 800 : match t.await.expect("task") {
2389 800 : StatusCode::CREATED => created += 1,
2390 700 : StatusCode::CONFLICT => {}
2391 4 : other => panic!("round {round}: unexpected status {other}"),
2392 4 : }
2393 4 : }
2394 100 : let stored = st
2395 100 : .store
2396 100 : .list(&tenant, Kind::Registration)
2397 100 : .await
2398 100 : .expect("registrations");
2399 100 : assert_eq!(
2400 100 : stored.len(),
2401 4 : 1,
2402 4 : "round {round}: 5.9.2.4 forbids a second exclusive registration \
2403 4 : for the same Entity ID and Attribute"
2404 4 : );
2405 100 : assert_eq!(created, 1, "round {round}: exactly one create may succeed");
2406 4 : }
2407 4 : }
2408 :
2409 : /// 5.9.3.4 applies the same 5.9.2.4 mode rules to the merged document,
2410 : /// so two patches that each flip an inclusive registration to exclusive
2411 : /// over one Entity ID and Attribute may not both take effect.
2412 : #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
2413 4 : async fn concurrent_patches_to_exclusive_leave_one_exclusive_registration() {
2414 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2415 100 : for round in 0..25 {
2416 100 : let st = crate::state::AppState::new("me".into());
2417 400 : for n in 0..4 {
2418 400 : let body = Bytes::from(
2419 400 : String::from_utf8_lossy(&exclusive_body(n))
2420 400 : .replace(r#""mode":"exclusive","#, ""),
2421 4 : );
2422 400 : let status = create_registration(
2423 400 : State(st.clone()),
2424 400 : CleanParams(HashMap::new()),
2425 400 : json_headers(),
2426 400 : body,
2427 4 : )
2428 400 : .await
2429 400 : .status();
2430 400 : assert_eq!(status, StatusCode::CREATED, "round {round}: seed {n}");
2431 4 : }
2432 100 : let gate = std::sync::Arc::new(tokio::sync::Barrier::new(4));
2433 100 : let mut tasks = Vec::new();
2434 400 : for n in 0..4 {
2435 400 : let (st, gate) = (st.clone(), gate.clone());
2436 400 : tasks.push(tokio::spawn(async move {
2437 400 : gate.wait().await;
2438 400 : update_registration(
2439 400 : State(st),
2440 400 : Path(format!("urn:ngsi-ld:ContextSourceRegistration:race{n}")),
2441 400 : CleanParams(HashMap::new()),
2442 400 : json_headers(),
2443 400 : Bytes::from(r#"{"mode":"exclusive"}"#),
2444 4 : )
2445 400 : .await
2446 400 : .status()
2447 400 : }));
2448 4 : }
2449 100 : let mut patched = 0;
2450 400 : for t in tasks {
2451 400 : match t.await.expect("task") {
2452 400 : StatusCode::NO_CONTENT => patched += 1,
2453 300 : StatusCode::CONFLICT => {}
2454 4 : other => panic!("round {round}: unexpected status {other}"),
2455 4 : }
2456 4 : }
2457 100 : let exclusive = st
2458 100 : .store
2459 100 : .list(&tenant, Kind::Registration)
2460 100 : .await
2461 100 : .expect("registrations")
2462 100 : .iter()
2463 400 : .filter(|r| r.get("mode").and_then(Value::as_str) == Some("exclusive"))
2464 100 : .count();
2465 100 : assert_eq!(
2466 4 : exclusive, 1,
2467 4 : "round {round}: 5.9.2.4 forbids a second exclusive registration \
2468 4 : for the same Entity ID and Attribute"
2469 4 : );
2470 100 : assert_eq!(patched, 1, "round {round}: exactly one patch may succeed");
2471 4 : }
2472 4 : }
2473 : }
|
