Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! MQTT notification binding — CIM 009 clause 7 (feature `mqtt`).
3 : //!
4 : //! 7.2: a subscription whose `notification.endpoint.uri` uses the mqtt(s)
5 : //! scheme gets its notifications as MQTT publishes. The message is a JSON
6 : //! object `{"metadata": {...}, "body": <Notification per 5.3.1>}`; protocol
7 : //! parameters ride in `notifier_info` (Table 7.2-1), receiver metadata in
8 : //! `receiver_info` (Table 7.2-2).
9 :
10 : use antares_jsonld::loader::EgressPolicy;
11 : use antares_model::NgsiError;
12 : use serde_json::{Map, Value};
13 : use std::collections::HashMap;
14 : use std::sync::Mutex;
15 : use std::time::{Duration, Instant};
16 :
17 24 : fn bad(m: String) -> NgsiError {
18 24 : NgsiError::BadRequestData(m)
19 24 : }
20 :
21 : use crate::redact_userinfo as redacted;
22 :
23 : /// Parsed `mqtt[s]://[user][:pass]@host[:port]/topic[/subtopic]*` (7.2).
24 : #[derive(Debug, Clone, PartialEq)]
25 : pub struct MqttEndpoint {
26 : pub secure: bool,
27 : pub username: Option<String>,
28 : pub password: Option<String>,
29 : pub host: String,
30 : pub port: u16,
31 : pub topic: String,
32 : }
33 :
34 : impl MqttEndpoint {
35 : /// 7.2 endpoint URI syntax. A URI that does not meet it fails the
36 : /// 5.2.15 restrictions, so the caller raises BadRequestData — 400 per
37 : /// Table 6.3.2-1. The message names only the redacted URI: the
38 : /// credentials 7.2 permits in the userinfo never reach the response body.
39 50 : pub fn parse(uri: &str) -> Result<Self, NgsiError> {
40 50 : let safe = redacted(uri);
41 : // IETF RFC 3986 3.1: scheme names are case-insensitive, and
42 : // `SinkRegistry::scheme_of` already lowercases to pick this sink.
43 50 : let (secure, rest) = match uri.split_once("://") {
44 50 : Some((s, r)) if s.eq_ignore_ascii_case("mqtts") => (true, r),
45 42 : Some((s, r)) if s.eq_ignore_ascii_case("mqtt") => (false, r),
46 4 : _ => return Err(bad(format!("not an mqtt(s) endpoint URI: {safe:?}"))),
47 : };
48 46 : let (authority, topic) = rest
49 46 : .split_once('/')
50 46 : .ok_or_else(|| bad(format!("mqtt endpoint {safe:?} has no topic")))?;
51 42 : if topic.is_empty() {
52 4 : return Err(bad(format!("mqtt endpoint {safe:?} has no topic")));
53 38 : }
54 38 : let (userinfo, hostport) = match authority.rsplit_once('@') {
55 18 : Some((u, h)) => (Some(u), h),
56 20 : None => (None, authority),
57 : };
58 38 : let (username, password) = match userinfo {
59 20 : None => (None, None),
60 18 : Some(u) => match u.split_once(':') {
61 14 : Some((user, pass)) => (Some(user.to_owned()), Some(pass.to_owned())),
62 4 : None => (Some(u.to_owned()), None),
63 : },
64 : };
65 : // Deliberately no IPv6-literal hosts — the binding's URI convention
66 : // (i.19) and the ETSI suite use hostnames; add bracket parsing when a
67 : // deployment needs it.
68 38 : let (host, port) = match hostport.split_once(':') {
69 8 : Some((h, p)) => (
70 8 : h.to_owned(),
71 8 : p.parse::<u16>()
72 8 : .map_err(|_| bad(format!("invalid mqtt port in {safe:?}")))?,
73 : ),
74 30 : None => (hostport.to_owned(), if secure { 8883 } else { 1883 }),
75 : };
76 34 : if host.is_empty() {
77 4 : return Err(bad(format!("mqtt endpoint {safe:?} has no host")));
78 30 : }
79 30 : Ok(Self {
80 30 : secure,
81 30 : username,
82 30 : password,
83 30 : host,
84 30 : port,
85 30 : topic: topic.to_owned(),
86 30 : })
87 50 : }
88 : }
89 :
90 : /// Table 7.2-1 protocol parameters from `notification.endpoint.notifierInfo`.
91 : #[derive(Debug, Clone, Copy, PartialEq)]
92 : pub struct MqttParams {
93 : pub qos: u8,
94 : pub v5: bool,
95 : }
96 :
97 : impl Default for MqttParams {
98 12 : fn default() -> Self {
99 12 : Self { qos: 0, v5: true } // defaults per Table 7.2-1: QoS 0, mqtt5.0
100 12 : }
101 : }
102 :
103 : impl MqttParams {
104 8 : pub fn from_notifier_info<'a>(
105 8 : pairs: impl IntoIterator<Item = (&'a str, &'a str)>,
106 8 : ) -> Result<Self, NgsiError> {
107 8 : let mut p = Self::default();
108 8 : for (k, v) in pairs {
109 8 : match k {
110 8 : "MQTT-QoS" => {
111 4 : p.qos = match v {
112 4 : "0" => 0,
113 4 : "1" => 1,
114 4 : "2" => 2,
115 2 : _ => return Err(bad(format!("MQTT-QoS must be 0, 1 or 2 (got {v:?})"))),
116 : }
117 : }
118 4 : "MQTT-Version" => {
119 4 : p.v5 = match v {
120 4 : "mqtt5.0" => true,
121 4 : "mqtt3.1.1" => false,
122 : _ => {
123 2 : return Err(bad(format!(
124 2 : "MQTT-Version must be mqtt3.1.1 or mqtt5.0 (got {v:?})"
125 2 : )))
126 : }
127 : }
128 : }
129 0 : _ => {} // unknown notifierInfo keys are not ours to police
130 : }
131 : }
132 4 : Ok(p)
133 8 : }
134 : }
135 :
136 : /// The 7.2 message: `{"metadata": {...}, "body": notification}`.
137 : /// `link` is the HTTP-Link-header-formatted @context reference; per Table
138 : /// 7.2-2 it is included only when the Content-Type is application/json
139 : /// (with ld+json the @context travels in the body).
140 6 : pub fn build_message(
141 6 : body: &Value,
142 6 : content_type: &str,
143 6 : link: Option<&str>,
144 6 : receiver_info: &[(String, String)],
145 6 : ) -> Value {
146 6 : let mut metadata = Map::new();
147 : // receiverInfo goes in first: 7.2 adds its entries "additionally", while
148 : // Table 7.2-2 names endpoint.accept as the source of Content-Type and
149 : // the served @context as the source of Link. A pair keyed like one of
150 : // those two is an entry, never that parameter's value.
151 8 : for (k, v) in receiver_info {
152 8 : metadata.insert(k.clone(), Value::String(v.clone()));
153 8 : }
154 6 : metadata.insert("Content-Type".into(), Value::String(content_type.into()));
155 6 : if content_type == "application/json" {
156 4 : if let Some(l) = link {
157 4 : metadata.insert("Link".into(), Value::String(l.to_owned()));
158 4 : }
159 2 : }
160 6 : let mut msg = Map::new();
161 6 : msg.insert("metadata".into(), Value::Object(metadata));
162 6 : msg.insert("body".into(), body.clone());
163 6 : Value::Object(msg)
164 6 : }
165 :
166 : /// The key a pooled MQTT connection is shared under. Everything that changes
167 : /// the identity the connection authenticates as (or how) must participate:
168 : /// two subscriptions whose endpoints differ only in password must never
169 : /// reuse one another's authenticated connection. Keys are map keys only:
170 : /// never log them.
171 16 : fn pool_key(ep: &MqttEndpoint, params: MqttParams) -> String {
172 : // The password participates via a one-way hash so the plaintext never
173 : // sits in a map key. DefaultHasher collisions are acceptable here — a
174 : // collision only merges two pool slots, it does not skip broker-side
175 : // authentication.
176 : use std::hash::{Hash, Hasher};
177 16 : let mut h = std::collections::hash_map::DefaultHasher::new();
178 16 : ep.password.hash(&mut h);
179 16 : format!(
180 : "{}:{:016x}:{}@{}:{}/v{}",
181 16 : ep.username.as_deref().unwrap_or(""),
182 16 : h.finish(),
183 : ep.secure,
184 : ep.host,
185 : ep.port,
186 16 : if params.v5 { 5 } else { 3 }
187 : )
188 16 : }
189 :
190 : /// The mqtts trust store, built ONCE per process. Loading the platform
191 : /// certificate store on every connect is wasted work, and rumqttc's
192 : /// `TlsConfiguration::default()` panics when the store is unreadable — a
193 : /// bad cert bundle must fail the one delivery, not the broker.
194 4 : fn shared_tls_config() -> Result<rumqttc::TlsConfiguration, NgsiError> {
195 : static TLS: std::sync::OnceLock<Option<rumqttc::TlsConfiguration>> = std::sync::OnceLock::new();
196 4 : TLS.get_or_init(|| {
197 : // `TlsConfiguration::default()` is the only rumqttc constructor
198 : // that loads the platform trust store, and it panics on failure —
199 : // contain that so a broken cert bundle degrades to failed mqtts
200 : // deliveries instead of killing the process. The failure is cached:
201 : // a store unreadable at first use will not become readable later.
202 2 : std::panic::catch_unwind(rumqttc::TlsConfiguration::default)
203 2 : .map_err(|_| tracing::error!("mqtts: loading the platform certificate store failed"))
204 2 : .ok()
205 2 : })
206 4 : .clone()
207 4 : .ok_or_else(|| NgsiError::InternalError("mqtts trust store unavailable".into()))
208 4 : }
209 :
210 : /// Is egress to private/loopback ranges allowed? The MQTT destination is
211 : /// client-supplied (`notification.endpoint.uri`, 7.2), so it is governed by
212 : /// the same deployment switch, and judged by the same address classifiers,
213 : /// as the HTTP callbacks and @context fetches: `EgressPolicy` is the one
214 : /// copy of both. Private egress is allowed by default (dev boxes, compose
215 : /// stacks and the conformance mocks all live there) and
216 : /// `ANTARES_EGRESS_ALLOW_PRIVATE=false` turns the deny on for
217 : /// internet-exposed deployments. Read once per process.
218 2 : fn allow_private_egress() -> bool {
219 : static ALLOW: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
220 2 : *ALLOW.get_or_init(|| EgressPolicy::from_env().allow_private)
221 2 : }
222 :
223 : /// Resolve the endpoint host ONCE and return the address to dial, or a
224 : /// denial. Resolving for the check and then handing the NAME to the MQTT
225 : /// client would leave a window in which the answer changes (DNS rebinding):
226 : /// the connector would dial an address the policy never saw. The address
227 : /// this returns is the address dialled, so check and connect see the same
228 : /// answer by construction. A host that cannot be resolved is a DENIAL — a
229 : /// destination the policy could not judge is never dialled — and the
230 : /// resolver runs under the sink's own deadline.
231 34 : async fn checked_addr(
232 34 : host: &str,
233 34 : port: u16,
234 34 : allow_private: bool,
235 34 : dns_timeout: Duration,
236 34 : ) -> Result<std::net::SocketAddr, NgsiError> {
237 34 : let denied = || {
238 20 : NgsiError::InternalError(format!(
239 20 : "mqtt egress to {host}:{port} denied (instance metadata or private range)"
240 20 : ))
241 20 : };
242 34 : let addrs: Vec<std::net::SocketAddr> = match host
243 34 : .trim_matches(['[', ']'])
244 34 : .parse::<std::net::IpAddr>()
245 : {
246 : // an IP literal needs no resolver, and is judged by the same rules
247 28 : Ok(ip) => vec![std::net::SocketAddr::new(ip, port)],
248 6 : Err(_) => tokio::time::timeout(dns_timeout, tokio::net::lookup_host((host, port)))
249 6 : .await
250 6 : .map_err(|_| {
251 0 : NgsiError::InternalError(format!("mqtt egress: resolving {host} timed out"))
252 0 : })?
253 6 : .map_err(|e| NgsiError::InternalError(format!("mqtt egress: resolving {host}: {e}")))?
254 4 : .collect(),
255 : };
256 32 : addrs
257 32 : .into_iter()
258 34 : .find(|a| {
259 34 : !EgressPolicy::ip_is_metadata(a.ip())
260 24 : && (allow_private || !EgressPolicy::ip_is_private(a.ip()))
261 34 : })
262 32 : .ok_or_else(denied)
263 34 : }
264 :
265 : /// What to hand rumqttc as the broker address. Plain MQTT dials the checked
266 : /// ADDRESS, which pins the resolution the policy judged (and makes the
267 : /// event loop's own re-dial after a dropped connection reuse it instead of
268 : /// resolving the name again, unchecked). mqtts keeps the host NAME: rumqttc
269 : /// verifies the server certificate against the string it is given, so an
270 : /// address there would demand an IP SAN and break certificate verification
271 : /// against every ordinary broker certificate. For mqtts the check above
272 : /// still gates the connect, and the certificate name check is what stops a
273 : /// changed answer from impersonating the endpoint.
274 6 : fn dial_host(ep: &MqttEndpoint, addr: std::net::SocketAddr) -> String {
275 6 : if ep.secure {
276 2 : ep.host.clone()
277 : } else {
278 4 : addr.ip().to_string()
279 : }
280 6 : }
281 :
282 : /// One pooled connection: the client plus its event-loop pump task.
283 : enum Client {
284 : V3(rumqttc::AsyncClient),
285 : V5(rumqttc::v5::AsyncClient),
286 : }
287 :
288 : struct Conn {
289 : client: Client,
290 : pump: tokio::task::JoinHandle<()>,
291 : last_used: Instant,
292 : }
293 :
294 : impl Drop for Conn {
295 0 : fn drop(&mut self) {
296 0 : self.pump.abort();
297 0 : }
298 : }
299 :
300 : /// MQTT delivery with a bounded per-endpoint connection pool (bounded
301 : /// WITH eviction; timeouts fixed at construction).
302 : pub struct MqttSink {
303 : pool: Mutex<HashMap<String, Conn>>,
304 : cap: usize,
305 : timeout: Duration,
306 : }
307 :
308 : impl Default for MqttSink {
309 3184 : fn default() -> Self {
310 3184 : Self::new(32, Duration::from_secs(5))
311 3184 : }
312 : }
313 :
314 : impl MqttSink {
315 3186 : pub fn new(cap: usize, timeout: Duration) -> Self {
316 3186 : Self {
317 3186 : pool: Mutex::new(HashMap::new()),
318 3186 : cap,
319 3186 : timeout,
320 3186 : }
321 3186 : }
322 :
323 : /// Deliver one notification message. `message` is the 7.2 wrapper from
324 : /// [`build_message`], serialized by the caller once per subscription.
325 2 : pub async fn publish_message(
326 2 : &self,
327 2 : ep: &MqttEndpoint,
328 2 : params: MqttParams,
329 2 : message: &[u8],
330 2 : ) -> Result<(), NgsiError> {
331 2 : let key = pool_key(ep, params);
332 : // one retry with a fresh connection: a pooled client whose broker
333 : // restarted fails the first publish; a dead broker fails both.
334 2 : for attempt in 0..2 {
335 2 : let conn = match self.checkout(&key) {
336 0 : Some(c) => c,
337 2 : None => self.connect(ep, params).await?,
338 : };
339 0 : let published = tokio::time::timeout(
340 0 : self.timeout,
341 0 : Self::publish(&conn.client, &ep.topic, params.qos, message),
342 0 : )
343 0 : .await;
344 0 : match published {
345 0 : Ok(Ok(())) if !conn.pump.is_finished() => {
346 0 : self.checkin(key, conn);
347 0 : return Ok(());
348 : }
349 0 : _ if attempt == 0 => continue, // drop conn, retry fresh
350 : Ok(Ok(())) => {
351 0 : return Err(NgsiError::InternalError(
352 0 : "mqtt connection lost during publish".into(),
353 0 : ))
354 : }
355 0 : Ok(Err(e)) => return Err(NgsiError::InternalError(format!("mqtt publish: {e}"))),
356 : Err(_) => {
357 0 : return Err(NgsiError::InternalError(format!(
358 0 : "mqtt publish to {}:{} timed out",
359 0 : ep.host, ep.port
360 0 : )))
361 : }
362 : }
363 : }
364 0 : Err(NgsiError::InternalError(
365 0 : "mqtt publish exhausted its retry without a verdict".into(),
366 0 : ))
367 2 : }
368 :
369 0 : async fn publish(client: &Client, topic: &str, qos: u8, payload: &[u8]) -> Result<(), String> {
370 0 : match client {
371 0 : Client::V3(c) => {
372 0 : let qos = rumqttc::qos(qos).map_err(|e| e.to_string())?;
373 0 : c.publish(topic, qos, false, payload.to_vec())
374 0 : .await
375 0 : .map_err(|e| e.to_string())
376 : }
377 0 : Client::V5(c) => {
378 0 : let qos =
379 0 : rumqttc::v5::mqttbytes::qos(qos).ok_or_else(|| format!("invalid QoS {qos}"))?;
380 0 : c.publish(topic, qos, false, payload.to_vec())
381 0 : .await
382 0 : .map_err(|e| e.to_string())
383 : }
384 : }
385 0 : }
386 :
387 2 : fn checkout(&self, key: &str) -> Option<Conn> {
388 2 : self.pool
389 2 : .lock()
390 2 : .unwrap_or_else(std::sync::PoisonError::into_inner)
391 2 : .remove(key)
392 2 : }
393 :
394 0 : fn checkin(&self, key: String, mut conn: Conn) {
395 0 : conn.last_used = Instant::now();
396 0 : let mut pool = self
397 0 : .pool
398 0 : .lock()
399 0 : .unwrap_or_else(std::sync::PoisonError::into_inner);
400 0 : pool.retain(|_, c| !c.pump.is_finished());
401 0 : pool.insert(key, conn);
402 : // bounded with eviction: drop the least-recently-used overflow.
403 0 : while pool.len() > self.cap {
404 0 : if let Some(oldest) = pool
405 0 : .iter()
406 0 : .min_by_key(|(_, c)| c.last_used)
407 0 : .map(|(k, _)| k.clone())
408 0 : {
409 0 : pool.remove(&oldest);
410 0 : }
411 : }
412 0 : }
413 :
414 : /// Connect and wait for ConnAck (a dead broker must fail delivery, not
415 : /// queue forever), then hand the event loop to a pump task.
416 2 : async fn connect(&self, ep: &MqttEndpoint, params: MqttParams) -> Result<Conn, NgsiError> {
417 : // A monotonic counter, not a timestamp: `Instant::now().elapsed()` is
418 : // ~0 for every caller, so two connects in one process would claim the
419 : // same client id and the broker would kick the older session.
420 : static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
421 2 : let id = format!(
422 : "antares-{}-{}",
423 2 : std::process::id(),
424 2 : SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
425 : );
426 2 : let refused = |e: String| {
427 0 : NgsiError::InternalError(format!("mqtt connect {}:{}: {e}", ep.host, ep.port))
428 0 : };
429 : // Egress policy first: resolve once, judge the answer, dial what was
430 : // judged. Nothing below opens a socket to an unchecked destination.
431 2 : let addr = checked_addr(&ep.host, ep.port, allow_private_egress(), self.timeout).await?;
432 0 : let dial = dial_host(ep, addr);
433 0 : if params.v5 {
434 0 : let mut opts = rumqttc::v5::MqttOptions::new(id, dial, addr.port());
435 0 : opts.set_keep_alive(Duration::from_secs(30));
436 0 : if let Some(u) = &ep.username {
437 0 : opts.set_credentials(u, ep.password.as_deref().unwrap_or(""));
438 0 : }
439 0 : if ep.secure {
440 0 : opts.set_transport(rumqttc::Transport::Tls(shared_tls_config()?));
441 0 : }
442 0 : let (client, mut eventloop) = rumqttc::v5::AsyncClient::new(opts, 16);
443 0 : tokio::time::timeout(self.timeout, async {
444 : loop {
445 0 : match eventloop.poll().await {
446 : Ok(rumqttc::v5::Event::Incoming(
447 : rumqttc::v5::mqttbytes::v5::Packet::ConnAck(_),
448 0 : )) => return Ok(()),
449 0 : Ok(_) => {}
450 0 : Err(e) => return Err(e.to_string()),
451 : }
452 : }
453 0 : })
454 0 : .await
455 0 : .map_err(|_| refused("connect timeout".into()))?
456 0 : .map_err(refused)?;
457 0 : let pump = tokio::spawn(async move { while eventloop.poll().await.is_ok() {} });
458 0 : Ok(Conn {
459 0 : client: Client::V5(client),
460 0 : pump,
461 0 : last_used: Instant::now(),
462 0 : })
463 : } else {
464 0 : let mut opts = rumqttc::MqttOptions::new(id, dial, addr.port());
465 0 : opts.set_keep_alive(Duration::from_secs(30));
466 0 : if let Some(u) = &ep.username {
467 0 : opts.set_credentials(u, ep.password.as_deref().unwrap_or(""));
468 0 : }
469 0 : if ep.secure {
470 0 : opts.set_transport(rumqttc::Transport::Tls(shared_tls_config()?));
471 0 : }
472 0 : let (client, mut eventloop) = rumqttc::AsyncClient::new(opts, 16);
473 0 : tokio::time::timeout(self.timeout, async {
474 : loop {
475 0 : match eventloop.poll().await {
476 0 : Ok(rumqttc::Event::Incoming(rumqttc::Packet::ConnAck(_))) => return Ok(()),
477 0 : Ok(_) => {}
478 0 : Err(e) => return Err(e.to_string()),
479 : }
480 : }
481 0 : })
482 0 : .await
483 0 : .map_err(|_| refused("connect timeout".into()))?
484 0 : .map_err(refused)?;
485 0 : let pump = tokio::spawn(async move { while eventloop.poll().await.is_ok() {} });
486 0 : Ok(Conn {
487 0 : client: Client::V3(client),
488 0 : pump,
489 0 : last_used: Instant::now(),
490 0 : })
491 : }
492 2 : }
493 : }
494 :
495 : /// Clause 7: the mqtt(s) binding. `parse_endpoint` holds Table 7.2-1 (the
496 : /// endpoint URI syntax and the notifier parameters) at subscription
497 : /// creation; `deliver` publishes the 7.2 message — `{"metadata": …,
498 : /// "body": <Notification>}` — on the endpoint's topic.
499 : impl crate::NotificationSink for MqttSink {
500 120 : fn schemes(&self) -> &'static [&'static str] {
501 120 : &["mqtt", "mqtts"]
502 120 : }
503 :
504 0 : fn parse_endpoint(&self, uri: &str, notifier_info: &[(&str, &str)]) -> Result<(), NgsiError> {
505 0 : MqttEndpoint::parse(uri)?;
506 0 : MqttParams::from_notifier_info(notifier_info.iter().copied())?;
507 0 : Ok(())
508 0 : }
509 :
510 0 : fn deliver<'a>(
511 0 : &'a self,
512 0 : uri: &'a str,
513 0 : out: &'a crate::Outbound,
514 0 : _timeout: Duration,
515 0 : ) -> crate::DeliveryFuture<'a> {
516 0 : Box::pin(async move {
517 : // Creation validated the endpoint (7.2 / Table 7.2-1); a parse
518 : // failure here means a hand-edited row. The message names only
519 : // the redacted URI: 7.2 permits credentials in the userinfo.
520 0 : let Ok(ep) = MqttEndpoint::parse(uri) else {
521 0 : return Err(crate::DeliveryError::failed(format!(
522 0 : "mqtt endpoint {:?} unusable",
523 0 : redacted(uri)
524 0 : )));
525 : };
526 0 : let Ok(params) = MqttParams::from_notifier_info(out.notifier_pairs()) else {
527 0 : return Err(crate::DeliveryError::failed(format!(
528 0 : "mqtt notifierInfo of {:?} unusable",
529 0 : redacted(uri)
530 0 : )));
531 : };
532 0 : let msg = build_message(&out.body, &out.accept, Some(&out.link), &out.receiver_info);
533 0 : let bytes = antares_model::ordered_vec(&msg);
534 0 : self.publish_message(&ep, params, &bytes)
535 0 : .await
536 : // broker/socket-level failure — keep the timeout guard
537 0 : .map_err(|e| crate::DeliveryError::timeout(e.to_string()))
538 0 : })
539 0 : }
540 : }
541 :
542 : #[cfg(test)]
543 : mod tests {
544 : use super::*;
545 : use serde_json::json;
546 :
547 : #[test]
548 2 : fn parses_endpoint_variants() {
549 2 : let e = MqttEndpoint::parse("mqtt://host/topic").expect("plain");
550 2 : assert_eq!(
551 : e,
552 2 : MqttEndpoint {
553 2 : secure: false,
554 2 : username: None,
555 2 : password: None,
556 2 : host: "host".into(),
557 2 : port: 1883,
558 2 : topic: "topic".into()
559 2 : }
560 : );
561 2 : let e = MqttEndpoint::parse("mqtt://host:8085/a/b/c").expect("port+subtopics");
562 2 : assert_eq!(e.port, 8085);
563 2 : assert_eq!(e.topic, "a/b/c");
564 2 : let e = MqttEndpoint::parse("mqtt://user@host/t").expect("user");
565 2 : assert_eq!(e.username.as_deref(), Some("user"));
566 2 : assert_eq!(e.password, None);
567 2 : let e = MqttEndpoint::parse("mqtt://u:p@host:9001/t").expect("user+pass+port");
568 2 : assert_eq!(e.username.as_deref(), Some("u"));
569 2 : assert_eq!(e.password.as_deref(), Some("p"));
570 2 : assert_eq!(e.port, 9001);
571 2 : let e = MqttEndpoint::parse("mqtts://host/t").expect("tls");
572 2 : assert!(e.secure);
573 2 : assert_eq!(e.port, 8883, "mqtts default port");
574 2 : }
575 :
576 : /// IETF RFC 3986 3.1: scheme names are case-insensitive.
577 : /// `SinkRegistry::scheme_of` lowercases before it picks this sink, so an
578 : /// uppercase-scheme endpoint is routed here and this parser decides —
579 : /// rejecting it would 400 an endpoint URI the registry says it serves.
580 : #[test]
581 2 : fn the_scheme_is_matched_case_insensitively() {
582 2 : assert!(
583 2 : MqttEndpoint::parse("MQTT://host/topic")
584 2 : .expect("upper")
585 2 : .port
586 2 : == 1883
587 : );
588 2 : let e = MqttEndpoint::parse("MQTTS://host/t").expect("upper tls");
589 2 : assert!(e.secure);
590 2 : assert_eq!(e.port, 8883);
591 2 : assert!(MqttEndpoint::parse("MqTt://host/t").is_ok());
592 2 : }
593 :
594 : #[test]
595 2 : fn rejects_bad_endpoints() {
596 10 : for uri in [
597 2 : "http://host/topic",
598 2 : "mqtt://host",
599 2 : "mqtt://host/",
600 2 : "mqtt:///topic",
601 2 : "mqtt://host:notaport/t",
602 2 : ] {
603 10 : assert!(MqttEndpoint::parse(uri).is_err(), "{uri} must be rejected");
604 : }
605 2 : }
606 :
607 : /// 7.2 endpoint URIs may carry credentials in the userinfo
608 : /// (`mqtt[s]://<username>:<password>@host`). A parse failure is answered
609 : /// to the client as BadRequestData (5.8.1.4, 400 per Table 6.3.2-1) and
610 : /// the message becomes the 5.5.3 ProblemDetails `detail`, so no password
611 : /// may appear in it.
612 : #[test]
613 2 : fn parse_errors_redact_endpoint_userinfo() {
614 10 : for uri in [
615 2 : "mqtt://user:hunter2@host", // no topic
616 2 : "mqtt://user:hunter2@host/", // empty topic
617 2 : "mqtt://user:hunter2@host:notaport/t", // bad port
618 2 : "mqtts://user:hunter2@/t", // no host
619 2 : "http://user:hunter2@host/t", // wrong scheme
620 2 : ] {
621 10 : let NgsiError::BadRequestData(msg) =
622 10 : MqttEndpoint::parse(uri).expect_err(&format!("{uri} must be rejected"))
623 : else {
624 0 : panic!("{uri} must be BadRequestData (400, Table 6.3.2-1)");
625 : };
626 10 : assert!(
627 10 : !msg.contains("hunter2"),
628 : "the password leaked into the 400 detail: {msg}"
629 : );
630 10 : assert!(
631 10 : !msg.contains("user:"),
632 : "the userinfo leaked into the 400 detail: {msg}"
633 : );
634 : // the detail must still be useful: scheme and host survive
635 10 : assert!(
636 10 : msg.contains("host") || uri.contains("@/"),
637 : "the redacted detail lost the host: {msg}"
638 : );
639 : }
640 2 : }
641 :
642 : #[test]
643 2 : fn notifier_info_defaults_and_validation() {
644 2 : let p = MqttParams::from_notifier_info([]).expect("defaults");
645 2 : assert_eq!(p, MqttParams { qos: 0, v5: true });
646 2 : let p = MqttParams::from_notifier_info([("MQTT-QoS", "2"), ("MQTT-Version", "mqtt3.1.1")])
647 2 : .expect("explicit");
648 2 : assert_eq!(p, MqttParams { qos: 2, v5: false });
649 2 : assert!(MqttParams::from_notifier_info([("MQTT-QoS", "3")]).is_err());
650 2 : assert!(MqttParams::from_notifier_info([("MQTT-Version", "mqtt4")]).is_err());
651 2 : }
652 :
653 : /// The pool key must separate sessions by credentials: same user with
654 : /// two different passwords = two different authenticated principals,
655 : /// which must never share one session. The plaintext password itself
656 : /// must not appear in the key.
657 : #[test]
658 2 : fn pool_key_separates_credentials() {
659 2 : let p = MqttParams::default();
660 2 : let a = MqttEndpoint::parse("mqtt://u:secret-one@host/t").expect("a");
661 2 : let b = MqttEndpoint::parse("mqtt://u:secret-two@host/t").expect("b");
662 2 : let c = MqttEndpoint::parse("mqtt://u:secret-one@host/t").expect("c");
663 2 : assert_ne!(
664 2 : pool_key(&a, p),
665 2 : pool_key(&b, p),
666 : "different passwords must not share an authenticated session"
667 : );
668 2 : assert_eq!(
669 2 : pool_key(&a, p),
670 2 : pool_key(&c, p),
671 : "identical endpoints must keep pooling"
672 : );
673 2 : assert!(
674 2 : !pool_key(&a, p).contains("secret-one"),
675 : "the plaintext password must never appear in the key"
676 : );
677 : // no password vs some password are different principals too
678 2 : let none = MqttEndpoint::parse("mqtt://u@host/t").expect("none");
679 2 : assert_ne!(pool_key(&a, p), pool_key(&none, p));
680 2 : }
681 :
682 : /// The mqtts trust store is loaded once and the same shared rustls
683 : /// config is handed to every connect. (The failure path — an unreadable
684 : /// platform store — cannot be simulated in a unit test; the helper's
685 : /// error contract covers it.)
686 : #[test]
687 2 : fn tls_config_is_built_once_and_shared() {
688 2 : let a = shared_tls_config().expect("first load");
689 2 : let b = shared_tls_config().expect("second load");
690 2 : let (rumqttc::TlsConfiguration::Rustls(a), rumqttc::TlsConfiguration::Rustls(b)) = (a, b)
691 : else {
692 0 : panic!("expected the injected-rustls variant");
693 : };
694 2 : assert!(
695 2 : std::sync::Arc::ptr_eq(&a, &b),
696 : "each call built a fresh trust store instead of sharing one"
697 : );
698 2 : }
699 :
700 : /// The MQTT destination is client-supplied (`notification.endpoint.uri`,
701 : /// 7.2), so it is an egress target: the cloud instance-metadata range is
702 : /// refused before any socket is opened, whatever the private-egress
703 : /// switch says, and the refusal must not echo the endpoint credentials.
704 : #[tokio::test]
705 2 : async fn deliver_refuses_instance_metadata_endpoint() {
706 2 : let ep = MqttEndpoint::parse("mqtt://user:hunter2@169.254.169.254/t").expect("parse");
707 2 : let sink = MqttSink::new(2, Duration::from_millis(250));
708 2 : let err = sink
709 2 : .publish_message(&ep, MqttParams::default(), b"{}")
710 2 : .await
711 2 : .expect_err("the metadata range must never be dialled");
712 2 : let msg = err.to_string();
713 2 : assert!(
714 2 : msg.contains("denied"),
715 : "expected an egress denial, got: {msg}"
716 : );
717 2 : assert!(
718 2 : !msg.contains("hunter2"),
719 2 : "the endpoint password leaked into the delivery error: {msg}"
720 2 : );
721 2 : }
722 :
723 : /// Egress classification, restated from the HTTP side's policy: the
724 : /// metadata range is refused unconditionally, private ranges only when
725 : /// the deployment switched private egress off, and IPv4-mapped IPv6
726 : /// spellings are judged as their IPv4 selves.
727 : #[tokio::test]
728 2 : async fn checked_addr_applies_the_egress_rules_to_resolved_addresses() {
729 2 : let d = Duration::from_secs(2);
730 : // metadata: denied with private egress ALLOWED (the default)
731 : // the 6to4 spelling too (IETF RFC 3056: the two segments after the
732 : // 2002:: prefix ARE the IPv4 destination), which proves this binding
733 : // goes through the shared classifier rather than its own check
734 8 : for host in [
735 2 : "169.254.169.254",
736 2 : "::ffff:169.254.169.254",
737 2 : "fd00:ec2::254",
738 2 : "2002:a9fe:a9fe::",
739 2 : ] {
740 8 : let e = checked_addr(host, 1883, true, d)
741 8 : .await
742 8 : .expect_err("metadata range must be refused whatever the switch says");
743 8 : assert!(e.to_string().contains("denied"), "{host}: {e}");
744 : }
745 : // loopback and RFC 1918: allowed by default, refused when the
746 : // deployment turns private egress off
747 10 : for host in [
748 2 : "127.0.0.1",
749 2 : "::ffff:127.0.0.1",
750 2 : "10.1.2.3",
751 2 : "::1",
752 2 : "localhost",
753 2 : ] {
754 10 : let ok = checked_addr(host, 1883, true, d)
755 10 : .await
756 10 : .unwrap_or_else(|e| panic!("{host} must be reachable by default: {e}"));
757 10 : assert_eq!(ok.port(), 1883);
758 10 : assert!(
759 10 : checked_addr(host, 1883, false, d).await.is_err(),
760 : "{host} must be refused with private egress off"
761 : );
762 : }
763 : // a public literal clears the strict policy and comes back as the
764 : // ADDRESS to dial — the resolution the check judged, pinned
765 2 : let a = checked_addr("93.184.216.34", 8883, false, d)
766 2 : .await
767 2 : .expect("public address allowed");
768 2 : assert_eq!(a.to_string(), "93.184.216.34:8883");
769 : // a name that cannot be resolved is a denial, not a pass-through
770 2 : assert!(
771 2 : checked_addr("no-such-host.invalid", 1883, true, d)
772 2 : .await
773 2 : .is_err(),
774 2 : "an unresolvable destination must not be dialled"
775 2 : );
776 2 : }
777 :
778 : /// The address the policy judged is the address dialled — except for
779 : /// mqtts, where the certificate is verified against the host NAME.
780 : #[test]
781 2 : fn dial_host_pins_the_address_and_keeps_the_tls_name() {
782 2 : let plain = MqttEndpoint::parse("mqtt://broker.example/t").expect("plain");
783 2 : let addr = "203.0.113.7:1883".parse().expect("addr");
784 2 : assert_eq!(dial_host(&plain, addr), "203.0.113.7");
785 2 : let secure = MqttEndpoint::parse("mqtts://broker.example/t").expect("secure");
786 2 : assert_eq!(
787 2 : dial_host(&secure, addr),
788 : "broker.example",
789 : "mqtts must dial the name so the certificate name check still applies"
790 : );
791 : // IPv6 comes back unbracketed, which is what rumqttc resolves
792 2 : let v6 = "[2001:db8::1]:1883".parse().expect("v6 addr");
793 2 : assert_eq!(dial_host(&plain, v6), "2001:db8::1");
794 2 : }
795 :
796 : #[test]
797 2 : fn message_wrapper_shape() {
798 2 : let body = json!({"id": "urn:n:1", "type": "Notification"});
799 2 : let m = build_message(
800 2 : &body,
801 2 : "application/json",
802 2 : Some("<https://ctx>; rel=\"http://www.w3.org/ns/json-ld#context\""),
803 2 : &[("MyKey".into(), "MyValue".into())],
804 : );
805 2 : assert_eq!(m["body"], body);
806 2 : assert_eq!(m["metadata"]["Content-Type"], "application/json");
807 2 : assert!(m["metadata"]["Link"]
808 2 : .as_str()
809 2 : .expect("link present")
810 2 : .contains("json-ld#context"));
811 2 : assert_eq!(m["metadata"]["MyKey"], "MyValue");
812 :
813 : // ld+json: @context is in the body, no Link in metadata (Table 7.2-2)
814 2 : let m = build_message(&body, "application/ld+json", Some("<x>"), &[]);
815 2 : assert_eq!(m["metadata"]["Content-Type"], "application/ld+json");
816 2 : assert!(m["metadata"].get("Link").is_none());
817 2 : }
818 :
819 : /// Table 7.2-2 names the source of two metadata keys: `Content-Type`
820 : /// comes from `endpoint.accept` and `Link` from the served @context.
821 : /// receiverInfo entries are added "additionally", so one keyed like a
822 : /// table parameter cannot become that parameter's value — a subscriber
823 : /// would otherwise decide the MIME type of its own notifications
824 : /// against 6.3.8's "changed by means of the endpoint.accept member",
825 : /// and could strip the @context reference the clause requires.
826 : #[test]
827 2 : fn receiver_info_cannot_take_over_a_table_7_2_2_parameter() {
828 2 : let body = json!({"id": "urn:n:1", "type": "Notification"});
829 2 : let link = "<https://ctx>; rel=\"http://www.w3.org/ns/json-ld#context\"";
830 2 : let m = build_message(
831 2 : &body,
832 2 : "application/json",
833 2 : Some(link),
834 2 : &[
835 2 : ("Content-Type".into(), "text/plain".into()),
836 2 : ("Link".into(), "<https://evil>".into()),
837 2 : ("NGSILD-Tenant".into(), "acme".into()),
838 2 : ],
839 : );
840 2 : assert_eq!(m["metadata"]["Content-Type"], "application/json");
841 2 : assert_eq!(m["metadata"]["Link"], link);
842 : // every other pair is still added, this one by the caller (6.3.22)
843 2 : assert_eq!(m["metadata"]["NGSILD-Tenant"], "acme");
844 2 : }
845 : }
|
