Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Notification delivery.
3 : //!
4 : //! A sink serves one family of `endpoint.uri` schemes (6.3.8, and 7.2 for
5 : //! the optional MQTT binding): it validates its own endpoints at
6 : //! subscription creation and delivers the prepared notification. The
7 : //! registry keys sinks by scheme and is the only way one is chosen — a
8 : //! scheme it does not hold is rejected at creation, never delivered through
9 : //! another binding. Sinks: http/reqwest, mqtt/rumqttc behind the `mqtt`
10 : //! feature.
11 : //!
12 : //! The egress policy (allowlist, private-range deny, per-destination
13 : //! breakers) runs in the caller before `deliver`, so a sink registered from
14 : //! outside this workspace cannot step around it.
15 : #![cfg_attr(not(test), warn(clippy::expect_used))]
16 :
17 : use antares_model::NgsiError;
18 : use serde_json::{Map, Value};
19 : use std::future::Future;
20 : use std::pin::Pin;
21 : use std::time::Duration;
22 :
23 : pub mod http;
24 : #[cfg(feature = "mqtt")]
25 : pub mod mqtt;
26 :
27 : pub use http::HttpSink;
28 :
29 : /// The TS 104 243 5.2 message (the MQTT binding's 7.2):
30 : /// `{"metadata": {...}, "body": notification}`. Every binding that writes
31 : /// this envelope (MQTT, an in-process socket binding) renders it
32 : /// here, so two bindings cannot drift apart.
33 : /// `link` is the HTTP-Link-header-formatted @context reference; per Table
34 : /// 7.2-2 it is included only when the Content-Type is application/json
35 : /// (with ld+json the @context travels in the body).
36 6 : pub fn build_message(
37 6 : body: &Value,
38 6 : content_type: &str,
39 6 : link: Option<&str>,
40 6 : receiver_info: &[(String, String)],
41 6 : ) -> Value {
42 6 : let mut metadata = Map::new();
43 : // receiverInfo goes in first: 7.2 adds its entries "additionally", while
44 : // Table 7.2-2 names endpoint.accept as the source of Content-Type and
45 : // the served @context as the source of Link. A pair keyed like one of
46 : // those two is an entry, never that parameter's value.
47 8 : for (k, v) in receiver_info {
48 8 : metadata.insert(k.clone(), Value::String(v.clone()));
49 8 : }
50 6 : metadata.insert("Content-Type".into(), Value::String(content_type.into()));
51 6 : if content_type == "application/json" {
52 4 : if let Some(l) = link {
53 4 : metadata.insert("Link".into(), Value::String(l.to_owned()));
54 4 : }
55 2 : }
56 6 : let mut msg = Map::new();
57 6 : msg.insert("metadata".into(), Value::Object(metadata));
58 6 : msg.insert("body".into(), body.clone());
59 6 : Value::Object(msg)
60 6 : }
61 :
62 : /// One prepared notification, deliverable any number of times: a retry or a
63 : /// dead-letter replay renders the identical message from the same parts.
64 : /// The parts are transport-neutral — 6.3.8 turns them into HTTP headers,
65 : /// Table 7.2-2 into the MQTT message's `metadata` object.
66 : #[derive(Clone, Debug, PartialEq)]
67 : pub struct Outbound {
68 : /// The Notification (5.3.1) to deliver.
69 : pub body: Value,
70 : /// `endpoint.accept` (Table 5.2.15-1): the MIME type of `body`.
71 : pub accept: String,
72 : /// The JSON-LD `@context` Link value belonging to `body` (6.3.8).
73 : pub link: String,
74 : /// `endpoint.receiverInfo` (Table 5.2.15-1) followed by the tenant and
75 : /// snapshot markers the binding has to convey (6.3.22).
76 : pub receiver_info: Vec<(String, String)>,
77 : /// `endpoint.notifierInfo` (Table 5.2.15-1): the parameters the binding
78 : /// needs to set up its channel, e.g. Table 7.2-1's MQTT-QoS. Opaque to
79 : /// every sink but the one whose scheme the endpoint names.
80 : pub notifier_info: Vec<(String, String)>,
81 : }
82 :
83 : impl Outbound {
84 : /// A dead letter back into deliverable form. Letters written before the
85 : /// bindings moved behind the registry carry a rendered HTTP header list, or
86 : /// an already-wrapped clause 7 message, instead of the endpoint members they
87 : /// were rendered from; both read back, so an upgrade does not strand the
88 : /// letters an operator has not replayed yet.
89 22 : pub fn from_dead_letter(letter: &Value) -> Result<Self, String> {
90 22 : let pairs = |v: &Value| -> Vec<(String, String)> {
91 4 : serde_json::from_value::<Vec<(String, String)>>(v.clone()).unwrap_or_default()
92 4 : };
93 22 : if letter.get("accept").and_then(Value::as_str).is_some() {
94 2 : return Ok(Self {
95 2 : body: letter["payload"].clone(),
96 2 : accept: letter["accept"].as_str().unwrap_or_default().to_owned(),
97 2 : link: letter["link"].as_str().unwrap_or_default().to_owned(),
98 2 : receiver_info: pairs(&letter["receiverInfo"]),
99 2 : notifier_info: pairs(&letter["notifierInfo"]),
100 2 : });
101 20 : }
102 : // Pre-registry HTTP: Content-Type and Link ARE the accept and link they
103 : // were rendered from; every other header came from receiverInfo.
104 20 : if let Some(headers) = letter.get("headers").filter(|h| !h.is_null()) {
105 16 : let headers = serde_json::from_value::<Vec<(String, String)>>(headers.clone())
106 16 : .map_err(|e| format!("dead letter headers unreadable: {e}"))?;
107 28 : let take = |name: &str| {
108 28 : headers
109 28 : .iter()
110 42 : .find(|(k, _)| k.eq_ignore_ascii_case(name))
111 28 : .map(|(_, v)| v.clone())
112 28 : .unwrap_or_default()
113 28 : };
114 : return Ok(Self {
115 14 : body: letter["payload"].clone(),
116 14 : accept: take("Content-Type"),
117 14 : link: take("Link"),
118 14 : receiver_info: headers
119 14 : .iter()
120 30 : .filter(|(k, _)| {
121 30 : !k.eq_ignore_ascii_case("Content-Type") && !k.eq_ignore_ascii_case("Link")
122 30 : })
123 14 : .cloned()
124 14 : .collect(),
125 14 : notifier_info: Vec::new(),
126 : });
127 4 : }
128 : // Pre-registry MQTT: the payload is the 7.2 message, so the notification
129 : // and its metadata come back out of the wrapper.
130 4 : let meta = &letter["payload"]["metadata"];
131 4 : if let Some(meta) = meta.as_object() {
132 2 : let mut notifier_info = Vec::new();
133 2 : if let Some(q) = letter["mqtt"]["qos"].as_u64() {
134 2 : notifier_info.push(("MQTT-QoS".to_owned(), q.to_string()));
135 2 : }
136 2 : if let Some(v5) = letter["mqtt"]["v5"].as_bool() {
137 2 : let v = if v5 { "mqtt5.0" } else { "mqtt3.1.1" };
138 2 : notifier_info.push(("MQTT-Version".to_owned(), v.to_owned()));
139 0 : }
140 : return Ok(Self {
141 2 : body: letter["payload"]["body"].clone(),
142 2 : accept: meta
143 2 : .get("Content-Type")
144 2 : .and_then(Value::as_str)
145 2 : .unwrap_or("application/json")
146 2 : .to_owned(),
147 2 : link: meta
148 2 : .get("Link")
149 2 : .and_then(Value::as_str)
150 2 : .unwrap_or_default()
151 2 : .to_owned(),
152 2 : receiver_info: meta
153 2 : .iter()
154 6 : .filter(|(k, _)| k.as_str() != "Content-Type" && k.as_str() != "Link")
155 2 : .filter_map(|(k, v)| Some((k.clone(), v.as_str()?.to_owned())))
156 2 : .collect(),
157 2 : notifier_info,
158 : });
159 2 : }
160 2 : Err("dead letter carries no deliverable notification".to_owned())
161 22 : }
162 :
163 : /// `notifier_info` in the borrowed pair form the sinks parse.
164 0 : pub fn notifier_pairs(&self) -> Vec<(&str, &str)> {
165 0 : self.notifier_info
166 0 : .iter()
167 0 : .map(|(k, v)| (k.as_str(), v.as_str()))
168 0 : .collect()
169 0 : }
170 : }
171 :
172 : /// Why one delivery attempt did not land. `timed_out` is the only class the
173 : /// caller's circuit breaker counts: an endpoint that answers — with any
174 : /// status — is alive.
175 : #[derive(Clone, Debug, PartialEq)]
176 : pub struct DeliveryError {
177 : /// The attempt ran out of time rather than being answered.
178 : pub timed_out: bool,
179 : /// Failure text for the log, the subscription status and the dead
180 : /// letter. Never carries endpoint credentials.
181 : pub message: String,
182 : }
183 :
184 : impl DeliveryError {
185 : /// The endpoint answered, or refused, within the deadline.
186 50 : pub fn failed(message: impl Into<String>) -> Self {
187 50 : Self {
188 50 : timed_out: false,
189 50 : message: message.into(),
190 50 : }
191 50 : }
192 :
193 : /// The deadline passed with no answer.
194 0 : pub fn timeout(message: impl Into<String>) -> Self {
195 0 : Self {
196 0 : timed_out: true,
197 0 : message: message.into(),
198 0 : }
199 0 : }
200 : }
201 :
202 : /// The future one `deliver` returns. `Send` on every target: the un-Send
203 : /// piece of a browser fetch is fenced inside `antares_jsonld::http_interaction`.
204 : pub type DeliveryFuture<'a> = Pin<Box<dyn Future<Output = Result<(), DeliveryError>> + Send + 'a>>;
205 :
206 : /// How one notification is delivered: how many attempts, how they are
207 : /// spaced, and how long after the first attempt the last one may still
208 : /// start. The default is a single attempt — exactly 5.8.6, which sends the
209 : /// notification once and books the outcome. Retries are an operator choice:
210 : /// they never move `timesSent` again (the notification is sent ONCE, the
211 : /// attempts are transport), a retry that succeeds books `lastSuccess` and
212 : /// `status` ok, an exhausted policy leaves a dead letter.
213 : #[derive(Clone, Copy, Debug, PartialEq)]
214 : pub struct DeliveryPolicy {
215 : /// Total attempts, first one included. 1 = never retry.
216 : pub attempts: u32,
217 : /// Delay before the first retry; doubles per retry up to `MAX_BACKOFF`.
218 : pub backoff: std::time::Duration,
219 : /// Fraction of the delay randomised in both directions (0.2 = ±20 %),
220 : /// so many subscriptions to one dead endpoint do not retry in lockstep.
221 : pub jitter: f32,
222 : /// A retry that would start later than this after the first attempt is
223 : /// not made.
224 : pub max_age: std::time::Duration,
225 : }
226 :
227 : impl Default for DeliveryPolicy {
228 4901 : fn default() -> Self {
229 4901 : Self {
230 4901 : attempts: 1,
231 4901 : backoff: std::time::Duration::from_secs(1),
232 4901 : jitter: 0.2,
233 4901 : max_age: std::time::Duration::from_secs(300),
234 4901 : }
235 4901 : }
236 : }
237 :
238 : impl DeliveryPolicy {
239 : /// Ceiling on one delay, whatever the doubling says.
240 : pub const MAX_BACKOFF: std::time::Duration = std::time::Duration::from_secs(60);
241 :
242 2 : pub fn with_max_age(self, max_age: std::time::Duration) -> Self {
243 2 : Self { max_age, ..self }
244 2 : }
245 :
246 : /// ANTARES_NOTIFY_ATTEMPTS / ANTARES_NOTIFY_BACKOFF_MS /
247 : /// ANTARES_NOTIFY_MAX_AGE_SECS, each optional; a value that is present
248 : /// but not a positive integer is a startup error, never a silent default.
249 61 : pub fn from_env() -> Result<Self, String> {
250 183 : let get = |k: &str| std::env::var(k).ok();
251 61 : Self::parse(
252 61 : get("ANTARES_NOTIFY_ATTEMPTS").as_deref(),
253 61 : get("ANTARES_NOTIFY_BACKOFF_MS").as_deref(),
254 61 : get("ANTARES_NOTIFY_MAX_AGE_SECS").as_deref(),
255 : )
256 61 : }
257 :
258 79 : pub fn parse(
259 79 : attempts: Option<&str>,
260 79 : backoff_ms: Option<&str>,
261 79 : max_age_secs: Option<&str>,
262 79 : ) -> Result<Self, String> {
263 219 : fn positive(name: &str, raw: Option<&str>) -> Result<Option<u64>, String> {
264 219 : match raw {
265 197 : None => Ok(None),
266 22 : Some(v) => v
267 22 : .trim()
268 22 : .parse::<u64>()
269 22 : .ok()
270 22 : .filter(|n| *n > 0)
271 22 : .map(Some)
272 22 : .ok_or_else(|| format!("{name} must be a positive integer, got {v:?}")),
273 : }
274 219 : }
275 79 : let d = Self::default();
276 : Ok(Self {
277 79 : attempts: positive("ANTARES_NOTIFY_ATTEMPTS", attempts)?
278 73 : .map_or(d.attempts, |n| u32::try_from(n).unwrap_or(u32::MAX)),
279 73 : backoff: positive("ANTARES_NOTIFY_BACKOFF_MS", backoff_ms)?
280 67 : .map_or(d.backoff, std::time::Duration::from_millis),
281 67 : jitter: d.jitter,
282 67 : max_age: positive("ANTARES_NOTIFY_MAX_AGE_SECS", max_age_secs)?
283 65 : .map_or(d.max_age, std::time::Duration::from_secs),
284 : })
285 79 : }
286 :
287 : /// The delay before the next attempt after `made` attempts, `elapsed`
288 : /// after the first one — `None` when the policy is exhausted or the
289 : /// retry would start past `max_age`.
290 152 : pub fn next_delay(
291 152 : &self,
292 152 : made: u32,
293 152 : elapsed: std::time::Duration,
294 152 : ) -> Option<std::time::Duration> {
295 152 : if made == 0 || made >= self.attempts {
296 12 : return None;
297 140 : }
298 140 : let doubled = self
299 140 : .backoff
300 140 : .checked_mul(1u32.checked_shl(made - 1).unwrap_or(u32::MAX))
301 140 : .unwrap_or(Self::MAX_BACKOFF)
302 140 : .min(Self::MAX_BACKOFF);
303 : // ±jitter, seeded from the clock's sub-second noise: cheap and
304 : // uncorrelated enough to spread retries; no RNG dependency.
305 140 : let noise = (std::time::SystemTime::now()
306 140 : .duration_since(std::time::UNIX_EPOCH)
307 140 : .map(|d| d.subsec_nanos())
308 140 : .unwrap_or(0)
309 : % 1000) as f32
310 : / 1000.0;
311 140 : let factor = 1.0 + self.jitter.clamp(0.0, 1.0) * (2.0 * noise - 1.0);
312 140 : let delay = doubled.mul_f32(factor.max(0.0));
313 140 : (elapsed + delay <= self.max_age).then_some(delay)
314 152 : }
315 : }
316 :
317 : /// Strip the authority's userinfo from an endpoint URI. 7.2 allows
318 : /// credentials there (`mqtt[s]://[<username>][:<password>]@<host>…`), and a
319 : /// rejected or failed endpoint travels back to the client as the `detail`
320 : /// member of the ProblemDetails body (5.5.3) and into the delivery logs —
321 : /// neither may carry the subscription's password. Everything after the
322 : /// authority's last `@` is kept; an `@` in the path or topic is data.
323 : ///
324 : /// A URI written without the `//` (`http:user:pass@host/n`) has no
325 : /// authority in RFC 3986 terms, yet a WHATWG URL parser (the one reqwest
326 : /// uses) reads one there and sends that userinfo. Such a URI is stripped the
327 : /// same way, from the first `:` on. The authority starts right after the
328 : /// scheme's `:` (and its `//`, when present), never at a `//` later in the
329 : /// path or query, which would leave the userinfo in place.
330 2113 : pub fn redact_userinfo(uri: &str) -> String {
331 2113 : let Some(colon) = uri.find(':') else {
332 4 : return uri.to_owned();
333 : };
334 2109 : let start = colon
335 2109 : + 1
336 2109 : + if uri[colon + 1..].starts_with("//") {
337 2087 : 2
338 : } else {
339 22 : 0
340 : };
341 2109 : let rest = &uri[start..];
342 2109 : let authority_end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
343 2109 : if let Some(at) = rest[..authority_end].rfind('@') {
344 144 : return format!("{}{}", &uri[..start], &rest[at + 1..]);
345 1965 : }
346 1965 : uri.to_owned()
347 2113 : }
348 :
349 : /// A delivery binding for one URI scheme family.
350 : pub trait NotificationSink: Send + Sync {
351 : /// Schemes this sink serves, e.g. `["http", "https"]`.
352 : fn schemes(&self) -> &'static [&'static str];
353 :
354 : /// 5.8.1.4: the endpoint's own syntax and parameters, checked at
355 : /// subscription creation rather than at first delivery. `notifier_info`
356 : /// is `endpoint.notifierInfo` (Table 5.2.15-1) as key/value pairs. An
357 : /// endpoint that does not meet the sink's requirements is
358 : /// BadRequestData; the message names the URI with any userinfo
359 : /// credentials stripped, since it travels back in `detail` (5.5.3).
360 : fn parse_endpoint(&self, uri: &str, notifier_info: &[(&str, &str)]) -> Result<(), NgsiError>;
361 :
362 : /// Does an endpoint of this binding name a network destination? The
363 : /// caller runs the egress guard — host and port policy, private-range
364 : /// and metadata-address deny, per-destination circuit breaker — against
365 : /// every endpoint that does, before `deliver` and never inside it, so
366 : /// one guard covers every binding whatever scheme it serves. The
367 : /// default is the safe answer: a sink is policed unless it declares
368 : /// that it opens no socket, and a release binary registers no sink that
369 : /// declares otherwise.
370 9529 : fn network(&self) -> bool {
371 9529 : true
372 9529 : }
373 :
374 : /// One attempt on the wire. `timeout` is `endpoint.timeout`
375 : /// (Table 5.2.15-1) already clamped by the caller.
376 : fn deliver<'a>(
377 : &'a self,
378 : uri: &'a str,
379 : out: &'a Outbound,
380 : timeout: Duration,
381 : ) -> DeliveryFuture<'a>;
382 :
383 : /// A notification of `subscription` the broker dropped before
384 : /// `deliver` (a full delivery lane). A binding whose consumer can be
385 : /// told of the loss in band tells it; the default has no such channel.
386 3829 : fn lost(&self, _uri: &str, _tenant: &str, _subscription: &str) {}
387 : }
388 :
389 : /// Scheme → sink registry; populated by the composition root. Choosing a
390 : /// binding goes through here and nowhere else, so an endpoint scheme with no
391 : /// sink can never fall through to the HTTP binding.
392 : #[derive(Default)]
393 : pub struct SinkRegistry {
394 : sinks: Vec<Box<dyn NotificationSink>>,
395 : }
396 :
397 : impl SinkRegistry {
398 : /// Add a binding. A scheme already served keeps its first sink.
399 9656 : pub fn register(&mut self, sink: Box<dyn NotificationSink>) {
400 9656 : self.sinks.push(sink);
401 9656 : }
402 :
403 : /// The scheme of an endpoint URI, lowercased per IETF RFC 3986 §3.1.
404 29945 : pub fn scheme_of(uri: &str) -> String {
405 29945 : uri.split(':').next().unwrap_or("").to_ascii_lowercase()
406 29945 : }
407 :
408 : /// The sink serving `scheme`, if any.
409 29941 : pub fn sink_for(&self, scheme: &str) -> Option<&dyn NotificationSink> {
410 : // Linear scan is fine at <5 sinks; switch to a map when sinks multiply.
411 29941 : self.sinks
412 29941 : .iter()
413 31085 : .find(|s| s.schemes().contains(&scheme))
414 29941 : .map(AsRef::as_ref)
415 29941 : }
416 :
417 : /// The sink serving an endpoint URI, by its scheme.
418 28665 : pub fn sink_for_uri(&self, uri: &str) -> Option<&dyn NotificationSink> {
419 28665 : self.sink_for(&Self::scheme_of(uri))
420 28665 : }
421 :
422 : /// 5.8.1.4 reject-at-creation: the sink for this endpoint, having
423 : /// accepted the endpoint's own syntax. An endpoint whose scheme this
424 : /// deployment cannot deliver to is input data that does not meet the
425 : /// requirements of the operation — BadRequestData (Table 5.5.2-1, 400
426 : /// per Table 6.3.2-1). Not OperationNotSupported: Create Subscription
427 : /// is supported, this endpoint value is not.
428 1274 : pub fn require(&self, uri: &str, notifier_info: &[(&str, &str)]) -> Result<(), NgsiError> {
429 1274 : let scheme = Self::scheme_of(uri);
430 1274 : let sink = self.sink_for(&scheme).ok_or_else(|| {
431 8 : NgsiError::BadRequestData(format!(
432 8 : "no notification binding registered for endpoint scheme {scheme:?} (6.3.8)"
433 8 : ))
434 8 : })?;
435 1266 : sink.parse_endpoint(uri, notifier_info)
436 1274 : }
437 :
438 : /// Schemes this deployment can deliver to, for `/q/health` and the
439 : /// startup banner.
440 88 : pub fn schemes(&self) -> Vec<&'static str> {
441 88 : let mut v: Vec<&'static str> = self
442 88 : .sinks
443 88 : .iter()
444 174 : .flat_map(|s| s.schemes())
445 88 : .copied()
446 88 : .collect();
447 88 : v.sort_unstable();
448 88 : v.dedup();
449 88 : v
450 88 : }
451 : }
452 :
453 : #[cfg(test)]
454 : mod tests {
455 : use super::*;
456 : use serde_json::json;
457 :
458 : struct FakeHttp;
459 : impl NotificationSink for FakeHttp {
460 18 : fn schemes(&self) -> &'static [&'static str] {
461 18 : &["http", "https"]
462 18 : }
463 8 : fn parse_endpoint(&self, uri: &str, _ni: &[(&str, &str)]) -> Result<(), NgsiError> {
464 8 : uri.contains("://")
465 8 : .then_some(())
466 8 : .ok_or_else(|| NgsiError::BadRequestData(format!("no authority in {uri:?}")))
467 8 : }
468 0 : fn deliver<'a>(
469 0 : &'a self,
470 0 : _uri: &'a str,
471 0 : _o: &'a Outbound,
472 0 : _t: Duration,
473 0 : ) -> DeliveryFuture<'a> {
474 0 : Box::pin(async { Ok(()) })
475 0 : }
476 : }
477 :
478 8 : fn registry() -> SinkRegistry {
479 8 : let mut reg = SinkRegistry::default();
480 8 : reg.register(Box::new(FakeHttp));
481 8 : reg
482 8 : }
483 :
484 : /// 5.8.1.4 + Table 5.5.2-1: an endpoint scheme no sink serves is input
485 : /// data that does not meet the operation's requirements — BadRequestData
486 : /// (400), not OperationNotSupported. Create Subscription is supported.
487 : #[test]
488 2 : fn unknown_scheme_is_bad_request_data() {
489 2 : let reg = registry();
490 2 : assert!(reg.require("http://h/n", &[]).is_ok());
491 2 : assert!(reg.require("https://h/n", &[]).is_ok());
492 2 : let err = reg
493 2 : .require("ws://h/n", &[])
494 2 : .expect_err("ws has no sink in v1");
495 2 : assert_eq!(err.status(), 400);
496 2 : assert!(matches!(err, NgsiError::BadRequestData(_)), "{err:?}");
497 2 : assert!(format!("{err}").contains("ws"), "{err}");
498 2 : }
499 :
500 : /// The scheme comparison is case-insensitive (IETF RFC 3986 §3.1),
501 : /// so `HTTP://…` is not silently unroutable.
502 : #[test]
503 2 : fn scheme_matching_ignores_case() {
504 2 : assert!(registry().require("HTTP://h/n", &[]).is_ok());
505 2 : assert_eq!(SinkRegistry::scheme_of("MQTTS://h/t"), "mqtts");
506 2 : }
507 :
508 : /// The sink's own endpoint check runs at creation, and its error is the
509 : /// one the client sees.
510 : #[test]
511 2 : fn the_sinks_own_rejection_is_returned() {
512 2 : let err = registry()
513 2 : .require("http:/malformed", &[])
514 2 : .expect_err("no authority");
515 2 : assert_eq!(err.status(), 400);
516 2 : assert!(format!("{err}").contains("no authority"), "{err}");
517 2 : }
518 :
519 : /// Every binding this workspace ships opens a socket, so every endpoint
520 : /// it delivers to passes the caller's egress policy. A sink that
521 : /// declares otherwise is in-process only and belongs to a test.
522 : #[test]
523 2 : fn every_shipped_sink_is_policed() {
524 2 : let mut reg = SinkRegistry::default();
525 2 : reg.register(Box::new(crate::HttpSink::new(
526 2 : antares_jsonld::client_builder(antares_jsonld::EgressPolicy {
527 2 : allow_private: true,
528 2 : private_hosts: &[],
529 2 : })
530 2 : .build()
531 2 : .expect("client"),
532 2 : )));
533 : #[cfg(feature = "mqtt")]
534 2 : reg.register(Box::new(crate::mqtt::MqttSink::default()));
535 2 : assert!(!reg.sinks.is_empty());
536 4 : for s in ®.sinks {
537 4 : assert!(
538 4 : s.network(),
539 : "shipped sink {:?} skips the egress policy",
540 0 : s.schemes()
541 : );
542 : }
543 2 : }
544 :
545 : /// A dead letter reads back into the same notification whichever broker
546 : /// wrote it: the current shape, and the two shapes written before the
547 : /// bindings moved behind the registry.
548 : #[test]
549 2 : fn dead_letters_of_every_shape_read_back() {
550 2 : let body = json!({"type": "Notification", "subscriptionId": "urn:s:1"});
551 2 : let link = "<https://ctx>; rel=\"http://www.w3.org/ns/json-ld#context\"";
552 :
553 2 : let current = json!({"uri": "http://h/n", "payload": body,
554 2 : "accept": "application/ld+json", "link": link,
555 2 : "receiverInfo": [["Authorization", "Bearer t"]],
556 2 : "notifierInfo": []});
557 2 : let o = Outbound::from_dead_letter(¤t).expect("current shape");
558 2 : assert_eq!(o.body, body);
559 2 : assert_eq!(o.accept, "application/ld+json");
560 2 : assert_eq!(
561 : o.receiver_info,
562 2 : [("Authorization".into(), "Bearer t".into())]
563 : );
564 :
565 2 : let legacy_http = json!({"uri": "http://h/n", "binding": "http", "payload": body,
566 2 : "headers": [["Content-Type", "application/json"], ["Link", link],
567 : ["Authorization", "Bearer t"]]});
568 2 : let o = Outbound::from_dead_letter(&legacy_http).expect("legacy http shape");
569 2 : assert_eq!(o.body, body);
570 2 : assert_eq!(o.accept, "application/json");
571 2 : assert_eq!(o.link, link);
572 2 : assert_eq!(
573 : o.receiver_info,
574 2 : [("Authorization".into(), "Bearer t".into())],
575 : "Content-Type and Link are the accept and link, not receiverInfo"
576 : );
577 :
578 2 : let legacy_mqtt = json!({"uri": "mqtt://h/t", "binding": "mqtt",
579 2 : "mqtt": {"qos": 2, "v5": false},
580 2 : "payload": {"metadata": {"Content-Type": "application/json", "Link": link,
581 2 : "NGSILD-Tenant": "acme"},
582 2 : "body": body}});
583 2 : let o = Outbound::from_dead_letter(&legacy_mqtt).expect("legacy mqtt shape");
584 2 : assert_eq!(
585 : o.body, body,
586 : "the notification comes back out of the wrapper"
587 : );
588 2 : assert_eq!(o.accept, "application/json");
589 2 : assert_eq!(o.receiver_info, [("NGSILD-Tenant".into(), "acme".into())]);
590 2 : assert_eq!(
591 : o.notifier_info,
592 2 : [
593 2 : ("MQTT-QoS".to_owned(), "2".to_owned()),
594 2 : ("MQTT-Version".to_owned(), "mqtt3.1.1".to_owned())
595 2 : ]
596 : );
597 :
598 2 : assert!(Outbound::from_dead_letter(&json!({"uri": "http://h/n"})).is_err());
599 2 : }
600 :
601 : /// Endpoint URIs may carry credentials (mqtt[s]://user:pass@host, 7.1);
602 : /// log lines must never leak them.
603 : #[test]
604 2 : fn log_redaction_strips_uri_userinfo() {
605 2 : let red = redact_userinfo("mqtts://alice:s3cret@broker:8883/topic");
606 2 : assert_eq!(red, "mqtts://broker:8883/topic");
607 2 : assert!(!red.contains("s3cret"));
608 2 : assert!(!red.contains("alice"));
609 2 : assert_eq!(
610 2 : redact_userinfo("http://host:9090/notify"),
611 : "http://host:9090/notify"
612 : );
613 : // an '@' beyond the authority is path data, not userinfo
614 2 : assert_eq!(redact_userinfo("http://h/p@x"), "http://h/p@x");
615 2 : }
616 :
617 : /// `http:user:pass@host` has no `//`, but a WHATWG parser still reads
618 : /// userinfo there: the redaction strips it, and the http binding refuses
619 : /// the URI without echoing the credential.
620 : #[test]
621 2 : fn userinfo_without_the_double_slash_is_redacted_and_refused() {
622 2 : let uri = "http:hook:pa55word@localhost:1/n";
623 2 : let red = redact_userinfo(uri);
624 2 : assert!(!red.contains("pa55word") && !red.contains("hook"), "{red}");
625 2 : assert_eq!(red, "http:localhost:1/n");
626 2 : let err = http::check_http_endpoint(uri).expect_err("no authority");
627 2 : assert!(!format!("{err:?}").contains("pa55word"), "{err:?}");
628 : // SECV-02: a `//` later in the path does not move the authority
629 6 : for uri in [
630 2 : "http:hook:pa55word@h/n//x",
631 2 : "http:hook:pa55word@h/n?u=//x",
632 2 : "mqtt://u:pa55word@h/a//b",
633 2 : ] {
634 6 : let red = redact_userinfo(uri);
635 6 : assert!(!red.contains("pa55word"), "{uri} -> {red}");
636 6 : if let Err(err) = http::check_http_endpoint(uri) {
637 6 : assert!(!format!("{err:?}").contains("pa55word"), "{err:?}");
638 0 : }
639 : }
640 2 : }
641 :
642 : /// An unregistered scheme resolves to no sink at delivery time either —
643 : /// there is no fall-through to the first registered binding.
644 : #[test]
645 2 : fn delivery_lookup_never_falls_through() {
646 2 : let reg = registry();
647 2 : assert!(reg.sink_for_uri("http://h/n").is_some());
648 2 : assert!(reg.sink_for_uri("memory://box").is_none());
649 2 : assert!(reg.sink_for("").is_none());
650 2 : assert_eq!(reg.schemes(), vec!["http", "https"]);
651 2 : }
652 : }
653 :
654 : #[cfg(test)]
655 : mod policy_tests {
656 : use super::DeliveryPolicy;
657 : use std::time::Duration;
658 :
659 10 : fn policy(attempts: u32, backoff_ms: u64, max_age_secs: u64) -> DeliveryPolicy {
660 10 : DeliveryPolicy {
661 10 : attempts,
662 10 : backoff: Duration::from_millis(backoff_ms),
663 10 : jitter: 0.0,
664 10 : max_age: Duration::from_secs(max_age_secs),
665 10 : }
666 10 : }
667 :
668 : /// Drive `op` the way the delivery loop does: one attempt, then a
669 : /// retry after every delay the policy grants. Returns the outcome and
670 : /// how many calls were made.
671 8 : fn run<E>(p: DeliveryPolicy, mut op: impl FnMut(u32) -> Result<(), E>) -> (Result<(), E>, u32) {
672 8 : let mut made = 1;
673 8 : let mut elapsed = Duration::ZERO;
674 8 : let mut last = op(made);
675 18 : while last.is_err() {
676 16 : let Some(d) = p.next_delay(made, elapsed) else {
677 6 : break;
678 : };
679 10 : elapsed += d;
680 10 : made += 1;
681 10 : last = op(made);
682 : }
683 8 : (last, made)
684 8 : }
685 :
686 : #[test]
687 2 : fn default_is_a_single_attempt() {
688 2 : let p = DeliveryPolicy::default();
689 2 : assert_eq!(p.attempts, 1);
690 2 : assert_eq!(p.next_delay(1, Duration::ZERO), None);
691 2 : let (res, made) = run(p, |_| Err::<(), _>("down"));
692 2 : assert_eq!(made, 1);
693 2 : assert_eq!(res, Err("down"));
694 2 : }
695 :
696 : #[test]
697 2 : fn fails_twice_then_succeeds_is_three_calls_and_one_ok() {
698 2 : let (res, made) = run(
699 2 : policy(3, 100, 60),
700 6 : |n| if n < 3 { Err("down") } else { Ok(()) },
701 : );
702 2 : assert_eq!(made, 3);
703 2 : assert_eq!(res, Ok(()));
704 2 : }
705 :
706 : #[test]
707 2 : fn always_failing_stops_after_attempts_with_the_last_error() {
708 2 : let (res, made) = run(policy(3, 100, 60), Err::<(), _>);
709 2 : assert_eq!(made, 3);
710 2 : assert_eq!(res, Err(3), "the LAST error is returned");
711 2 : }
712 :
713 : #[test]
714 2 : fn max_age_cuts_the_schedule_short() {
715 : // 100 ms, 200 ms, 400 ms … but only 250 ms of age allowed: the
716 : // second retry would land at 300 ms, so it is never made.
717 2 : let (res, made) = run(
718 2 : policy(10, 100, 0).with_max_age(Duration::from_millis(250)),
719 : |_| Err::<(), _>("down"),
720 : );
721 2 : assert_eq!(made, 2);
722 2 : assert!(res.is_err());
723 2 : }
724 :
725 : #[test]
726 2 : fn backoff_doubles_per_retry_and_jitter_stays_within_bounds() {
727 2 : let p = policy(5, 100, 60);
728 2 : assert_eq!(
729 2 : p.next_delay(1, Duration::ZERO),
730 2 : Some(Duration::from_millis(100))
731 : );
732 2 : assert_eq!(
733 2 : p.next_delay(2, Duration::ZERO),
734 2 : Some(Duration::from_millis(200))
735 : );
736 2 : assert_eq!(
737 2 : p.next_delay(3, Duration::ZERO),
738 2 : Some(Duration::from_millis(400))
739 : );
740 2 : let j = DeliveryPolicy { jitter: 0.5, ..p };
741 2 : for _ in 0..50 {
742 100 : let d = j.next_delay(1, Duration::ZERO).expect("granted");
743 100 : assert!(
744 100 : d >= Duration::from_millis(50) && d <= Duration::from_millis(150),
745 : "{d:?}"
746 : );
747 : }
748 2 : let d = j.next_delay(30, Duration::ZERO);
749 2 : assert_eq!(d, None, "attempt 30 of 5 is over");
750 2 : let big = policy(40, 100, 3600);
751 2 : let d = big.next_delay(35, Duration::ZERO).expect("granted");
752 2 : assert!(
753 2 : d <= DeliveryPolicy::MAX_BACKOFF,
754 : "2^34 backoff must not overflow: {d:?}"
755 : );
756 2 : }
757 :
758 : #[test]
759 2 : fn env_parsing_rejects_garbage_and_zero_attempts() {
760 2 : let parse =
761 18 : |a: Option<&str>, b: Option<&str>, m: Option<&str>| DeliveryPolicy::parse(a, b, m);
762 2 : assert_eq!(parse(None, None, None), Ok(DeliveryPolicy::default()));
763 2 : let p = parse(Some("3"), Some("250"), Some("30")).expect("valid");
764 2 : assert_eq!(p.attempts, 3);
765 2 : assert_eq!(p.backoff, Duration::from_millis(250));
766 2 : assert_eq!(p.max_age, Duration::from_secs(30));
767 14 : for bad in [
768 2 : parse(Some("0"), None, None),
769 2 : parse(Some("-1"), None, None),
770 2 : parse(Some("three"), None, None),
771 2 : parse(None, Some("0"), None),
772 2 : parse(None, Some("1e3"), None),
773 2 : parse(None, None, Some("never")),
774 2 : parse(Some("2"), Some(""), None),
775 2 : ] {
776 14 : assert!(bad.is_err(), "{bad:?}");
777 : }
778 2 : }
779 : }
|
