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