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