Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! /subscriptions and /csourceSubscriptions (5.8, 5.11; resources 6.10/6.11,
3 : //! 6.12/6.13). One implementation, two store kinds — both use the
4 : //! Subscription data type (5.2.12).
5 :
6 : use crate::negotiate::*;
7 : use crate::state::{now_iso, AppState};
8 : use antares_jsonld::{parse_datetime, Context};
9 : use antares_model::{NgsiError, TenantId};
10 : use antares_store::CurrentStateDriverExt;
11 : use antares_store::Kind;
12 : use axum::body::Bytes;
13 : use axum::extract::{Path, State};
14 : use axum::http::{HeaderMap, StatusCode};
15 : use axum::response::{IntoResponse, Response};
16 : use serde_json::{Map, Value};
17 : use std::collections::HashMap;
18 :
19 : use crate::negotiate::CleanParams;
20 :
21 438 : fn resource_path(kind: Kind) -> &'static str {
22 438 : match kind {
23 62 : Kind::CSourceSubscription => "csourceSubscriptions",
24 376 : _ => "subscriptions",
25 : }
26 438 : }
27 :
28 : /// Table 5.2.33-1 EntitySelector: the `entities` member of a Subscription
29 : /// (5.2.12). A selector needs a type; ids are URIs, an idPattern is a
30 : /// regular expression, and a type-selection expression (4.17) stays raw
31 : /// because it is evaluated at match time.
32 1248 : fn norm_entity_selectors(v: &Value, ctx: &Context) -> Result<Value, NgsiError> {
33 1248 : let bad = NgsiError::BadRequestData;
34 1248 : let arr = v
35 1248 : .as_array()
36 1248 : .filter(|a| !a.is_empty())
37 1248 : .ok_or_else(|| bad("entities must be a non-empty array".into()))?;
38 1240 : let mut entities = Vec::new();
39 1240 : for e in arr {
40 1240 : let eo = e
41 1240 : .as_object()
42 1240 : .ok_or_else(|| bad("entities entries must be objects".into()))?;
43 1236 : let mut ne = Map::new();
44 1246 : for (ek, ev) in eo {
45 1246 : match ek.as_str() {
46 1246 : "type" => {
47 1208 : let t = ev
48 1208 : .as_str()
49 1208 : .filter(|t| !t.is_empty())
50 1208 : .ok_or_else(|| bad("EntitySelector type is required".into()))?;
51 : // 4.17 type-selection expressions stay raw and
52 : // are evaluated at match time (046_16). Table
53 : // 5.2.33-1's "*" — "a request for all
54 : // Entities" — is not a term either: expanded,
55 : // it becomes an IRI no entity carries and the
56 : // subscription notifies nothing.
57 1204 : if t == "*" || t.contains(['|', ',', ';', '(']) {
58 8 : ne.insert("type".into(), ev.clone());
59 1196 : } else {
60 1196 : ne.insert("type".into(), Value::String(ctx.expand_key(t)));
61 1196 : }
62 : }
63 38 : "id" => {
64 : // Table 5.2.33-1: id is "String or String[]"
65 : // of valid URIs
66 24 : match ev {
67 8 : Value::String(id) => {
68 8 : antares_model::EntityId::new(id)?;
69 : }
70 16 : Value::Array(a) => {
71 32 : for i in a {
72 32 : let id = i.as_str().ok_or_else(|| {
73 8 : bad("EntitySelector id entries must be URIs (5.2.33)".into())
74 8 : })?;
75 24 : antares_model::EntityId::new(id)?;
76 : }
77 : }
78 : _ => {
79 0 : return Err(bad(
80 0 : "EntitySelector id must be a URI string or array (5.2.33)".into(),
81 0 : ))
82 : }
83 : }
84 12 : ne.insert("id".into(), ev.clone());
85 : }
86 14 : "idPattern" => {
87 14 : let p = ev
88 14 : .as_str()
89 14 : .ok_or_else(|| bad("idPattern must be a string".into()))?;
90 10 : antares_ql::regex::compile(p)
91 10 : .map_err(|_| bad(format!("invalid idPattern {p:?}")))?;
92 2 : ne.insert("idPattern".into(), ev.clone());
93 : }
94 0 : _ => {
95 0 : ne.insert(ek.clone(), ev.clone());
96 0 : }
97 : }
98 : }
99 1208 : if !ne.contains_key("type") {
100 4 : return Err(bad("EntitySelector requires type (5.2.33)".into()));
101 1204 : }
102 1204 : entities.push(Value::Object(ne));
103 : }
104 1204 : Ok(Value::Array(entities))
105 1248 : }
106 :
107 : /// Table 5.2.14.1-1 NotificationParams: the `notification` member of a
108 : /// Subscription (5.2.12). Output-only members are dropped, every other
109 : /// member is held to its value space, and the Endpoint is checked by
110 : /// `check_endpoint_params`.
111 1136 : fn norm_notification(v: &Value, ctx: &Context) -> Result<Value, NgsiError> {
112 1136 : let bad = NgsiError::BadRequestData;
113 1136 : let n = v
114 1136 : .as_object()
115 1136 : .ok_or_else(|| bad("notification must be an object (5.2.14)".into()))?;
116 1132 : let mut nn = n.clone();
117 : // 5.2.14.2: output-only members are read-only — provided
118 : // ones are ignored, never stored.
119 5660 : for k in [
120 1132 : "timesSent",
121 1132 : "timesFailed",
122 1132 : "lastNotification",
123 1132 : "lastSuccess",
124 1132 : "lastFailure",
125 5660 : ] {
126 5660 : nn.remove(k);
127 5660 : }
128 1132 : if let Some(f) = n.get("format").and_then(Value::as_str) {
129 56 : if !["normalized", "keyValues", "simplified", "concise"].contains(&f) {
130 4 : return Err(bad(format!("invalid notification format {f:?}")));
131 52 : }
132 1076 : }
133 : // Table 5.2.14.1-1 p.120: "showChanges cannot be true in case
134 : // format is keyValues" — "simplified" is the declared synonym
135 1128 : if n.get("showChanges").and_then(Value::as_bool) == Some(true)
136 16 : && matches!(
137 28 : n.get("format").and_then(Value::as_str),
138 24 : Some("keyValues") | Some("simplified")
139 : )
140 : {
141 16 : return Err(bad(
142 16 : "showChanges cannot be true when format is keyValues (5.2.14)".into(),
143 16 : ));
144 1112 : }
145 : // Table 5.2.14.1-1: join / joinLevel / sysAttrs /
146 : // showChanges value spaces.
147 1112 : if let Some(j) = n.get("join") {
148 48 : if !j
149 48 : .as_str()
150 48 : .is_some_and(|j| ["flat", "inline", "@none"].contains(&j))
151 : {
152 16 : return Err(bad(format!("invalid notification join {j:?} (5.2.14)")));
153 32 : }
154 1064 : }
155 : // Table 5.2.14.1-1: a positive integer. The depth it names is
156 : // the same Linked Entity traversal (4.5.23) a query drives, so
157 : // it carries the same ceiling — every notification of this
158 : // Subscription pays that traversal, and an unbounded level
159 : // makes one accepted Subscription an amplification lever.
160 1096 : if let Some(jl) = n.get("joinLevel") {
161 56 : let cap = crate::bounds::MAX_JOIN_LEVEL as u64;
162 56 : let ok = jl.as_u64().is_some_and(|v| (1..=cap).contains(&v));
163 56 : if !ok {
164 32 : return Err(bad(format!(
165 32 : "notification.joinLevel must be an integer in 1..={cap} (5.2.14)"
166 32 : )));
167 24 : }
168 1040 : }
169 2120 : for key in ["sysAttrs", "showChanges"] {
170 2120 : if n.get(key).is_some_and(|v| !v.is_boolean()) {
171 16 : return Err(bad(format!(
172 16 : "notification.{key} must be a boolean (5.2.14)"
173 16 : )));
174 2104 : }
175 : }
176 1048 : if let Some(attrs) = n.get("attributes").and_then(Value::as_array) {
177 : // Table 5.2.14.1-1 p.119: "Empty array (0 length) is not
178 : // allowed" — same restriction on pick and omit below
179 44 : if attrs.is_empty() {
180 8 : return Err(bad(
181 8 : "notification.attributes must not be empty (5.2.14)".into()
182 8 : ));
183 36 : }
184 36 : let mut na = Vec::new();
185 36 : for a in attrs {
186 36 : let s = a
187 36 : .as_str()
188 36 : .ok_or_else(|| bad("notification.attributes must be strings".into()))?;
189 : // "A synonym for pick, except that id, type, scope
190 : // are not allowed."
191 32 : if ["id", "type", "scope"].contains(&s) {
192 16 : return Err(bad(format!(
193 16 : "notification.attributes may not name {s:?} (5.2.14)"
194 16 : )));
195 16 : }
196 16 : na.push(Value::String(ctx.expand_key(s)));
197 : }
198 16 : nn.insert("attributes".into(), Value::Array(na));
199 1004 : }
200 2004 : for key in ["pick", "omit"] {
201 2004 : let Some(members) = n.get(key).and_then(Value::as_array) else {
202 1892 : continue;
203 : };
204 112 : if members.is_empty() {
205 16 : return Err(bad(format!(
206 16 : "notification.{key} must not be empty (5.2.14)"
207 16 : )));
208 96 : }
209 : // Table 5.2.14.1-1: each member is "a valid attribute
210 : // projection language string as per clause 4.21". The
211 : // notification path parses it again at delivery and drops
212 : // a member it cannot parse, so an unparseable `omit`
213 : // accepted here would deliver the Attribute the subscriber
214 : // asked to have removed.
215 118 : for m in members {
216 118 : let term = m
217 118 : .as_str()
218 118 : .ok_or_else(|| bad(format!("notification.{key} members are Strings (5.2.14)")))?;
219 110 : crate::repr::parse_projection(term, ctx).map_err(|_| {
220 48 : bad(format!(
221 48 : "notification.{key} member {term:?} is not an attribute \
222 48 : projection language string (4.21)"
223 48 : ))
224 48 : })?;
225 : }
226 : }
227 948 : let ep = n
228 948 : .get("endpoint")
229 948 : .and_then(Value::as_object)
230 948 : .ok_or_else(|| bad("notification.endpoint is required (5.2.14)".into()))?;
231 944 : let uri = ep
232 944 : .get("uri")
233 944 : .and_then(Value::as_str)
234 944 : .ok_or_else(|| bad("endpoint.uri is required (5.2.15)".into()))?;
235 936 : antares_model::EntityId::new(uri)
236 936 : .map_err(|_| bad(format!("endpoint.uri is not a valid URI: {uri:?}")))?;
237 :
238 1844 : let member_names = |key: &str| -> Vec<String> {
239 1844 : n.get(key)
240 1844 : .and_then(Value::as_array)
241 1844 : .map(|a| {
242 40 : a.iter()
243 40 : .filter_map(Value::as_str)
244 40 : .map(str::to_owned)
245 40 : .collect()
246 40 : })
247 1844 : .unwrap_or_default()
248 1844 : };
249 922 : let pick = member_names("pick");
250 922 : let omit = member_names("omit");
251 922 : if !pick.is_empty() && n.contains_key("attributes") {
252 4 : return Err(bad("notification.pick and attributes are exclusive".into()));
253 918 : }
254 918 : if !omit.is_empty() && n.contains_key("attributes") {
255 4 : return Err(bad("notification.omit and attributes are exclusive".into()));
256 914 : }
257 914 : if pick.iter().any(|p| omit.contains(p)) {
258 4 : return Err(bad(
259 4 : "notification.pick and omit name the same entity member".into(),
260 4 : ));
261 910 : }
262 910 : check_endpoint_params(ep)?;
263 722 : Ok(Value::Object(nn))
264 1136 : }
265 :
266 : /// Table 5.2.15-1 Endpoint: the members that decide how a notification is
267 : /// sent. `uri` is checked by the caller, which needs it first.
268 910 : fn check_endpoint_params(ep: &Map<String, Value>) -> Result<(), NgsiError> {
269 910 : let bad = NgsiError::BadRequestData;
270 : // Table 5.2.15-1: receiverInfo/notifierInfo are
271 : // KeyValuePair[] — per Table 5.2.22-1 both key and value
272 : // are Strings, cardinality 1.
273 1772 : for key in ["receiverInfo", "notifierInfo"] {
274 1772 : if let Some(arr) = ep.get(key) {
275 200 : let ok = arr.as_array().is_some_and(|a| {
276 198 : a.iter().all(|kv| {
277 198 : kv.get("key").is_some_and(Value::is_string)
278 178 : && kv.get("value").is_some_and(Value::is_string)
279 198 : })
280 192 : });
281 200 : if !ok {
282 72 : return Err(bad(format!(
283 72 : "endpoint.{key} entries must be {{key, value}} pairs (5.2.15/5.2.22)"
284 72 : )));
285 128 : }
286 1572 : }
287 : }
288 : // 6.3.8 and 6.3.9: each receiverInfo pair becomes one custom
289 : // header on the notification POST, and "'Key' and 'value'
290 : // members shall adhere to IETF RFC 7230 ... definitions
291 : // concerning HTTP headers". A pair that cannot be a header is
292 : // input the operation cannot meet (5.8.1.4), so it is refused
293 : // here rather than accepted into a Subscription that can only
294 : // ever dead-letter. notifierInfo is not headers — its own
295 : // binding validates it through the sink.
296 838 : if let Some(arr) = ep.get("receiverInfo").and_then(Value::as_array) {
297 126 : for kv in arr {
298 126 : let (k, v) = (kv["key"].as_str(), kv["value"].as_str());
299 126 : if !k.is_some_and(is_field_name) || !v.is_some_and(is_field_value) {
300 76 : return Err(bad(format!(
301 76 : "endpoint.receiverInfo entry {kv} is not a valid HTTP header \
302 76 : (RFC 7230, 6.3.8)"
303 76 : )));
304 50 : }
305 : }
306 718 : }
307 : // Table 5.2.15-1: cooldown and timeout are Numbers "Greater
308 : // than 0"
309 1508 : for key in ["cooldown", "timeout"] {
310 1508 : if let Some(v) = ep.get(key) {
311 80 : v.as_f64().filter(|n| *n > 0.0).ok_or_else(|| {
312 32 : bad(format!(
313 32 : "endpoint.{key} must be a number greater than 0 (5.2.15)"
314 32 : ))
315 32 : })?;
316 1428 : }
317 : }
318 730 : if let Some(acc) = ep.get("accept").and_then(Value::as_str) {
319 26 : if ![
320 26 : "application/json",
321 26 : "application/ld+json",
322 26 : "application/geo+json",
323 26 : ]
324 26 : .contains(&acc)
325 : {
326 8 : return Err(bad(format!("invalid endpoint accept {acc:?}")));
327 18 : }
328 704 : }
329 722 : Ok(())
330 910 : }
331 :
332 : /// Table 5.2.12-1: "Valid notification triggers are entityCreated,
333 : /// entityUpdated, entityDeleted, attributeCreated, attributeUpdated,
334 : /// attributeDeleted." A trigger outside that set is accepted and then
335 : /// matches nothing, leaving a subscription that never fires.
336 42 : fn check_notification_triggers(v: &Value) -> Result<(), NgsiError> {
337 42 : let bad = NgsiError::BadRequestData;
338 : const TRIGGERS: [&str; 6] = [
339 : "entityCreated",
340 : "entityUpdated",
341 : "entityDeleted",
342 : "attributeCreated",
343 : "attributeUpdated",
344 : "attributeDeleted",
345 : ];
346 42 : let list = v.as_array().filter(|a| !a.is_empty()).ok_or_else(|| {
347 8 : bad("notificationTrigger must be a non-empty array of strings (5.2.12)".into())
348 8 : })?;
349 50 : for t in list {
350 50 : let t = t
351 50 : .as_str()
352 50 : .ok_or_else(|| bad("notificationTrigger entries must be strings (5.2.12)".into()))?;
353 46 : if !TRIGGERS.contains(&t) {
354 8 : return Err(bad(format!(
355 8 : "{t} is not a valid notification trigger (5.2.12)"
356 8 : )));
357 38 : }
358 : }
359 22 : Ok(())
360 42 : }
361 :
362 : /// One member of Table 5.2.12-1, validated and normalized into `out`.
363 : /// A member the table does not name is kept verbatim: 5.5.9 asks a
364 : /// receiver to tolerate what it does not know.
365 3880 : fn norm_member(
366 3880 : k: &str,
367 3880 : v: &Value,
368 3880 : ctx: &Context,
369 3880 : out: &mut Map<String, Value>,
370 3880 : ) -> Result<(), NgsiError> {
371 3880 : let bad = NgsiError::BadRequestData;
372 44 : match k {
373 3880 : "@context" | "createdAt" | "modifiedAt" | "status" => return Ok(()),
374 3850 : "id" => {
375 372 : let id = v
376 372 : .as_str()
377 372 : .ok_or_else(|| bad("subscription id must be a string URI".into()))?;
378 372 : antares_model::EntityId::new(id)?;
379 372 : out.insert("id".into(), v.clone());
380 : }
381 3478 : "type" => {
382 602 : if v.as_str() != Some("Subscription") {
383 0 : return Err(bad("type must be \"Subscription\" (5.2.12)".into()));
384 602 : }
385 602 : out.insert("type".into(), v.clone());
386 : }
387 2876 : "entities" => {
388 1248 : out.insert("entities".into(), norm_entity_selectors(v, ctx)?);
389 : }
390 1628 : "watchedAttributes" => {
391 28 : let arr = v
392 28 : .as_array()
393 28 : .filter(|a| !a.is_empty())
394 28 : .ok_or_else(|| bad("watchedAttributes must be a non-empty array".into()))?;
395 20 : let mut attrs = Vec::new();
396 20 : for a in arr {
397 20 : let s = a
398 20 : .as_str()
399 20 : .filter(|s| !s.is_empty())
400 20 : .ok_or_else(|| bad("watchedAttributes entries must be strings".into()))?;
401 12 : attrs.push(Value::String(ctx.expand_key(s)));
402 : }
403 12 : out.insert("watchedAttributes".into(), Value::Array(attrs));
404 : }
405 1600 : "q" => {
406 38 : let q = v.as_str().ok_or_else(|| bad("q must be a string".into()))?;
407 : // Validate the string the MATCHER will parse. `conditions_match`
408 : // percent-decodes first (4.9, 046_05), so validating the raw
409 : // form would let `%28%28%28…` through create-time checks and
410 : // only become thousands of real parens at notification time —
411 : // inside a spawned task, where the parser's own limits are the
412 : // last line of defence.
413 34 : let decoded = crate::negotiate::percent_decode(q.as_bytes());
414 34 : antares_ql::parse_q(&decoded)?;
415 26 : out.insert("q".into(), v.clone());
416 : }
417 1562 : "geoQ" => {
418 36 : let g = v
419 36 : .as_object()
420 36 : .ok_or_else(|| bad("geoQ must be an object".into()))?;
421 32 : antares_ql::geo::GeoQuery::from_params(&antares_matcher::geo_params(g))?
422 12 : .ok_or_else(|| bad("geoQ requires georel (5.2.13)".into()))?;
423 12 : let mut ng = g.clone();
424 12 : if let Some(gp) = g.get("geoproperty").and_then(Value::as_str) {
425 8 : ng.insert("geoproperty".into(), Value::String(ctx.expand_key(gp)));
426 8 : }
427 12 : out.insert("geoQ".into(), Value::Object(ng));
428 : }
429 1526 : "notification" => {
430 1136 : out.insert("notification".into(), norm_notification(v, ctx)?);
431 : }
432 390 : "expiresAt" => {
433 48 : let s = v
434 48 : .as_str()
435 48 : .filter(|s| parse_datetime(s))
436 48 : .ok_or_else(|| bad("expiresAt must be an ISO 8601 DateTime".into()))?;
437 : // 4.6.3 admits several spellings of one instant, so whether a
438 : // DateTime has passed cannot be read off the raw strings.
439 28 : if antares_model::dt_key(s) < antares_model::dt_key(&now_iso()) {
440 12 : return Err(bad("expiresAt is in the past (5.8.1)".into()));
441 16 : }
442 16 : out.insert("expiresAt".into(), v.clone());
443 : }
444 342 : "throttling" => {
445 30 : v.as_f64()
446 30 : .filter(|n| *n > 0.0)
447 30 : .ok_or_else(|| bad("throttling must be a positive number".into()))?;
448 10 : out.insert("throttling".into(), v.clone());
449 : }
450 312 : "timeInterval" => {
451 40 : v.as_f64()
452 40 : .filter(|n| *n > 0.0)
453 40 : .ok_or_else(|| bad("timeInterval must be a positive number".into()))?;
454 20 : out.insert("timeInterval".into(), v.clone());
455 : }
456 : // Table 5.2.12-1: all three are Booleans. `localOnly` and
457 : // `splitEntities` decide how far the Subscription reaches — a
458 : // string read through `as_bool()` is falsy, so accepting one
459 : // would turn a subscriber's request for local scope (5.5.13)
460 : // into a distributed subscription without telling it.
461 272 : "isActive" | "localOnly" | "splitEntities" => {
462 80 : if !v.is_boolean() {
463 32 : return Err(bad(format!("{k} must be a boolean (5.2.12)")));
464 48 : }
465 48 : out.insert(k.to_owned(), v.clone());
466 : }
467 192 : "temporalQ" => {
468 : // 5.2.21 TemporalQuery: timerel and timeAt are cardinality 1
469 : // and every member must sit in its Table 5.2.21-1 value
470 : // space (used by CSR subscriptions, 5.11.7).
471 40 : let tq = v
472 40 : .as_object()
473 40 : .ok_or_else(|| bad("temporalQ must be a TemporalQuery object (5.2.21)".into()))?;
474 32 : let mut p = std::collections::HashMap::new();
475 32 : crate::paging::temporal_q_params(tq, &mut p)?;
476 32 : crate::temporalq::TemporalQ::from_params(&p, true)?;
477 8 : out.insert(k.to_owned(), v.clone());
478 : }
479 : // Table 5.2.12-1: "Valid notification triggers are entityCreated,
480 : // entityUpdated, entityDeleted, attributeCreated, attributeUpdated,
481 : // attributeDeleted." A trigger outside that set is accepted and
482 : // then matches nothing, leaving a subscription that never fires.
483 152 : "notificationTrigger" => {
484 42 : check_notification_triggers(v)?;
485 22 : out.insert(k.to_owned(), v.clone());
486 : }
487 : // Table 5.2.12-1: csf is "A valid query string as per clause 4.9".
488 : // Unparsed here it is stored and only fails at Context Source
489 : // Registration matching (5.11.2.4), where it silently matches
490 : // nothing instead of telling the subscriber the filter is broken.
491 110 : "csf" => {
492 24 : let s = v
493 24 : .as_str()
494 24 : .ok_or_else(|| bad("csf must be a query string (5.2.12)".into()))?;
495 16 : antares_ql::parse_q(s)?;
496 8 : out.insert(k.to_owned(), v.clone());
497 : }
498 : // Table 5.2.12-1: both are a String, "comma separated list of
499 : // attribute names", and the pair decides how the notification
500 : // condition compares a value (4.9). A non-string stored here reads
501 : // as an absent list at matching time, so the Subscription would
502 : // silently compare the wrong thing rather than say the member is
503 : // malformed.
504 86 : "expandValues" | "jsonKeys" => {
505 42 : if !v.is_string() {
506 32 : return Err(bad(format!(
507 32 : "{k} must be a comma separated list of attribute names (5.2.12)"
508 32 : )));
509 10 : }
510 10 : out.insert(k.to_owned(), v.clone());
511 : }
512 : // 5.8.6: the @context governing a subscription's notifications is
513 : // the @context of the creating request, held in a broker-internal
514 : // member. This function only ever sees client input — a create
515 : // body or a patch fragment — so dropping the member here stops a
516 : // subscriber both from seeding it and from replacing it later.
517 : // __via is the same class: the 6.3.18 chain comes from the Via
518 : // HTTP header of the creating request, never from the body, and so
519 : // is __subject. The prefix as a whole is the broker's, so a member
520 : // added later cannot be forgotten here.
521 44 : k if k.starts_with("__") => return Ok(()),
522 24 : "scopeQ" | "lang" | "subscriptionName" | "name" | "description" | "jsonldContext"
523 20 : | "ngsildConformance" | "datasetId" => {
524 4 : out.insert(k.to_owned(), v.clone());
525 4 : }
526 : // tolerant reader: keep unknown members
527 20 : _ => {
528 20 : out.insert(k.to_owned(), v.clone());
529 20 : }
530 : }
531 3116 : Ok(())
532 3880 : }
533 :
534 : /// The Table 5.2.12-1 rules that hold between members rather than over
535 : /// one: what a Subscription must carry, and the pairs it may not.
536 622 : fn check_subscription_members(out: &Map<String, Value>, is_patch: bool) -> Result<(), NgsiError> {
537 622 : let bad = NgsiError::BadRequestData;
538 622 : if !is_patch {
539 578 : if !out.contains_key("type") {
540 0 : return Err(bad("type must be \"Subscription\" (5.2.12)".into()));
541 578 : }
542 : // 5.2.12: "At least one of (a) entities or (b) watchedAttributes
543 : // shall be present, unless the member localOnly is set to true"
544 : // (local scope, 5.5.13).
545 578 : let local_only = out.get("localOnly").and_then(Value::as_bool) == Some(true);
546 578 : if !local_only && !out.contains_key("entities") && !out.contains_key("watchedAttributes") {
547 12 : return Err(bad(
548 12 : "one of entities or watchedAttributes is required (5.2.12)".into(),
549 12 : ));
550 566 : }
551 566 : if !out.contains_key("notification") {
552 4 : return Err(bad("notification is required (5.2.12)".into()));
553 562 : }
554 44 : }
555 606 : if out.contains_key("timeInterval") && out.contains_key("watchedAttributes") {
556 8 : return Err(bad(
557 8 : "timeInterval and watchedAttributes are mutually exclusive (5.2.12)".into(),
558 8 : ));
559 598 : }
560 598 : if out.contains_key("timeInterval") && out.contains_key("throttling") {
561 4 : return Err(bad(
562 4 : "timeInterval and throttling are mutually exclusive (5.2.12)".into(),
563 4 : ));
564 594 : }
565 594 : Ok(())
566 622 : }
567 :
568 : /// Validate + normalize a subscription document (5.8.1). Types/attribute
569 : /// names are expanded to IRIs; the rest is stored verbatim.
570 1348 : pub fn normalize_subscription(
571 1348 : doc: &Map<String, Value>,
572 1348 : ctx: &Context,
573 1348 : is_patch: bool,
574 1348 : ) -> Result<Map<String, Value>, NgsiError> {
575 1348 : let bad = NgsiError::BadRequestData;
576 : // 5.5.4: first-level member nulls are only legal in fragments (patch)
577 1348 : if !is_patch {
578 1288 : antares_jsonld::reject_first_level_nulls(doc)?;
579 60 : }
580 1344 : let mut out = Map::new();
581 3892 : for (k, v) in doc {
582 : // 5.4 Fragment member removal: null and the NGSI-LD Null both delete
583 : // the member. Read literally, the NGSI-LD Null was stored as the
584 : // string it is spelled with, so the member survived carrying it.
585 3892 : if is_patch && k != "id" && (v.is_null() || v.as_str() == Some("urn:ngsi-ld:null")) {
586 12 : if ["type", "notification"].contains(&k.as_str()) {
587 8 : return Err(bad(format!("cannot remove mandatory member {k} (5.8.3)")));
588 4 : }
589 4 : out.insert(k.clone(), Value::Null);
590 4 : continue;
591 3880 : }
592 3880 : norm_member(k, v, ctx, &mut out)?;
593 : }
594 622 : check_subscription_members(&out, is_patch)?;
595 594 : Ok(out)
596 1348 : }
597 :
598 : /// Output shaping: compact IRIs, add status (5.8.3).
599 98 : pub fn present_subscription(doc: &Value, ctx: &Context, sys_attrs: bool, csource: bool) -> Value {
600 98 : let Some(obj) = doc.as_object() else {
601 0 : return doc.clone();
602 : };
603 98 : let mut out = Map::new();
604 664 : for (k, v) in obj {
605 664 : match k.as_str() {
606 : // 5.8.3/5.8.4 serve the 5.2.12 data type, which has no member
607 : // under this prefix — see policy::SUBJECT_MEMBER
608 664 : k if k.starts_with("__") => continue,
609 584 : "createdAt" | "modifiedAt" if !sys_attrs => continue,
610 468 : "entities" => {
611 98 : let entities: Vec<Value> = v
612 98 : .as_array()
613 98 : .cloned()
614 98 : .unwrap_or_default()
615 98 : .iter()
616 98 : .map(|e| {
617 98 : let mut ne = e.as_object().cloned().unwrap_or_default();
618 98 : if let Some(t) = ne.get("type").and_then(Value::as_str) {
619 98 : ne.insert("type".into(), Value::String(ctx.compact_iri(t)));
620 98 : }
621 98 : Value::Object(ne)
622 98 : })
623 98 : .collect();
624 98 : out.insert("entities".into(), Value::Array(entities));
625 : }
626 370 : "watchedAttributes" => {
627 24 : let attrs: Vec<Value> = v
628 24 : .as_array()
629 24 : .cloned()
630 24 : .unwrap_or_default()
631 24 : .iter()
632 24 : .filter_map(Value::as_str)
633 24 : .map(|a| Value::String(ctx.compact_iri(a)))
634 24 : .collect();
635 24 : out.insert("watchedAttributes".into(), Value::Array(attrs));
636 : }
637 346 : "notification" => {
638 98 : let mut n = v.as_object().cloned().unwrap_or_default();
639 98 : if let Some(attrs) = n.get("attributes").and_then(Value::as_array) {
640 24 : let na: Vec<Value> = attrs
641 24 : .iter()
642 24 : .filter_map(Value::as_str)
643 24 : .map(|a| Value::String(ctx.compact_iri(a)))
644 24 : .collect();
645 24 : n.insert("attributes".into(), Value::Array(na));
646 74 : }
647 98 : out.insert("notification".into(), Value::Object(n));
648 : }
649 248 : "geoQ" => {
650 24 : let mut g = v.as_object().cloned().unwrap_or_default();
651 24 : if let Some(gp) = g.get("geoproperty").and_then(Value::as_str) {
652 24 : g.insert("geoproperty".into(), Value::String(ctx.compact_iri(gp)));
653 24 : }
654 24 : out.insert("geoQ".into(), Value::Object(g));
655 : }
656 252 : _ => {
657 252 : out.insert(k.clone(), v.clone());
658 252 : }
659 : }
660 : }
661 : // default notificationTrigger surfaced on output (5.2.12; 028_06) —
662 : // entity subscriptions only, csource subs have no such default (5.11)
663 98 : if !csource && !out.contains_key("notificationTrigger") && !out.contains_key("timeInterval") {
664 82 : out.insert(
665 82 : "notificationTrigger".into(),
666 82 : serde_json::json!(["attributeCreated", "attributeUpdated"]),
667 82 : );
668 82 : }
669 : // status (5.2.12 output): active | paused | expired
670 98 : let expired = obj
671 98 : .get("expiresAt")
672 98 : .and_then(Value::as_str)
673 98 : .is_some_and(|e| antares_model::dt_key(e) < antares_model::dt_key(&now_iso()));
674 98 : let paused = obj.get("isActive") == Some(&Value::Bool(false));
675 98 : let status = if expired {
676 8 : "expired"
677 90 : } else if paused {
678 4 : "paused"
679 86 : } else if obj.get("status").and_then(Value::as_str) == Some("failed") {
680 0 : "failed" // 5.8.6 / 5.11.7 delivery-failure status
681 : } else {
682 86 : "active"
683 : };
684 98 : out.insert("status".into(), Value::String(status.into()));
685 98 : Value::Object(out)
686 98 : }
687 :
688 : // ---------- handlers (parameterized by Kind) ----------
689 :
690 : /// Validate a subscription's jsonldContext member (5.2.12): must be a
691 : /// dereferenceable @context — invalid value ⇒ 400, unresolvable ⇒ 504.
692 : /// 5.8.1.4 and 5.8.2.4: the notification endpoint has to be one this
693 : /// deployment can deliver to. The sink registered for the URI's scheme
694 : /// (6.3.8, and 7.2 for the optional MQTT binding) validates the endpoint's
695 : /// own syntax and its `notifierInfo`; a scheme no sink serves is input data
696 : /// that does not meet the requirements of the operation — BadRequestData,
697 : /// never a fall-through to the HTTP binding. A fragment that carries no
698 : /// endpoint leaves the stored one in place and has nothing to check.
699 354 : fn check_endpoint(st: &AppState, norm: &Map<String, Value>) -> Result<(), NgsiError> {
700 354 : let Some(ep) = norm
701 354 : .get("notification")
702 354 : .and_then(|n| n.get("endpoint"))
703 354 : .and_then(Value::as_object)
704 : else {
705 20 : return Ok(());
706 : };
707 334 : let Some(uri) = ep.get("uri").and_then(Value::as_str) else {
708 0 : return Ok(());
709 : };
710 334 : let notifier_info = ep
711 334 : .get("notifierInfo")
712 334 : .and_then(Value::as_array)
713 334 : .map(|ni| {
714 4 : ni.iter()
715 4 : .filter_map(|kv| Some((kv.get("key")?.as_str()?, kv.get("value")?.as_str()?)))
716 4 : .collect::<Vec<_>>()
717 4 : })
718 334 : .unwrap_or_default();
719 334 : st.sinks.require(uri, ¬ifier_info)
720 354 : }
721 :
722 348 : async fn check_jsonld_context(
723 348 : st: &AppState,
724 348 : tenant: &TenantId,
725 348 : norm: &Map<String, Value>,
726 348 : ) -> Result<(), ApiError> {
727 348 : let Some(v) = norm.get("jsonldContext") else {
728 344 : return Ok(());
729 : };
730 4 : let is_url = |s: &str| s.starts_with("http://") || s.starts_with("https://");
731 4 : let ok_shape = match v {
732 4 : Value::String(s) => is_url(s),
733 0 : Value::Array(a) => a.iter().all(|e| e.as_str().is_some_and(is_url)),
734 0 : _ => false,
735 : };
736 4 : if !ok_shape {
737 0 : return Err(NgsiError::BadRequestData(format!(
738 0 : "jsonldContext is not a valid @context reference: {v}"
739 0 : ))
740 0 : .into());
741 4 : }
742 4 : st.loader.resolve_for(tenant, v).await?;
743 2 : Ok(())
744 348 : }
745 :
746 : /// One function serves both subscription resources, and they are different
747 : /// clauses: 5.8 for Subscriptions, 5.11 for Context Source Registration
748 : /// Subscriptions. The seam is asked about the operation the caller actually
749 : /// made.
750 88 : fn clause_of(kind: Kind, sub: &'static str, csub: &'static str) -> &'static str {
751 88 : if kind == Kind::CSourceSubscription {
752 28 : csub
753 : } else {
754 60 : sub
755 : }
756 88 : }
757 :
758 940 : pub async fn create(
759 940 : st: &AppState,
760 940 : kind: Kind,
761 940 : params: &HashMap<String, String>,
762 940 : headers: &HeaderMap,
763 940 : body: &[u8],
764 940 : ) -> ApiResult<Response> {
765 940 : let tenant = tenant_from(headers)?;
766 940 : check_params(params, &["local"])?;
767 936 : let parsed = parse_body(&st.loader, headers, body, BodyKind::Standard).await?;
768 492 : let obj = parsed.object(NgsiError::BadRequestData(
769 492 : "subscription must be a JSON object".into(),
770 492 : ))?;
771 : // ADR-0020: the client may choose the id (5.8.1.4 / 5.11.2.4), and it is
772 : // in hand here — an engine that owns a segment of the id space has to be
773 : // told which one is being claimed. A body that names none leaves the
774 : // list empty, which is right: there is no client choice to decide about.
775 492 : let named = obj.get("id").and_then(Value::as_str);
776 492 : gate!(
777 : st, &tenant, headers, clause_of(kind, "5.8.1", "5.11.2"),
778 : ids: named.as_slice(),
779 : )
780 492 : .await?;
781 492 : let mut norm = normalize_subscription(obj, &parsed.ctx, false)?;
782 326 : check_endpoint(st, &norm)?;
783 320 : check_jsonld_context(st, &tenant, &norm).await?;
784 318 : let id = match norm.get("id").and_then(Value::as_str) {
785 242 : Some(id) => id.to_owned(),
786 : None => {
787 76 : let id = format!("urn:ngsi-ld:Subscription:{}", uuid::Uuid::new_v4());
788 76 : norm.insert("id".into(), Value::String(id.clone()));
789 76 : id
790 : }
791 : };
792 : // The Registration Subscriptions the distributed half owns (5.8.1.4)
793 : // are stored outside the client kinds, under this id namespace: a
794 : // client document claiming it would be looked up in that store on the
795 : // notification path and would silently never notify.
796 318 : if id.starts_with(crate::registry::INTERNAL_CSR_PREFIX) {
797 0 : return Err(NgsiError::BadRequestData(format!(
798 0 : "subscription id {id} is reserved by the broker"
799 0 : ))
800 0 : .into());
801 318 : }
802 318 : let ts = now_iso();
803 318 : norm.insert("createdAt".into(), Value::String(ts.clone()));
804 318 : norm.insert("modifiedAt".into(), Value::String(ts.clone()));
805 : // notification @context = the creating request's context (5.8.6),
806 : // stored as its own column — internal member, stripped on output.
807 318 : norm.insert("__context".into(), parsed.ctx.source.clone());
808 : // 6.3.17/6.3.18: a Subscription arriving as a forwarded copy (5.8.1.4)
809 : // carries the Via chain of the brokers it has passed through — kept on
810 : // the stored document so the distributed half can extend the chain
811 : // outbound and refuse to re-forward a copy that has looped back.
812 318 : if let Some(via) = crate::federation::inbound_via(headers) {
813 16 : norm.insert("__via".into(), Value::String(via));
814 302 : }
815 : // ADR-0020: 5.8.6 delivery is broker-initiated, so there is no request
816 : // to read a subject off when the notification is sent. The subject is
817 : // the subscriber's, taken from the creating request and kept the way
818 : // the notification @context is — internal member, never rendered, never
819 : // forwarded, never settable by a client.
820 20 : if let Some(subject) =
821 318 : crate::policy::subject_member(&crate::policy::subject_of(&tenant, headers))
822 20 : {
823 20 : norm.insert(crate::policy::SUBJECT_MEMBER.into(), subject);
824 298 : }
825 : // Array @context (>1 entry): the broker must host it at its own URL as an
826 : // ImplicitlyCreated @context, surfaced via jsonldContext (5.13.1, 050_03)
827 318 : if !norm.contains_key("jsonldContext") {
828 316 : if let Value::Array(a) = &parsed.ctx.source {
829 2 : if a.len() > 1 {
830 2 : let local_id = uuid::Uuid::new_v4().to_string();
831 2 : let url = format!("{}/{local_id}", crate::contexts::base_url(headers));
832 2 : st.store
833 2 : .context_put(
834 2 : Some(&tenant),
835 2 : &local_id,
836 2 : serde_json::json!({
837 2 : "url": url,
838 2 : "localId": local_id,
839 2 : "kind": "ImplicitlyCreated",
840 2 : "createdAt": ts,
841 2 : // owned by the tenant whose subscription created it
842 2 : "owner": tenant.as_str(),
843 2 : "body": {"@context": parsed.ctx.source.clone()},
844 2 : }),
845 2 : )
846 2 : .await?;
847 2 : st.loader
848 2 : .put_local_for(&tenant, url.clone(), parsed.ctx.source.clone())
849 2 : .await;
850 2 : norm.insert("jsonldContext".into(), Value::String(url));
851 0 : }
852 314 : }
853 2 : }
854 318 : let doc = Value::Object(norm);
855 318 : if !st.store.create(&tenant, kind, &id, doc.clone()).await? {
856 0 : return Err(NgsiError::AlreadyExists(format!("subscription {id} already exists")).into());
857 318 : }
858 318 : st.sub_changed(&tenant, kind, &id, Some(&doc));
859 318 : if kind == Kind::Subscription {
860 : // 5.8.1.4: a distributed Subscription creates its internal Context
861 : // Source Registration Subscription (consumer half)
862 308 : crate::distsub::on_subscription_created(st, &tenant, &doc).await;
863 10 : }
864 318 : if kind == Kind::CSourceSubscription {
865 : // initial CSourceNotification with all matching registrations (5.11.2.4)
866 10 : let (st2, t2, id2) = (st.clone(), tenant.clone(), id.clone());
867 10 : crate::spawn(async move {
868 10 : crate::notify::csource_initial(&st2, &t2, &id2).await;
869 10 : });
870 308 : }
871 318 : Ok(created(
872 318 : format!("/ngsi-ld/v1/{}/{id}", resource_path(kind)),
873 318 : &tenant,
874 318 : ))
875 940 : }
876 :
877 : /// 5.8.3 Retrieve Subscription: invalid URI 400, unknown id 404, else the
878 : /// 5.2.12 subscription document (status/timesSent are output members).
879 122 : pub async fn retrieve(
880 122 : st: &AppState,
881 122 : kind: Kind,
882 122 : id: &str,
883 122 : params: &HashMap<String, String>,
884 122 : headers: &HeaderMap,
885 122 : ) -> ApiResult<Response> {
886 122 : let tenant = tenant_from(headers)?;
887 122 : antares_model::EntityId::new(id)
888 122 : .map_err(|_| NgsiError::BadRequestData(format!("invalid subscription id {id:?}")))?;
889 112 : check_params(params, &["options", "format", "sysAttrs", "local"])?;
890 112 : let accept = parse_accept(headers)?;
891 104 : let ctx = request_context(&st.loader, headers).await?;
892 100 : gate!(st, &tenant, headers, clause_of(kind, "5.8.3", "5.11.4"), ids: &[id]).await?;
893 100 : let doc = st
894 100 : .store
895 100 : .get(&tenant, kind, id)
896 100 : .await?
897 100 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("subscription {id} not found")))?;
898 24 : let sys = sys_attrs_asked(params);
899 24 : let payload = present_subscription(&doc, &ctx, sys, kind == Kind::CSourceSubscription);
900 24 : Ok(respond(StatusCode::OK, payload, &ctx, accept, &tenant))
901 122 : }
902 :
903 : /// 5.8.4 Query Subscriptions: list with 5.5.9 pagination; each element a
904 : /// 5.2.12 subscription document.
905 140 : pub async fn list(
906 140 : st: &AppState,
907 140 : kind: Kind,
908 140 : params: &HashMap<String, String>,
909 140 : headers: &HeaderMap,
910 140 : ) -> ApiResult<Response> {
911 140 : let tenant = tenant_from(headers)?;
912 140 : check_params(
913 140 : params,
914 140 : &["limit", "offset", "count", "options", "format", "local"],
915 4 : )?;
916 136 : let accept = parse_accept(headers)?;
917 132 : let ctx = request_context(&st.loader, headers).await?;
918 132 : gate!(st, &tenant, headers, clause_of(kind, "5.8.4", "5.11.5")).await?;
919 : // 5.5.9.1: "only up to a maximum of L NGSI-LD Elements are RETRIEVED
920 : // and returned". 5.8.4 takes no filter parameters — `check_params`
921 : // above is the whole list — so the tenant IS the match set and the
922 : // window is the store's to apply. Reading the tenant and slicing it
923 : // here made a tenant at the document ceiling unable to list at all,
924 : // because that read is the one carrying the ceiling for client queries.
925 132 : let (offset, limit, _) = crate::paging::page_params(st, params)?;
926 120 : let (page_docs, total) = st.store.list_slice(&tenant, kind, offset, limit).await?;
927 120 : let (page, count_hdr, links) = crate::paging::paginate_pre_accept(
928 120 : st,
929 120 : params,
930 120 : page_docs,
931 120 : &format!("/ngsi-ld/v1/{}", resource_path(kind)),
932 120 : accept,
933 120 : total,
934 0 : )?;
935 120 : let sys = sys_attrs_asked(params);
936 120 : let payload: Vec<Value> = page
937 120 : .iter()
938 120 : .map(|d| present_subscription(d, &ctx, sys, kind == Kind::CSourceSubscription))
939 120 : .collect();
940 120 : let mut resp = crate::negotiate::respond_list(StatusCode::OK, payload, &ctx, accept, &tenant);
941 120 : attach_paging(&mut resp, count_hdr, &links);
942 120 : Ok(resp)
943 140 : }
944 :
945 : /// 5.8.2 Update Subscription: invalid URI 400, unknown id 404, fragment
946 : /// validated per 5.5.4 + 5.2.12 (past expiresAt 400), jsonldContext
947 : /// unavailable -> LdContextNotAvailable / invalid -> 400, modify per 5.5.8;
948 : /// the 5.8.2.4 status table falls out of the computed status
949 : /// (isActive/expiresAt) on read.
950 48 : pub async fn update(
951 48 : st: &AppState,
952 48 : kind: Kind,
953 48 : id: &str,
954 48 : params: &HashMap<String, String>,
955 48 : headers: &HeaderMap,
956 48 : body: &[u8],
957 48 : ) -> ApiResult<Response> {
958 48 : let tenant = tenant_from(headers)?;
959 48 : antares_model::EntityId::new(id)
960 48 : .map_err(|_| NgsiError::BadRequestData(format!("invalid subscription id {id:?}")))?;
961 40 : check_params(params, &["local"])?;
962 40 : gate!(st, &tenant, headers, clause_of(kind, "5.8.2", "5.11.3"), ids: &[id]).await?;
963 40 : let parsed = parse_body(&st.loader, headers, body, BodyKind::MergePatch).await?;
964 28 : let obj = parsed.object(NgsiError::BadRequestData(
965 28 : "fragment must be a JSON object".into(),
966 28 : ))?;
967 28 : if let Some(bid) = obj.get("id").and_then(Value::as_str) {
968 0 : if bid != id {
969 0 : return Err(NgsiError::BadRequestData("fragment id mismatch".into()).into());
970 0 : }
971 28 : }
972 28 : let norm = normalize_subscription(obj, &parsed.ctx, true)?;
973 28 : check_endpoint(st, &norm)?;
974 28 : check_jsonld_context(st, &tenant, &norm).await?;
975 28 : let ts = now_iso();
976 28 : let res = st
977 28 : .store
978 28 : .mutate(&tenant, kind, id, |doc| {
979 4 : let target = antares_store::stored_object(doc)?;
980 4 : crate::apply_doc_fragment(target, &norm, &ts);
981 4 : Ok::<(), NgsiError>(())
982 4 : })
983 28 : .await?;
984 4 : match res {
985 24 : None => Err(NgsiError::ResourceNotFound(format!("subscription {id} not found")).into()),
986 0 : Some(Err(e)) => Err(e.into()),
987 : Some(Ok(())) => {
988 4 : if kind == Kind::CSourceSubscription {
989 : // 5.11.3.4: after update, notify with all currently matching
990 0 : let (st2, t2, id2) = (st.clone(), tenant.clone(), id.to_owned());
991 0 : crate::spawn(async move {
992 0 : crate::notify::csource_initial(&st2, &t2, &id2).await;
993 0 : });
994 4 : }
995 4 : if st.sub_sync.is_some() {
996 4 : let doc = st.store.get(&tenant, kind, id).await?;
997 4 : st.sub_changed(&tenant, kind, id, doc.as_ref());
998 0 : }
999 4 : if kind == Kind::Subscription {
1000 : // 5.8.2.4: the CSR subscription and the mapped remote
1001 : // subscriptions follow the update (5.11.3)
1002 4 : crate::distsub::on_subscription_updated(st, &tenant, id).await;
1003 0 : }
1004 4 : Ok(no_content(&tenant))
1005 : }
1006 : }
1007 48 : }
1008 :
1009 : /// 5.8.5 Delete Subscription: invalid URI 400, unknown id 404, 204 on
1010 : /// success and no further notifications (sub_changed drops the mirror).
1011 266 : pub async fn delete(
1012 266 : st: &AppState,
1013 266 : kind: Kind,
1014 266 : id: &str,
1015 266 : params: &HashMap<String, String>,
1016 266 : headers: &HeaderMap,
1017 266 : ) -> ApiResult<Response> {
1018 266 : let tenant = tenant_from(headers)?;
1019 266 : antares_model::EntityId::new(id)
1020 266 : .map_err(|_| NgsiError::BadRequestData(format!("invalid subscription id {id:?}")))?;
1021 258 : check_params(params, &["local"])?;
1022 258 : gate!(st, &tenant, headers, clause_of(kind, "5.8.5", "5.11.6"), ids: &[id]).await?;
1023 258 : if st.store.delete(&tenant, kind, id).await? {
1024 46 : st.sub_changed(&tenant, kind, id, None);
1025 46 : if kind == Kind::Subscription {
1026 : // 5.8.5.4: forward the delete to every mapped Context Source
1027 : // and drop the internal CSR subscription (5.11.6)
1028 46 : crate::distsub::on_subscription_deleted(st, &tenant, id).await;
1029 0 : }
1030 46 : Ok(no_content(&tenant))
1031 : } else {
1032 212 : Err(NgsiError::ResourceNotFound(format!("subscription {id} not found")).into())
1033 : }
1034 266 : }
1035 :
1036 : // axum route fns
1037 :
1038 : macro_rules! route4 {
1039 : ($create:ident, $retrieve:ident, $list:ident, $update:ident, $delete:ident, $kind:expr, $c:literal) => {
1040 : #[doc = concat!("HTTP handler for ", $c, " Create: the axum seam over [`create`].")]
1041 940 : pub async fn $create(
1042 940 : State(st): State<AppState>,
1043 940 : CleanParams(params): CleanParams,
1044 940 : headers: HeaderMap,
1045 940 : body: Bytes,
1046 940 : ) -> Response {
1047 940 : create(&st, $kind, ¶ms, &headers, &body)
1048 940 : .await
1049 940 : .unwrap_or_else(|e| e.into_response())
1050 940 : }
1051 : #[doc = concat!("HTTP handler for ", $c, " Retrieve: the axum seam over [`retrieve`].")]
1052 122 : pub async fn $retrieve(
1053 122 : State(st): State<AppState>,
1054 122 : Path(id): Path<String>,
1055 122 : CleanParams(params): CleanParams,
1056 122 : headers: HeaderMap,
1057 122 : ) -> Response {
1058 122 : retrieve(&st, $kind, &id, ¶ms, &headers)
1059 122 : .await
1060 122 : .unwrap_or_else(|e| e.into_response())
1061 122 : }
1062 : #[doc = concat!("HTTP handler for ", $c, " Query: the axum seam over [`list`].")]
1063 140 : pub async fn $list(
1064 140 : State(st): State<AppState>,
1065 140 : CleanParams(params): CleanParams,
1066 140 : headers: HeaderMap,
1067 140 : ) -> Response {
1068 140 : list(&st, $kind, ¶ms, &headers)
1069 140 : .await
1070 140 : .unwrap_or_else(|e| e.into_response())
1071 140 : }
1072 : #[doc = concat!("HTTP handler for ", $c, " Update: the axum seam over [`update`].")]
1073 48 : pub async fn $update(
1074 48 : State(st): State<AppState>,
1075 48 : Path(id): Path<String>,
1076 48 : CleanParams(params): CleanParams,
1077 48 : headers: HeaderMap,
1078 48 : body: Bytes,
1079 48 : ) -> Response {
1080 48 : update(&st, $kind, &id, ¶ms, &headers, &body)
1081 48 : .await
1082 48 : .unwrap_or_else(|e| e.into_response())
1083 48 : }
1084 : #[doc = concat!("HTTP handler for ", $c, " Delete: the axum seam over [`delete`].")]
1085 266 : pub async fn $delete(
1086 266 : State(st): State<AppState>,
1087 266 : Path(id): Path<String>,
1088 266 : CleanParams(params): CleanParams,
1089 266 : headers: HeaderMap,
1090 266 : ) -> Response {
1091 266 : delete(&st, $kind, &id, ¶ms, &headers)
1092 266 : .await
1093 266 : .unwrap_or_else(|e| e.into_response())
1094 266 : }
1095 : };
1096 : }
1097 :
1098 : route4!(
1099 : create_subscription,
1100 : retrieve_subscription,
1101 : query_subscriptions,
1102 : update_subscription,
1103 : delete_subscription,
1104 : Kind::Subscription,
1105 : "5.8 Subscription"
1106 : );
1107 : route4!(
1108 : create_csource_subscription,
1109 : retrieve_csource_subscription,
1110 : query_csource_subscriptions,
1111 : update_csource_subscription,
1112 : delete_csource_subscription,
1113 : Kind::CSourceSubscription,
1114 : "5.11 Context Source Registration Subscription"
1115 : );
1116 :
1117 : #[cfg(test)]
1118 : mod tests {
1119 : use super::*;
1120 : use antares_jsonld::Loader;
1121 : use serde_json::json;
1122 :
1123 : /// Table 5.2.33-1 `type`: "To indicate a request for all Entities (with
1124 : /// implied local scope), \"*\" is also allowed as a value." It is neither
1125 : /// a term nor a 4.17 expression: expanding it yields an IRI no entity
1126 : /// carries, so the subscription is created and then silently notifies
1127 : /// nothing. The stored form has to survive normalization for the matcher
1128 : /// to read it.
1129 : #[test]
1130 4 : fn clause_5_2_33_a_star_selector_survives_normalization_and_matches() {
1131 4 : let ctx = Loader::new().core();
1132 4 : let doc = json!({
1133 4 : "id": "urn:ngsi-ld:Subscription:star",
1134 4 : "type": "Subscription",
1135 4 : "entities": [{"type": "*"}],
1136 4 : "notification": {"endpoint": {"uri": "http://localhost:1/n"}},
1137 : });
1138 4 : let norm = normalize_subscription(doc.as_object().expect("object"), &ctx, false)
1139 4 : .expect("Table 5.2.33-1 allows \"*\"");
1140 4 : assert_eq!(
1141 4 : norm["entities"][0]["type"],
1142 4 : json!("*"),
1143 : "\"*\" must stay raw: an expanded term matches no entity"
1144 : );
1145 4 : let stored = Value::Object(norm);
1146 4 : assert!(
1147 4 : antares_matcher::selector_match(
1148 4 : &stored,
1149 4 : &json!({"id": "urn:ngsi-ld:Vehicle:1",
1150 4 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/Vehicle"]}),
1151 4 : &ctx,
1152 : ),
1153 : "a stored \"*\" subscription must match every Entity"
1154 : );
1155 4 : }
1156 :
1157 : /// 5.5.4: "urn:ngsi-ld:null" as a first-level member value is
1158 : /// BadRequestData on create; in a patch fragment it is the removal form.
1159 : #[test]
1160 4 : fn clause_5_5_4_first_level_null_in_subscription() {
1161 4 : let ctx = Loader::new().core();
1162 4 : let doc = json!({
1163 4 : "type": "Subscription",
1164 4 : "entities": [{"type": "Building"}],
1165 4 : "description": "urn:ngsi-ld:null",
1166 4 : "notification": {"endpoint": {"uri": "http://localhost:1111/notify"}}
1167 : });
1168 4 : assert!(
1169 4 : normalize_subscription(doc.as_object().unwrap(), &ctx, false).is_err(),
1170 : "create with a first-level null URN must be rejected"
1171 : );
1172 : // patch fragment: 5.4 removal semantics — the member is marked for
1173 : // deletion, never stored carrying the string it is spelled with
1174 4 : let patch = json!({"description": "urn:ngsi-ld:null"});
1175 4 : let n = normalize_subscription(patch.as_object().unwrap(), &ctx, true).expect("fragment");
1176 4 : assert_eq!(
1177 4 : n["description"],
1178 : Value::Null,
1179 : "the NGSI-LD Null removes the member, like a JSON null"
1180 : );
1181 : // and a mandatory member cannot be removed at all (5.8.3)
1182 8 : for k in ["type", "notification"] {
1183 8 : let patch = json!({ k: "urn:ngsi-ld:null" });
1184 8 : assert!(
1185 8 : normalize_subscription(patch.as_object().unwrap(), &ctx, true).is_err(),
1186 : "removing the mandatory member {k} must be refused"
1187 : );
1188 : }
1189 4 : }
1190 :
1191 : #[test]
1192 4 : fn validates_subscription() {
1193 4 : let ctx = Loader::new().core();
1194 4 : let doc = json!({
1195 4 : "id": "urn:ngsi-ld:Subscription:1",
1196 4 : "type": "Subscription",
1197 4 : "entities": [{"type": "Building"}],
1198 4 : "notification": {"endpoint": {"uri": "http://localhost:1111/notify"}}
1199 : });
1200 4 : let n = normalize_subscription(doc.as_object().unwrap(), &ctx, false).expect("valid");
1201 4 : assert_eq!(
1202 4 : n["entities"][0]["type"],
1203 : "https://uri.etsi.org/ngsi-ld/default-context/Building"
1204 : );
1205 :
1206 4 : let missing_notification = json!({
1207 4 : "type": "Subscription",
1208 4 : "entities": [{"type": "Building"}]
1209 : });
1210 4 : assert!(
1211 4 : normalize_subscription(missing_notification.as_object().unwrap(), &ctx, false).is_err()
1212 : );
1213 :
1214 4 : let past_expiry = json!({
1215 4 : "type": "Subscription",
1216 4 : "entities": [{"type": "Building"}],
1217 4 : "expiresAt": "2020-01-01T00:00:00Z",
1218 4 : "notification": {"endpoint": {"uri": "http://localhost:1111/notify"}}
1219 : });
1220 4 : assert!(normalize_subscription(past_expiry.as_object().unwrap(), &ctx, false).is_err());
1221 4 : }
1222 :
1223 : /// Table 5.2.14.1-1 p.120: "showChanges cannot be true in case format is
1224 : /// keyValues". "simplified" is the table's declared synonym.
1225 : #[test]
1226 4 : fn show_changes_with_key_values_is_rejected() {
1227 4 : let ctx = Loader::new().core();
1228 20 : let mk = |format: &str, show: bool| {
1229 20 : json!({
1230 20 : "type": "Subscription",
1231 20 : "entities": [{"type": "Building"}],
1232 20 : "notification": {
1233 20 : "format": format,
1234 20 : "showChanges": show,
1235 20 : "endpoint": {"uri": "http://localhost:1111/notify"}
1236 : }
1237 : })
1238 20 : };
1239 8 : for f in ["keyValues", "simplified"] {
1240 8 : assert!(
1241 8 : normalize_subscription(mk(f, true).as_object().unwrap(), &ctx, false).is_err(),
1242 : "showChanges+{f} must be rejected"
1243 : );
1244 8 : assert!(
1245 8 : normalize_subscription(mk(f, false).as_object().unwrap(), &ctx, false).is_ok(),
1246 : "{f} without showChanges is fine"
1247 : );
1248 : }
1249 4 : assert!(
1250 4 : normalize_subscription(mk("normalized", true).as_object().unwrap(), &ctx, false)
1251 4 : .is_ok(),
1252 : "showChanges+normalized is fine"
1253 : );
1254 4 : }
1255 :
1256 : /// Table 5.2.14.1-1 p.119: "Empty array (0 length) is not allowed" on
1257 : /// notification.attributes / pick / omit.
1258 : #[test]
1259 4 : fn empty_projection_arrays_are_rejected() {
1260 4 : let ctx = Loader::new().core();
1261 12 : for key in ["attributes", "pick", "omit"] {
1262 12 : let doc = json!({
1263 12 : "type": "Subscription",
1264 12 : "entities": [{"type": "Building"}],
1265 12 : "notification": {
1266 12 : key: [],
1267 12 : "endpoint": {"uri": "http://localhost:1111/notify"}
1268 : }
1269 : });
1270 12 : assert!(
1271 12 : normalize_subscription(doc.as_object().unwrap(), &ctx, false).is_err(),
1272 : "empty notification.{key} must be rejected"
1273 : );
1274 : }
1275 4 : }
1276 :
1277 : // ---------- shared fixtures ----------
1278 :
1279 : /// The minimal valid 5.2.12 Subscription, with `extra` merged over it.
1280 736 : fn sub(extra: Value) -> Value {
1281 736 : let mut base = json!({
1282 736 : "type": "Subscription",
1283 736 : "entities": [{"type": "Building"}],
1284 736 : "notification": {"endpoint": {"uri": "http://localhost:1111/notify"}}
1285 : });
1286 736 : let obj = base.as_object_mut().expect("object");
1287 760 : for (k, v) in extra.as_object().expect("object") {
1288 760 : obj.insert(k.clone(), v.clone());
1289 760 : }
1290 736 : base
1291 736 : }
1292 :
1293 740 : fn norm(doc: &Value) -> Result<Map<String, Value>, NgsiError> {
1294 740 : normalize_subscription(
1295 740 : doc.as_object().expect("object"),
1296 740 : &Loader::new().core(),
1297 : false,
1298 : )
1299 740 : }
1300 :
1301 8 : fn frag(doc: &Value) -> Result<Map<String, Value>, NgsiError> {
1302 8 : normalize_subscription(
1303 8 : doc.as_object().expect("object"),
1304 8 : &Loader::new().core(),
1305 : true,
1306 : )
1307 8 : }
1308 :
1309 : /// 6.3.8 / 6.3.9: "each of the KeyValuePairs shall be included as a
1310 : /// custom HTTP header … 'Key' and 'value' members shall adhere to IETF
1311 : /// RFC 7230 clause 3.2 definitions concerning HTTP headers". A pair that
1312 : /// cannot be written as a header is refused at create, not at delivery:
1313 : /// a subscriber that could smuggle CR or LF into a receiverInfo value
1314 : /// would write its own headers — or its own request line — into every
1315 : /// notification the broker sends to that endpoint.
1316 : #[test]
1317 4 : fn clause_6_3_8_receiver_info_that_cannot_be_a_header_is_refused() {
1318 52 : let with = |k: &str, v: &str| {
1319 52 : sub(json!({"notification": {"endpoint": {
1320 52 : "uri": "http://localhost:1111/notify",
1321 52 : "receiverInfo": [{"key": k, "value": v}],
1322 : }}}))
1323 52 : };
1324 44 : for (k, v, why) in [
1325 4 : ("X-Ok", "a\r\nInjected: 1", "a CRLF splits the header block"),
1326 4 : (
1327 4 : "X-Ok",
1328 4 : "a\nInjected: 1",
1329 4 : "a bare LF is enough for most parsers",
1330 4 : ),
1331 4 : ("X-Ok", "a\rInjected: 1", "so is a bare CR"),
1332 4 : ("X-Ok", "a\0b", "NUL is not field-content"),
1333 4 : ("X-Ok", " leading", "OWS may not be part of the value"),
1334 4 : ("X-Ok", "trailing\t", "nor may it trail it"),
1335 4 : ("Bad Key", "v", "SP is not a tchar"),
1336 4 : ("X:Ok", "v", "neither is the field separator"),
1337 4 : ("", "v", "a field-name is one or more tchar"),
1338 4 : ("X-Ok\r\nInjected", "v", "the key is a header name too"),
1339 4 : ("X-Ünicode", "v", "field-name is ASCII"),
1340 4 : ] {
1341 44 : assert!(
1342 44 : norm(&with(k, v)).is_err(),
1343 : "receiverInfo {k:?}: {v:?} must be refused — {why}"
1344 : );
1345 : }
1346 : // and the negative: an empty value and the full tchar repertoire are
1347 : // legal headers, so they stay legal Subscriptions
1348 8 : for (k, v) in [
1349 4 : ("X-Empty", ""),
1350 4 : ("!#$%&x*+-.^_`|~0Aa", "value with spaces inside"),
1351 4 : ] {
1352 8 : assert!(norm(&with(k, v)).is_ok(), "receiverInfo {k:?}: {v:?}");
1353 : }
1354 4 : }
1355 :
1356 : /// Table 5.2.12-1: `localOnly` and `splitEntities` are Booleans. Both
1357 : /// decide how far a Subscription reaches — `localOnly=true` confines it
1358 : /// to "the Entities stored locally" (5.5.13), and `splitEntities` decides
1359 : /// whether filters may be applied locally at all. A string `"true"` read
1360 : /// through `as_bool()` is falsy, so accepting one turns a subscriber's
1361 : /// request for local scope into a distributed subscription that forwards
1362 : /// to every matching Context Source, and the subscriber is never told.
1363 : #[test]
1364 4 : fn clause_5_2_12_local_only_and_split_entities_are_booleans() {
1365 8 : for member in ["localOnly", "splitEntities"] {
1366 32 : for v in [json!("true"), json!(1), json!("false"), json!([true])] {
1367 32 : assert!(
1368 32 : norm(&sub(json!({member: v.clone()}))).is_err(),
1369 : "{member}: {v} is not a Boolean (5.2.12)"
1370 : );
1371 : }
1372 16 : for v in [json!(true), json!(false)] {
1373 16 : let out = norm(&sub(json!({member: v.clone()}))).expect("a Boolean is accepted");
1374 16 : assert_eq!(out[member], v, "{member} is stored as given");
1375 : }
1376 : }
1377 4 : }
1378 :
1379 : /// Table 5.2.12-1: `expandValues` and `jsonKeys` are each a String,
1380 : /// "comma separated list of attribute names". They decide whether the
1381 : /// notification condition compares a term's value as written or after
1382 : /// JSON-LD type coercion (4.9), and a non-string reads as an absent
1383 : /// list at matching time — the Subscription would then quietly match on
1384 : /// the other reading instead of refusing the member.
1385 : #[test]
1386 4 : fn clause_5_2_12_expand_values_and_json_keys_are_strings() {
1387 8 : for member in ["expandValues", "jsonKeys"] {
1388 32 : for v in [json!(1), json!(["category"]), json!(true), json!({})] {
1389 32 : assert!(
1390 32 : norm(&sub(json!({member: v.clone()}))).is_err(),
1391 : "{member}: {v} is not a comma separated list (5.2.12)"
1392 : );
1393 : }
1394 8 : let v = json!("category,brandName");
1395 8 : let out = norm(&sub(json!({member: v.clone()}))).expect("a String is accepted");
1396 8 : assert_eq!(out[member], v, "{member} is stored as given");
1397 : }
1398 4 : }
1399 :
1400 : // ---------- 5.2.12 read-only and internal members ----------
1401 :
1402 : /// 5.8.6: the @context governing a Subscription's notifications is the
1403 : /// @context of the creating request, kept in a broker-internal member.
1404 : /// A Context Subscriber can neither set it (create) nor replace it
1405 : /// (update), and it appears in no served representation — 5.8.3 and
1406 : /// 5.8.4 serve the 5.2.12 data type, which has no such member.
1407 : #[test]
1408 4 : fn clause_5_8_6_internal_context_member_is_client_proof() {
1409 4 : let hostile = json!({"__context": "http://attacker.invalid/ctx.jsonld",
1410 4 : "__via": "1.1 forged-alias"});
1411 4 : let created = norm(&sub(hostile.clone())).expect("valid subscription");
1412 4 : assert!(
1413 4 : !created.contains_key("__context"),
1414 : "a client-supplied __context must not reach storage: {created:?}"
1415 : );
1416 4 : assert!(
1417 4 : !created.contains_key("__via"),
1418 : "a body member must not forge the Via chain — the chain comes \
1419 : from the HTTP header only: {created:?}"
1420 : );
1421 4 : let patched = frag(&hostile).expect("valid fragment");
1422 4 : assert!(
1423 4 : !patched.contains_key("__context"),
1424 : "a patch fragment must not replace the notification @context: {patched:?}"
1425 : );
1426 4 : assert!(
1427 4 : !patched.contains_key("__via"),
1428 : "a patch fragment must not rewrite the stored Via chain: {patched:?}"
1429 : );
1430 4 : let ctx = Loader::new().core();
1431 4 : let stored = json!({
1432 4 : "id": "urn:ngsi-ld:Subscription:ctx",
1433 4 : "type": "Subscription",
1434 4 : "entities": [{"type": "Building"}],
1435 4 : "notification": {"endpoint": {"uri": "http://localhost:1111/notify"}},
1436 4 : "__context": "https://example.org/private-ctx.jsonld",
1437 4 : "__via": "1.1 upstream-broker",
1438 : });
1439 8 : for csource in [false, true] {
1440 16 : for sys in [false, true] {
1441 16 : let out = present_subscription(&stored, &ctx, sys, csource);
1442 16 : assert!(
1443 16 : !out.to_string().contains("__context"),
1444 : "served representation leaked the internal @context member: {out}"
1445 : );
1446 16 : assert!(
1447 16 : !out.to_string().contains("private-ctx"),
1448 : "served representation leaked the internal @context value: {out}"
1449 : );
1450 16 : assert!(
1451 16 : !out.to_string().contains("__via")
1452 16 : && !out.to_string().contains("upstream-broker"),
1453 : "5.8.3/5.8.4 serve the 5.2.12 data type, which has no Via member: {out}"
1454 : );
1455 : }
1456 : }
1457 4 : }
1458 :
1459 : /// Table 5.2.12-2 and 5.2.14.2: read-only members "shall not be provided
1460 : /// by Context Subscribers. In the event that they are provided (in
1461 : /// update or create operations) NGSI-LD implementations shall ignore
1462 : /// them."
1463 : #[test]
1464 4 : fn clause_5_2_12_read_only_members_are_ignored() {
1465 4 : let doc = sub(json!({
1466 4 : "status": "active",
1467 4 : "createdAt": "2020-01-01T00:00:00Z",
1468 4 : "modifiedAt": "2020-01-01T00:00:00Z",
1469 4 : "notification": {
1470 4 : "endpoint": {"uri": "http://localhost:1111/notify"},
1471 4 : "timesSent": 42,
1472 4 : "lastNotification": "2020-01-01T00:00:00Z",
1473 4 : "lastSuccess": "2020-01-01T00:00:00Z",
1474 4 : "lastFailure": "2020-01-01T00:00:00Z",
1475 : }
1476 : }));
1477 8 : for is_patch in [false, true] {
1478 8 : let n = normalize_subscription(
1479 8 : doc.as_object().expect("object"),
1480 8 : &Loader::new().core(),
1481 8 : is_patch,
1482 : )
1483 8 : .expect("valid");
1484 24 : for k in ["status", "createdAt", "modifiedAt"] {
1485 24 : assert!(
1486 24 : !n.contains_key(k),
1487 : "{k} must not persist (patch={is_patch})"
1488 : );
1489 : }
1490 8 : let notif = n["notification"].as_object().expect("notification");
1491 32 : for k in [
1492 8 : "timesSent",
1493 8 : "lastNotification",
1494 8 : "lastSuccess",
1495 8 : "lastFailure",
1496 8 : ] {
1497 32 : assert!(
1498 32 : !notif.contains_key(k),
1499 : "notification.{k} must not persist (patch={is_patch})"
1500 : );
1501 : }
1502 : }
1503 4 : }
1504 :
1505 : // ---------- 5.2.12 value spaces ----------
1506 :
1507 : /// Table 5.2.12-1: "Valid notification triggers are entityCreated,
1508 : /// entityUpdated, entityDeleted, attributeCreated, attributeUpdated,
1509 : /// attributeDeleted". A trigger outside that set would leave a
1510 : /// subscription that silently never fires.
1511 : #[test]
1512 4 : fn clause_5_2_12_notification_trigger_value_space() {
1513 12 : for good in [
1514 4 : json!(["entityCreated"]),
1515 4 : json!(["entityUpdated", "entityDeleted"]),
1516 4 : json!(["attributeCreated", "attributeUpdated", "attributeDeleted"]),
1517 4 : ] {
1518 12 : assert!(
1519 12 : norm(&sub(json!({ "notificationTrigger": good }))).is_ok(),
1520 : "{good} must be accepted"
1521 : );
1522 : }
1523 20 : for bad in [
1524 4 : json!(["entityChanged"]),
1525 4 : json!(["attributeCreated", "nope"]),
1526 4 : json!("entityCreated"),
1527 4 : json!([1]),
1528 4 : json!({}),
1529 4 : ] {
1530 20 : assert!(
1531 20 : norm(&sub(json!({ "notificationTrigger": bad }))).is_err(),
1532 : "{bad} must be rejected"
1533 : );
1534 : }
1535 4 : }
1536 :
1537 : /// Table 5.2.12-1: csf is "A valid query string as per clause 4.9". An
1538 : /// unparseable filter is otherwise stored and only fails at Context
1539 : /// Source Registration matching time (5.11.2.4), where it silently
1540 : /// matches nothing.
1541 : #[test]
1542 4 : fn clause_5_2_12_csf_must_be_a_valid_query() {
1543 4 : assert!(norm(&sub(json!({"csf": "endpoint==\"http://a/x\""}))).is_ok());
1544 16 : for bad in [json!("(("), json!("a==(("), json!(5), json!(["a==1"])] {
1545 16 : assert!(
1546 16 : norm(&sub(json!({ "csf": bad }))).is_err(),
1547 : "{bad} must be rejected"
1548 : );
1549 : }
1550 4 : }
1551 :
1552 : /// Table 5.2.12-1: throttling and timeInterval are Numbers "Greater
1553 : /// than 0"; throttling allows fractional values, and neither accepts a
1554 : /// stringified number. The table sets no lower bound on `timeInterval`
1555 : /// beyond that, so a sub-second one is a legal Subscription and is
1556 : /// rejected by nothing.
1557 : #[test]
1558 4 : fn clause_5_2_12_throttling_and_time_interval_value_space() {
1559 4 : assert!(norm(&sub(json!({"throttling": 0.5}))).is_ok());
1560 4 : assert!(norm(&sub(json!({"timeInterval": 5}))).is_ok());
1561 4 : assert!(norm(&sub(json!({"timeInterval": 0.5}))).is_ok());
1562 20 : for bad in [json!(0), json!(-1), json!("5"), json!(true), json!(null)] {
1563 20 : assert!(
1564 20 : norm(&sub(json!({ "throttling": bad }))).is_err(),
1565 : "throttling {bad} must be rejected"
1566 : );
1567 20 : assert!(
1568 20 : norm(&sub(json!({ "timeInterval": bad }))).is_err(),
1569 : "timeInterval {bad} must be rejected"
1570 : );
1571 : }
1572 4 : }
1573 :
1574 : /// Table 5.2.12-1: expiresAt is a 4.6.3 DateTime, and 5.8.1.4/5.8.2.4
1575 : /// reject one "referring to a DateTime in the past". 4.6.3 admits both
1576 : /// fraction separators, so the past/future decision must be taken on the
1577 : /// instant, not on the raw string.
1578 : #[test]
1579 4 : fn clause_5_2_12_expires_at_boundary() {
1580 4 : assert!(norm(&sub(json!({"expiresAt": "2099-01-01T00:00:00Z"}))).is_ok());
1581 : // comma is the other legal fraction separator (4.6.3); comparing the
1582 : // raw strings would place ',' before now_iso()'s '.' and reject it
1583 4 : assert!(
1584 4 : norm(&sub(json!({"expiresAt": "2099-01-01T00:00:00,500Z"}))).is_ok(),
1585 : "a comma fraction separator is legal (4.6.3)"
1586 : );
1587 4 : let soon = (chrono::Utc::now() + chrono::Duration::seconds(30))
1588 4 : .to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
1589 4 : assert!(
1590 4 : norm(&sub(json!({ "expiresAt": soon }))).is_ok(),
1591 : "a whole-second DateTime 30s ahead is in the future: {soon}"
1592 : );
1593 20 : for bad in [
1594 4 : json!("2020-01-01T00:00:00Z"),
1595 4 : json!("tomorrow"),
1596 4 : json!("2099-01-01"),
1597 4 : json!("2099-01-01T00:00:00+05:00"),
1598 4 : json!(1234),
1599 4 : ] {
1600 20 : assert!(
1601 20 : norm(&sub(json!({ "expiresAt": bad }))).is_err(),
1602 : "expiresAt {bad} must be rejected"
1603 : );
1604 : }
1605 4 : }
1606 :
1607 : /// Table 5.2.12-1: watchedAttributes is a String[] of Attribute names,
1608 : /// "Empty array (0 length) is not allowed"; names expand per 5.5.7.
1609 : #[test]
1610 4 : fn clause_5_2_12_watched_attributes_contract() {
1611 4 : let n = norm(&sub(json!({"watchedAttributes": ["temperature"]}))).expect("valid");
1612 4 : assert_eq!(
1613 4 : n["watchedAttributes"][0],
1614 : "https://uri.etsi.org/ngsi-ld/default-context/temperature"
1615 : );
1616 16 : for bad in [json!([]), json!("temperature"), json!([1]), json!([""])] {
1617 16 : assert!(
1618 16 : norm(&sub(json!({ "watchedAttributes": bad }))).is_err(),
1619 : "watchedAttributes {bad} must be rejected"
1620 : );
1621 : }
1622 4 : }
1623 :
1624 : /// Table 5.2.12-1: q is "A valid query string as per clause 4.9". The
1625 : /// matcher percent-decodes before parsing (4.9), so the DECODED form is
1626 : /// what must be validated here — otherwise an encoded paren bomb passes
1627 : /// creation and only unfolds inside the notification task.
1628 : #[test]
1629 4 : fn clause_5_2_12_q_is_validated_in_its_decoded_form() {
1630 4 : assert!(norm(&sub(json!({"q": "temperature>20"}))).is_ok());
1631 4 : assert!(norm(&sub(json!({"q": "(("}))).is_err());
1632 4 : assert!(
1633 4 : norm(&sub(json!({"q": "%28%28%28"}))).is_err(),
1634 : "a percent-encoded unbalanced expression must be rejected at creation"
1635 : );
1636 4 : assert!(norm(&sub(json!({"q": 5}))).is_err());
1637 4 : }
1638 :
1639 : /// Table 5.2.12-1: "At least one of (a) entities or (b)
1640 : /// watchedAttributes shall be present, unless the member localOnly is
1641 : /// set to true"; timeInterval excludes watchedAttributes and throttling.
1642 : #[test]
1643 4 : fn clause_5_2_12_mutual_exclusions_and_local_only() {
1644 4 : let bare = json!({
1645 4 : "type": "Subscription",
1646 4 : "notification": {"endpoint": {"uri": "http://localhost:1111/notify"}}
1647 : });
1648 4 : assert!(
1649 4 : norm(&bare).is_err(),
1650 : "neither entities nor watchedAttributes"
1651 : );
1652 4 : let mut local = bare.clone();
1653 4 : local["localOnly"] = json!(true);
1654 4 : assert!(
1655 4 : norm(&local).is_ok(),
1656 : "localOnly=true waives the rule (5.5.13)"
1657 : );
1658 4 : assert!(norm(&sub(json!({
1659 4 : "timeInterval": 5,
1660 4 : "watchedAttributes": ["temperature"]
1661 4 : })))
1662 4 : .is_err());
1663 4 : assert!(norm(&sub(json!({"timeInterval": 5, "throttling": 5}))).is_err());
1664 : // a fragment carries no mandatory members of its own (5.8.2.4)
1665 4 : assert!(frag(&json!({"isActive": false})).is_ok());
1666 4 : }
1667 :
1668 : /// Table 5.2.33-1 EntitySelector: type is required, id is "String or
1669 : /// String[]" of valid URIs, idPattern is a regular expression; 4.17
1670 : /// type-selection expressions stay unexpanded.
1671 : #[test]
1672 4 : fn clause_5_2_33_entity_selector_contract() {
1673 4 : let n = norm(&sub(json!({"entities": [{"type": "Building|Room"}]}))).expect("valid");
1674 4 : assert_eq!(
1675 4 : n["entities"][0]["type"], "Building|Room",
1676 : "a 4.17 type-selection expression is evaluated at match time"
1677 : );
1678 4 : let n = norm(&sub(json!({
1679 4 : "entities": [{"type": "Building", "id": ["urn:ngsi-ld:B:1", "urn:ngsi-ld:B:2"]}]
1680 4 : })))
1681 4 : .expect("valid");
1682 4 : assert_eq!(n["entities"][0]["id"][1], "urn:ngsi-ld:B:2");
1683 40 : for bad in [
1684 4 : json!([]),
1685 4 : json!("Building"),
1686 4 : json!([{"id": "urn:ngsi-ld:B:1"}]),
1687 4 : json!([{"type": ""}]),
1688 4 : json!([{"type": "Building", "id": "not a uri"}]),
1689 4 : json!([{"type": "Building", "id": ["urn:ngsi-ld:B:1", 7]}]),
1690 4 : json!([{"type": "Building", "idPattern": "["}]),
1691 4 : // 21 bytes, a 16 MiB automaton: above the compile ceiling the
1692 4 : // pattern is refused here rather than compiled per event
1693 4 : json!([{"type": "Building", "idPattern": r"(?:\p{Any}{100}){100}"}]),
1694 4 : json!([{"type": "Building", "idPattern": 7}]),
1695 4 : json!(["Building"]),
1696 4 : ] {
1697 40 : assert!(
1698 40 : norm(&sub(json!({ "entities": bad }))).is_err(),
1699 : "entities {bad} must be rejected"
1700 : );
1701 : }
1702 4 : }
1703 :
1704 : // ---------- 5.2.14 / 5.2.15 notification parameters ----------
1705 :
1706 : /// Table 5.2.15-1 Endpoint: uri is mandatory and a valid URI; cooldown
1707 : /// and timeout are Numbers "Greater than 0"; accept is one of the three
1708 : /// media types. Which schemes are deliverable is the sink registry's
1709 : /// question, answered by `check_endpoint` where the state is in hand.
1710 : #[test]
1711 4 : fn clause_5_2_15_endpoint_contract() {
1712 132 : let mk = |ep: Value| sub(json!({"notification": {"endpoint": ep}}));
1713 4 : assert!(
1714 4 : norm(&sub(json!({"notification": {}}))).is_err(),
1715 : "endpoint required"
1716 : );
1717 4 : assert!(norm(&mk(json!({}))).is_err(), "endpoint.uri required");
1718 4 : assert!(norm(&mk(json!({"uri": "no-scheme"}))).is_err());
1719 4 : assert!(norm(&mk(json!({"uri": "http://a/x\r\nX: y"}))).is_err());
1720 8 : for key in ["cooldown", "timeout"] {
1721 24 : for bad in [json!(0), json!(-1), json!("5")] {
1722 24 : assert!(
1723 24 : norm(&mk(json!({"uri": "http://a/x", key: bad}))).is_err(),
1724 : "endpoint.{key} {bad} must be rejected"
1725 : );
1726 : }
1727 8 : assert!(norm(&mk(json!({"uri": "http://a/x", key: 1.5}))).is_ok());
1728 : }
1729 8 : for key in ["receiverInfo", "notifierInfo"] {
1730 24 : for bad in [
1731 8 : json!({}),
1732 8 : json!([{"key": "k"}]),
1733 8 : json!([{"key": 1, "value": "v"}]),
1734 8 : ] {
1735 24 : assert!(
1736 24 : norm(&mk(json!({"uri": "http://a/x", key: bad}))).is_err(),
1737 : "endpoint.{key} {bad} must be rejected"
1738 : );
1739 : }
1740 8 : assert!(norm(&mk(
1741 8 : json!({"uri": "http://a/x", key: [{"key": "k", "value": "v"}]})
1742 8 : ))
1743 8 : .is_ok());
1744 : }
1745 : // 6.3.8/6.3.9: each receiverInfo pair becomes one custom HTTP header,
1746 : // and "'Key' and 'value' members shall adhere to IETF RFC 7230 ...
1747 : // definitions concerning HTTP headers" — so a pair that cannot be a
1748 : // header is input the operation cannot meet, not a delivery that fails
1749 : // later.
1750 32 : for bad in [
1751 4 : json!([{"key": "", "value": "v"}]),
1752 4 : json!([{"key": "Bad Key", "value": "v"}]),
1753 4 : json!([{"key": "X:Y", "value": "v"}]),
1754 4 : json!([{"key": "X\r\nInjected", "value": "v"}]),
1755 4 : json!([{"key": "X", "value": "a\r\nInjected: 1"}]),
1756 4 : json!([{"key": "X", "value": "tab\u{7f}del"}]),
1757 4 : json!([{"key": "X", "value": " leading"}]),
1758 4 : json!([{"key": "X", "value": "trailing "}]),
1759 4 : ] {
1760 32 : assert!(
1761 32 : norm(&mk(json!({"uri": "http://a/x", "receiverInfo": bad}))).is_err(),
1762 : "receiverInfo {bad} must be rejected"
1763 : );
1764 : }
1765 16 : for ok in [
1766 4 : json!([{"key": "Authorization", "value": "Bearer t"}]),
1767 4 : json!([{"key": "X-Custom_1!", "value": ""}]),
1768 4 : json!([{"key": "Prefer", "value": "body=json"}]),
1769 4 : json!([{"key": "X", "value": "a\tb"}]),
1770 4 : ] {
1771 16 : assert!(
1772 16 : norm(&mk(json!({"uri": "http://a/x", "receiverInfo": ok}))).is_ok(),
1773 : "receiverInfo {ok} must be accepted"
1774 : );
1775 : }
1776 4 : assert!(norm(&mk(json!({"uri": "http://a/x", "accept": "text/html"}))).is_err());
1777 4 : assert!(norm(&mk(
1778 4 : json!({"uri": "http://a/x", "accept": "application/geo+json"})
1779 4 : ))
1780 4 : .is_ok());
1781 4 : }
1782 :
1783 : /// Table 5.2.14.1-1 NotificationParams: format value space, join /
1784 : /// joinLevel, boolean sysAttrs/showChanges, and the pick/omit/attributes
1785 : /// exclusivity — "A synonym for pick, except that id, type, scope are
1786 : /// not allowed."
1787 : #[test]
1788 4 : fn clause_5_2_14_notification_params_contract() {
1789 128 : let mk = |extra: Value| {
1790 128 : let mut n = json!({"endpoint": {"uri": "http://localhost:1111/notify"}});
1791 128 : let o = n.as_object_mut().expect("object");
1792 144 : for (k, v) in extra.as_object().expect("object") {
1793 144 : o.insert(k.clone(), v.clone());
1794 144 : }
1795 128 : sub(json!({ "notification": n }))
1796 128 : };
1797 4 : assert!(
1798 4 : norm(&sub(json!({"notification": []}))).is_err(),
1799 : "not an object"
1800 : );
1801 4 : assert!(norm(&mk(json!({"format": "verbose"}))).is_err());
1802 16 : for f in ["normalized", "keyValues", "simplified", "concise"] {
1803 16 : assert!(norm(&mk(json!({ "format": f }))).is_ok(), "{f}");
1804 : }
1805 12 : for bad in [json!("nested"), json!(1), json!("")] {
1806 12 : assert!(norm(&mk(json!({ "join": bad }))).is_err(), "join {bad}");
1807 : }
1808 12 : for good in ["flat", "inline", "@none"] {
1809 12 : assert!(norm(&mk(json!({ "join": good }))).is_ok(), "{good}");
1810 : }
1811 : // Table 5.2.14.1-1 restricts joinLevel to a positive integer, and the
1812 : // depth it names is the same Linked Entity traversal the query
1813 : // parameter drives — so the ceiling this deployment publishes as
1814 : // maxJoinLevel bounds it on both surfaces, not on the query alone.
1815 4 : let cap = crate::bounds::MAX_JOIN_LEVEL;
1816 24 : for bad in [
1817 4 : json!(0),
1818 4 : json!(-1),
1819 4 : json!("2"),
1820 4 : json!(1.5),
1821 4 : json!(cap + 1),
1822 4 : json!(u64::MAX),
1823 4 : ] {
1824 24 : assert!(
1825 24 : norm(&mk(json!({ "joinLevel": bad }))).is_err(),
1826 : "joinLevel {bad}"
1827 : );
1828 : }
1829 4 : assert!(norm(&mk(json!({"joinLevel": 1}))).is_ok());
1830 4 : assert!(norm(&mk(json!({ "joinLevel": cap }))).is_ok(), "at the cap");
1831 8 : for key in ["sysAttrs", "showChanges"] {
1832 8 : assert!(
1833 8 : norm(&mk(json!({ key: "true" }))).is_err(),
1834 : "{key} must be a boolean"
1835 : );
1836 8 : assert!(norm(&mk(json!({ key: true }))).is_ok());
1837 : }
1838 12 : for name in ["id", "type", "scope"] {
1839 12 : assert!(
1840 12 : norm(&mk(json!({"attributes": [name]}))).is_err(),
1841 : "notification.attributes may not name {name}"
1842 : );
1843 : }
1844 4 : assert!(norm(&mk(json!({"attributes": [1]}))).is_err());
1845 4 : assert!(norm(&mk(json!({"attributes": ["a"], "pick": ["b"]}))).is_err());
1846 4 : assert!(norm(&mk(json!({"attributes": ["a"], "omit": ["b"]}))).is_err());
1847 4 : assert!(norm(&mk(json!({"pick": ["a"], "omit": ["a"]}))).is_err());
1848 4 : assert!(norm(&mk(json!({"pick": ["a"], "omit": ["b"]}))).is_ok());
1849 : // attribute names expand per 5.5.7
1850 4 : let n = norm(&mk(json!({"attributes": ["temperature"]}))).expect("valid");
1851 4 : assert_eq!(
1852 4 : n["notification"]["attributes"][0],
1853 : "https://uri.etsi.org/ngsi-ld/default-context/temperature"
1854 : );
1855 4 : }
1856 :
1857 : /// Table 5.2.14.1-1: a `pick` or `omit` member is "a valid attribute
1858 : /// projection language string as per clause 4.21". The notification path
1859 : /// parses the member again at delivery time and drops one it cannot
1860 : /// parse, so a Subscription accepted with an unparseable `omit` delivers
1861 : /// the Attribute the subscriber asked to have removed — the failure is
1862 : /// silent and it is in the leaking direction. The value space belongs at
1863 : /// the door, where 5.8.1.4 raises BadRequestData for a Subscription that
1864 : /// does not validate.
1865 : #[test]
1866 4 : fn clause_5_2_14_pick_and_omit_members_are_projection_language() {
1867 56 : let mk = |extra: Value| {
1868 56 : let mut n = json!({"endpoint": {"uri": "http://localhost:1111/notify"}});
1869 56 : let o = n.as_object_mut().expect("object");
1870 56 : for (k, v) in extra.as_object().expect("object") {
1871 56 : o.insert(k.clone(), v.clone());
1872 56 : }
1873 56 : sub(json!({ "notification": n }))
1874 56 : };
1875 8 : for key in ["pick", "omit"] {
1876 48 : for term in [
1877 8 : json!("street address"), // a space is outside the 4.21 character set
1878 8 : json!("{model"), // unbalanced brace
1879 8 : json!("a,,b"), // empty projection member
1880 8 : json!(""), // empty term
1881 8 : json!("{model}"), // a LinkedEntityTerm with no AttrName
1882 8 : json!(42), // not a String
1883 8 : ] {
1884 48 : assert!(
1885 48 : norm(&mk(json!({ key: [term] }))).is_err(),
1886 : "notification.{key} member {term} is not 4.21 projection language"
1887 : );
1888 : }
1889 8 : assert!(
1890 8 : norm(&mk(json!({ key: ["refDevice{model}", "temperature"] }))).is_ok(),
1891 : "a nested term and a bare term are both valid 4.21 (notification.{key})"
1892 : );
1893 : // 5.8.2.4 replaces the member from a fragment, so the same value
1894 : // space has to hold on that way in: the delivery path cannot tell
1895 : // which operation wrote the Subscription it reads. The fragment
1896 : // carries a complete endpoint deliberately — `notification`
1897 : // without one is refused for the missing endpoint (5.2.14,
1898 : // cardinality 1), which would make this assertion pass while
1899 : // proving nothing about the projection member.
1900 8 : let ctx = antares_jsonld::Loader::new().core();
1901 8 : let fragment = json!({"notification": {
1902 8 : "endpoint": {"uri": "http://localhost:1111/notify"}, key: ["a b"]}});
1903 8 : assert!(
1904 8 : normalize_subscription(fragment.as_object().expect("object"), &ctx, true).is_err(),
1905 : "an update fragment wrote a notification.{key} the 4.21 grammar refuses"
1906 : );
1907 : }
1908 4 : }
1909 :
1910 : /// Table 5.2.13-1 GeoQuery: georel is mandatory and the geoproperty
1911 : /// name expands per 5.5.7.
1912 : #[test]
1913 4 : fn clause_5_2_13_geo_q_contract() {
1914 4 : let ok = json!({
1915 4 : "georel": "near;maxDistance==2000",
1916 4 : "geometry": "Point",
1917 4 : "coordinates": [-8.5, 41.2],
1918 4 : "geoproperty": "location"
1919 : });
1920 4 : let n = norm(&sub(json!({ "geoQ": ok }))).expect("valid");
1921 4 : assert_eq!(
1922 4 : n["geoQ"]["geoproperty"],
1923 : "https://uri.etsi.org/ngsi-ld/location"
1924 : );
1925 12 : for bad in [
1926 4 : json!("near"),
1927 4 : json!({"geometry": "Point", "coordinates": [1, 2]}),
1928 4 : json!({"georel": "sideways", "geometry": "Point", "coordinates": [1, 2]}),
1929 4 : ] {
1930 12 : assert!(
1931 12 : norm(&sub(json!({ "geoQ": bad }))).is_err(),
1932 : "geoQ {bad} must be rejected"
1933 : );
1934 : }
1935 4 : }
1936 :
1937 : /// 5.2.21 TemporalQuery, used by Context Source Registration
1938 : /// Subscriptions (5.11): timerel and timeAt are cardinality 1.
1939 : #[test]
1940 4 : fn clause_5_2_21_temporal_q_contract() {
1941 4 : assert!(norm(&sub(json!({
1942 4 : "temporalQ": {"timerel": "after", "timeAt": "2020-01-01T00:00:00Z"}
1943 4 : })))
1944 4 : .is_ok());
1945 16 : for bad in [
1946 4 : json!("after"),
1947 4 : json!({"timerel": "after"}),
1948 4 : json!({"timerel": "sideways", "timeAt": "2020-01-01T00:00:00Z"}),
1949 4 : json!({"timerel": "after", "timeAt": "yesterday"}),
1950 4 : ] {
1951 16 : assert!(
1952 16 : norm(&sub(json!({ "temporalQ": bad }))).is_err(),
1953 : "temporalQ {bad} must be rejected"
1954 : );
1955 : }
1956 4 : }
1957 :
1958 : // ---------- 5.8.3 / 5.8.4 presentation ----------
1959 :
1960 : /// 5.2.12 Table 5.2.12-2: status is "Provided by the system"; 5.8.2.4
1961 : /// fixes its value space to active | paused | expired. The default
1962 : /// notificationTrigger is surfaced for entity Subscriptions only, and
1963 : /// createdAt/modifiedAt stay behind the sysAttrs gate.
1964 : #[test]
1965 4 : fn clause_5_8_3_presented_subscription_shape() {
1966 4 : let ctx = Loader::new().core();
1967 4 : let base = json!({
1968 4 : "id": "urn:ngsi-ld:Subscription:p1",
1969 4 : "type": "Subscription",
1970 4 : "entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Building"}],
1971 4 : "watchedAttributes": ["https://uri.etsi.org/ngsi-ld/default-context/temperature"],
1972 4 : "notification": {
1973 4 : "endpoint": {"uri": "http://localhost:1111/notify"},
1974 4 : "attributes": ["https://uri.etsi.org/ngsi-ld/default-context/temperature"]
1975 : },
1976 4 : "geoQ": {"georel": "near;maxDistance==1", "geoproperty": "https://uri.etsi.org/ngsi-ld/location"},
1977 4 : "createdAt": "2020-01-01T00:00:00Z",
1978 4 : "modifiedAt": "2020-01-01T00:00:00Z",
1979 : });
1980 4 : let out = present_subscription(&base, &ctx, false, false);
1981 4 : assert_eq!(out["status"], "active");
1982 4 : assert_eq!(out["entities"][0]["type"], "Building");
1983 4 : assert_eq!(out["watchedAttributes"][0], "temperature");
1984 4 : assert_eq!(out["notification"]["attributes"][0], "temperature");
1985 4 : assert_eq!(out["geoQ"]["geoproperty"], "location");
1986 4 : assert!(
1987 4 : out.get("createdAt").is_none() && out.get("modifiedAt").is_none(),
1988 : "sysAttrs are gated (6.3.11): {out}"
1989 : );
1990 4 : assert_eq!(
1991 4 : out["notificationTrigger"],
1992 4 : json!(["attributeCreated", "attributeUpdated"])
1993 : );
1994 4 : let sys = present_subscription(&base, &ctx, true, false);
1995 4 : assert_eq!(sys["createdAt"], "2020-01-01T00:00:00Z");
1996 : // a Context Source Registration Subscription has no such default
1997 4 : let cs = present_subscription(&base, &ctx, false, true);
1998 4 : assert!(cs.get("notificationTrigger").is_none(), "{cs}");
1999 :
2000 4 : let mut paused = base.clone();
2001 4 : paused["isActive"] = json!(false);
2002 4 : assert_eq!(
2003 4 : present_subscription(&paused, &ctx, false, false)["status"],
2004 : "paused"
2005 : );
2006 4 : let mut expired = base.clone();
2007 4 : expired["expiresAt"] = json!("2020-01-01T00:00:00Z");
2008 4 : expired["isActive"] = json!(false);
2009 4 : assert_eq!(
2010 4 : present_subscription(&expired, &ctx, false, false)["status"],
2011 : "expired",
2012 : "expiry wins over paused (5.8.2.4)"
2013 : );
2014 4 : let mut periodic = base.clone();
2015 4 : periodic["timeInterval"] = json!(30);
2016 4 : assert!(
2017 4 : present_subscription(&periodic, &ctx, false, false)
2018 4 : .get("notificationTrigger")
2019 4 : .is_none(),
2020 : "a periodic subscription has no attribute triggers"
2021 : );
2022 4 : }
2023 : }
|
