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 crate::build_message;
11 : use antares_jsonld::loader::EgressPolicy;
12 : use antares_model::NgsiError;
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 : /// Debug prints the password as `[redacted]`: it is the subscriber's
25 : /// credential.
26 : #[derive(Clone, PartialEq)]
27 : pub struct MqttEndpoint {
28 : pub secure: bool,
29 : pub username: Option<String>,
30 : pub password: Option<String>,
31 : pub host: String,
32 : pub port: u16,
33 : pub topic: String,
34 : }
35 :
36 : impl std::fmt::Debug for MqttEndpoint {
37 4 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38 4 : f.debug_struct("MqttEndpoint")
39 4 : .field("secure", &self.secure)
40 4 : .field("username", &self.username)
41 4 : .field("password", &self.password.as_ref().map(|_| "[redacted]"))
42 4 : .field("host", &self.host)
43 4 : .field("port", &self.port)
44 4 : .field("topic", &self.topic)
45 4 : .finish()
46 4 : }
47 : }
48 :
49 : impl MqttEndpoint {
50 : /// 7.2 endpoint URI syntax. A URI that does not meet it fails the
51 : /// 5.2.15 restrictions, so the caller raises BadRequestData — 400 per
52 : /// Table 6.3.2-1. The message names only the redacted URI: the
53 : /// credentials 7.2 permits in the userinfo never reach the response body.
54 58 : pub fn parse(uri: &str) -> Result<Self, NgsiError> {
55 58 : let safe = redacted(uri);
56 : // IETF RFC 3986 3.1: scheme names are case-insensitive, and
57 : // `SinkRegistry::scheme_of` already lowercases to pick this sink.
58 58 : let (secure, rest) = match uri.split_once("://") {
59 58 : Some((s, r)) if s.eq_ignore_ascii_case("mqtts") => (true, r),
60 50 : Some((s, r)) if s.eq_ignore_ascii_case("mqtt") => (false, r),
61 4 : _ => return Err(bad(format!("not an mqtt(s) endpoint URI: {safe:?}"))),
62 : };
63 54 : let (authority, topic) = rest
64 54 : .split_once('/')
65 54 : .ok_or_else(|| bad(format!("mqtt endpoint {safe:?} has no topic")))?;
66 50 : if topic.is_empty() {
67 4 : return Err(bad(format!("mqtt endpoint {safe:?} has no topic")));
68 46 : }
69 46 : let (userinfo, hostport) = match authority.rsplit_once('@') {
70 22 : Some((u, h)) => (Some(u), h),
71 24 : None => (None, authority),
72 : };
73 46 : let (username, password) = match userinfo {
74 24 : None => (None, None),
75 22 : Some(u) => match u.split_once(':') {
76 16 : Some((user, pass)) => (Some(user.to_owned()), Some(pass.to_owned())),
77 6 : None => (Some(u.to_owned()), None),
78 : },
79 : };
80 : // Deliberately no IPv6-literal hosts — the binding's URI convention
81 : // (i.19) and the ETSI suite use hostnames; add bracket parsing when a
82 : // deployment needs it.
83 46 : let (host, port) = match hostport.split_once(':') {
84 14 : Some((h, p)) => (
85 14 : h.to_owned(),
86 14 : p.parse::<u16>()
87 14 : .map_err(|_| bad(format!("invalid mqtt port in {safe:?}")))?,
88 : ),
89 32 : None => (hostport.to_owned(), if secure { 8883 } else { 1883 }),
90 : };
91 42 : if host.is_empty() {
92 4 : return Err(bad(format!("mqtt endpoint {safe:?} has no host")));
93 38 : }
94 38 : Ok(Self {
95 38 : secure,
96 38 : username,
97 38 : password,
98 38 : host,
99 38 : port,
100 38 : topic: topic.to_owned(),
101 38 : })
102 58 : }
103 : }
104 :
105 : /// Table 7.2-1 protocol parameters from `notification.endpoint.notifierInfo`.
106 : #[derive(Debug, Clone, Copy, PartialEq)]
107 : pub struct MqttParams {
108 : pub qos: u8,
109 : pub v5: bool,
110 : }
111 :
112 : impl Default for MqttParams {
113 12 : fn default() -> Self {
114 12 : Self { qos: 0, v5: true } // defaults per Table 7.2-1: QoS 0, mqtt5.0
115 12 : }
116 : }
117 :
118 : impl MqttParams {
119 8 : pub fn from_notifier_info<'a>(
120 8 : pairs: impl IntoIterator<Item = (&'a str, &'a str)>,
121 8 : ) -> Result<Self, NgsiError> {
122 8 : let mut p = Self::default();
123 8 : for (k, v) in pairs {
124 8 : match k {
125 8 : "MQTT-QoS" => {
126 4 : p.qos = match v {
127 4 : "0" => 0,
128 4 : "1" => 1,
129 4 : "2" => 2,
130 2 : _ => return Err(bad(format!("MQTT-QoS must be 0, 1 or 2 (got {v:?})"))),
131 : }
132 : }
133 4 : "MQTT-Version" => {
134 4 : p.v5 = match v {
135 4 : "mqtt5.0" => true,
136 4 : "mqtt3.1.1" => false,
137 : _ => {
138 2 : return Err(bad(format!(
139 2 : "MQTT-Version must be mqtt3.1.1 or mqtt5.0 (got {v:?})"
140 2 : )))
141 : }
142 : }
143 : }
144 0 : _ => {} // unknown notifierInfo keys are not ours to police
145 : }
146 : }
147 4 : Ok(p)
148 8 : }
149 : }
150 :
151 : /// The key a pooled MQTT connection is shared under. Everything that changes
152 : /// the identity the connection authenticates as (or how) must participate:
153 : /// two subscriptions whose endpoints differ only in password must never
154 : /// reuse one another's authenticated connection. Keys are map keys only:
155 : /// never log them.
156 20 : fn pool_key(ep: &MqttEndpoint, params: MqttParams) -> String {
157 : // The password participates via a one-way hash so the plaintext never
158 : // sits in a map key. DefaultHasher collisions are acceptable here — a
159 : // collision only merges two pool slots, it does not skip broker-side
160 : // authentication.
161 : use std::hash::{Hash, Hasher};
162 20 : let mut h = std::collections::hash_map::DefaultHasher::new();
163 20 : ep.password.hash(&mut h);
164 20 : format!(
165 : "{}:{:016x}:{}@{}:{}/v{}",
166 20 : ep.username.as_deref().unwrap_or(""),
167 20 : h.finish(),
168 : ep.secure,
169 : ep.host,
170 : ep.port,
171 20 : if params.v5 { 5 } else { 3 }
172 : )
173 20 : }
174 :
175 : /// The client options of one MQTT endpoint, for either protocol version
176 : /// (rumqttc has one options type per version, with the same setters): 30 s
177 : /// keep-alive, the endpoint's credentials, TLS when it is `mqtts`.
178 : macro_rules! options {
179 : ($ty:ty, $id:expr, $dial:expr, $port:expr, $ep:expr) => {{
180 : let mut opts = <$ty>::new($id, $dial, $port);
181 : opts.set_keep_alive(Duration::from_secs(30));
182 : if let Some(u) = &$ep.username {
183 : opts.set_credentials(u, $ep.password.as_deref().unwrap_or(""));
184 : }
185 : if $ep.secure {
186 : opts.set_transport(rumqttc::Transport::Tls(shared_tls_config()?));
187 : }
188 : opts
189 : }};
190 : }
191 :
192 : /// The mqtts trust store, built ONCE per process. Loading the platform
193 : /// certificate store on every connect is wasted work, and rumqttc's
194 : /// `TlsConfiguration::default()` panics when the store is unreadable — a
195 : /// bad cert bundle must fail the one delivery, not the broker.
196 4 : fn shared_tls_config() -> Result<rumqttc::TlsConfiguration, NgsiError> {
197 : static TLS: std::sync::OnceLock<Option<rumqttc::TlsConfiguration>> = std::sync::OnceLock::new();
198 4 : TLS.get_or_init(|| {
199 : // `TlsConfiguration::default()` is the only rumqttc constructor
200 : // that loads the platform trust store, and it panics on failure —
201 : // contain that so a broken cert bundle degrades to failed mqtts
202 : // deliveries instead of killing the process. The failure is cached:
203 : // a store unreadable at first use will not become readable later.
204 2 : std::panic::catch_unwind(rumqttc::TlsConfiguration::default)
205 2 : .map_err(|_| tracing::error!("mqtts: loading the platform certificate store failed"))
206 2 : .ok()
207 2 : })
208 4 : .clone()
209 4 : .ok_or_else(|| NgsiError::InternalError("mqtts trust store unavailable".into()))
210 4 : }
211 :
212 : /// Is egress to private/loopback ranges allowed? The MQTT destination is
213 : /// client-supplied (`notification.endpoint.uri`, 7.2), so it is governed by
214 : /// the same deployment switch, and judged by the same address classifiers,
215 : /// as the HTTP callbacks and @context fetches: `EgressPolicy` is the one
216 : /// copy of both. Private egress is allowed by default (dev boxes, compose
217 : /// stacks and the conformance mocks all live there) and
218 : /// `ANTARES_EGRESS_ALLOW_PRIVATE=false` turns the deny on for
219 : /// internet-exposed deployments, with `ANTARES_EGRESS_PRIVATE_HOSTS` naming
220 : /// the peers it still reaches. Read once per process.
221 8 : fn egress_policy() -> EgressPolicy {
222 : static POLICY: std::sync::OnceLock<EgressPolicy> = std::sync::OnceLock::new();
223 8 : *POLICY.get_or_init(EgressPolicy::from_env)
224 8 : }
225 :
226 : /// Resolve the endpoint host ONCE and return the address to dial, or a
227 : /// denial. Resolving for the check and then handing the NAME to the MQTT
228 : /// client would leave a window in which the answer changes (DNS rebinding):
229 : /// the connector would dial an address the policy never saw. The address
230 : /// this returns is the address dialled, so check and connect see the same
231 : /// answer by construction. A host that cannot be resolved is a DENIAL — a
232 : /// destination the policy could not judge is never dialled — and the
233 : /// resolver runs under the sink's own deadline.
234 40 : async fn checked_addr(
235 40 : host: &str,
236 40 : port: u16,
237 40 : policy: EgressPolicy,
238 40 : dns_timeout: Duration,
239 40 : ) -> Result<std::net::SocketAddr, NgsiError> {
240 40 : let denied = || {
241 20 : NgsiError::InternalError(format!(
242 20 : "mqtt egress to {host}:{port} denied (instance metadata or private range)"
243 20 : ))
244 20 : };
245 40 : let addrs: Vec<std::net::SocketAddr> = match host
246 40 : .trim_matches(['[', ']'])
247 40 : .parse::<std::net::IpAddr>()
248 : {
249 : // an IP literal needs no resolver, and is judged by the same rules
250 34 : Ok(ip) => vec![std::net::SocketAddr::new(ip, port)],
251 6 : Err(_) => tokio::time::timeout(dns_timeout, tokio::net::lookup_host((host, port)))
252 6 : .await
253 6 : .map_err(|_| {
254 0 : NgsiError::InternalError(format!("mqtt egress: resolving {host} timed out"))
255 0 : })?
256 6 : .map_err(|e| NgsiError::InternalError(format!("mqtt egress: resolving {host}: {e}")))?
257 4 : .collect(),
258 : };
259 38 : addrs
260 38 : .into_iter()
261 40 : .find(|a| policy.admits(host, a.ip()))
262 38 : .ok_or_else(denied)
263 40 : }
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 12 : fn dial_host(ep: &MqttEndpoint, addr: std::net::SocketAddr) -> String {
275 12 : if ep.secure {
276 2 : ep.host.clone()
277 : } else {
278 10 : addr.ip().to_string()
279 : }
280 12 : }
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 : /// One `()` per PUBACK or PUBCOMP the broker sent: a connection is
292 : /// checked out by one publish at a time, so the next one answers it.
293 : acks: tokio::sync::mpsc::Receiver<()>,
294 : last_used: Instant,
295 : }
296 :
297 : impl Drop for Conn {
298 6 : fn drop(&mut self) {
299 6 : self.pump.abort();
300 6 : }
301 : }
302 :
303 : /// MQTT delivery with a bounded per-endpoint connection pool (bounded
304 : /// WITH eviction; timeouts fixed at construction).
305 : pub struct MqttSink {
306 : pool: Mutex<HashMap<String, Conn>>,
307 : cap: usize,
308 : timeout: Duration,
309 : }
310 :
311 : impl Default for MqttSink {
312 4816 : fn default() -> Self {
313 4816 : Self::new(32, Duration::from_secs(5))
314 4816 : }
315 : }
316 :
317 : impl MqttSink {
318 4820 : pub fn new(cap: usize, timeout: Duration) -> Self {
319 4820 : Self {
320 4820 : pool: Mutex::new(HashMap::new()),
321 4820 : cap,
322 4820 : timeout,
323 4820 : }
324 4820 : }
325 :
326 : /// Deliver one notification message. `message` is the 7.2 wrapper from
327 : /// [`build_message`], serialized by the caller once per subscription.
328 6 : pub async fn publish_message(
329 6 : &self,
330 6 : ep: &MqttEndpoint,
331 6 : params: MqttParams,
332 6 : message: &[u8],
333 6 : ) -> Result<(), NgsiError> {
334 6 : let key = pool_key(ep, params);
335 : // one retry with a fresh connection: a pooled client whose broker
336 : // restarted fails the first publish; a dead broker fails both.
337 8 : for attempt in 0..2 {
338 8 : let mut conn = match self.checkout(&key) {
339 0 : Some(c) => c,
340 8 : None => self.connect(ep, params).await?,
341 : };
342 : // an ack a timed-out publish left behind answers nothing now
343 6 : while conn.acks.try_recv().is_ok() {}
344 6 : let (client, acks) = (&conn.client, &mut conn.acks);
345 : // QoS 1 and 2 are delivered once the broker acknowledged them
346 : // (PUBACK, PUBCOMP), not once rumqttc queued them
347 6 : let published = tokio::time::timeout(self.timeout, async {
348 6 : Self::publish(client, &ep.topic, params.qos, message).await?;
349 6 : if params.qos > 0 && acks.recv().await.is_none() {
350 0 : return Err("connection lost before the broker acknowledged".to_owned());
351 2 : }
352 2 : Ok(())
353 2 : })
354 6 : .await;
355 0 : match published {
356 2 : Ok(Ok(())) if !conn.pump.is_finished() => {
357 2 : self.checkin(key, conn);
358 2 : return Ok(());
359 : }
360 4 : _ if attempt == 0 => continue, // drop conn, retry fresh
361 : Ok(Ok(())) => {
362 0 : return Err(NgsiError::InternalError(
363 0 : "mqtt connection lost during publish".into(),
364 0 : ))
365 : }
366 0 : Ok(Err(e)) => return Err(NgsiError::InternalError(format!("mqtt publish: {e}"))),
367 : Err(_) => {
368 2 : return Err(NgsiError::InternalError(format!(
369 2 : "mqtt publish to {}:{} timed out",
370 2 : ep.host, ep.port
371 2 : )))
372 : }
373 : }
374 : }
375 0 : Err(NgsiError::InternalError(
376 0 : "mqtt publish exhausted its retry without a verdict".into(),
377 0 : ))
378 6 : }
379 :
380 6 : async fn publish(client: &Client, topic: &str, qos: u8, payload: &[u8]) -> Result<(), String> {
381 6 : match client {
382 6 : Client::V3(c) => {
383 6 : let qos = rumqttc::qos(qos).map_err(|e| e.to_string())?;
384 6 : c.publish(topic, qos, false, payload.to_vec())
385 6 : .await
386 6 : .map_err(|e| e.to_string())
387 : }
388 0 : Client::V5(c) => {
389 0 : let qos =
390 0 : rumqttc::v5::mqttbytes::qos(qos).ok_or_else(|| format!("invalid QoS {qos}"))?;
391 0 : c.publish(topic, qos, false, payload.to_vec())
392 0 : .await
393 0 : .map_err(|e| e.to_string())
394 : }
395 : }
396 6 : }
397 :
398 8 : fn checkout(&self, key: &str) -> Option<Conn> {
399 8 : self.pool
400 8 : .lock()
401 8 : .unwrap_or_else(std::sync::PoisonError::into_inner)
402 8 : .remove(key)
403 8 : }
404 :
405 2 : fn checkin(&self, key: String, mut conn: Conn) {
406 2 : conn.last_used = Instant::now();
407 2 : let mut pool = self
408 2 : .pool
409 2 : .lock()
410 2 : .unwrap_or_else(std::sync::PoisonError::into_inner);
411 2 : pool.retain(|_, c| !c.pump.is_finished());
412 2 : pool.insert(key, conn);
413 : // bounded with eviction: drop the least-recently-used overflow.
414 2 : while pool.len() > self.cap {
415 0 : if let Some(oldest) = pool
416 0 : .iter()
417 0 : .min_by_key(|(_, c)| c.last_used)
418 0 : .map(|(k, _)| k.clone())
419 0 : {
420 0 : pool.remove(&oldest);
421 0 : }
422 : }
423 2 : }
424 :
425 : /// Connect and wait for ConnAck (a dead broker must fail delivery, not
426 : /// queue forever), then hand the event loop to a pump task.
427 8 : async fn connect(&self, ep: &MqttEndpoint, params: MqttParams) -> Result<Conn, NgsiError> {
428 : // A monotonic counter, not a timestamp: `Instant::now().elapsed()` is
429 : // ~0 for every caller, so two connects in one process would claim the
430 : // same client id and the broker would kick the older session.
431 : static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
432 8 : let id = format!(
433 : "antares-{}-{}",
434 8 : std::process::id(),
435 8 : SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
436 : );
437 8 : let refused = |e: String| {
438 0 : NgsiError::InternalError(format!("mqtt connect {}:{}: {e}", ep.host, ep.port))
439 0 : };
440 : // Egress policy first: resolve once, judge the answer, dial what was
441 : // judged. Nothing below opens a socket to an unchecked destination.
442 8 : let addr = checked_addr(&ep.host, ep.port, egress_policy(), self.timeout).await?;
443 6 : let dial = dial_host(ep, addr);
444 6 : if params.v5 {
445 0 : let opts = options!(rumqttc::v5::MqttOptions, id, dial, addr.port(), ep);
446 0 : let (client, mut eventloop) = rumqttc::v5::AsyncClient::new(opts, 16);
447 0 : tokio::time::timeout(self.timeout, async {
448 : loop {
449 0 : match eventloop.poll().await {
450 : Ok(rumqttc::v5::Event::Incoming(
451 : rumqttc::v5::mqttbytes::v5::Packet::ConnAck(_),
452 0 : )) => return Ok(()),
453 0 : Ok(_) => {}
454 0 : Err(e) => return Err(e.to_string()),
455 : }
456 : }
457 0 : })
458 0 : .await
459 0 : .map_err(|_| refused("connect timeout".into()))?
460 0 : .map_err(refused)?;
461 0 : let (ack, acks) = tokio::sync::mpsc::channel(16);
462 0 : let pump = tokio::spawn(async move {
463 0 : while let Ok(event) = eventloop.poll().await {
464 0 : if matches!(
465 0 : event,
466 : rumqttc::v5::Event::Incoming(
467 : rumqttc::v5::mqttbytes::v5::Packet::PubAck(_)
468 : | rumqttc::v5::mqttbytes::v5::Packet::PubComp(_)
469 : )
470 0 : ) {
471 0 : let _ = ack.try_send(());
472 0 : }
473 : }
474 0 : });
475 0 : Ok(Conn {
476 0 : client: Client::V5(client),
477 0 : pump,
478 0 : acks,
479 0 : last_used: Instant::now(),
480 0 : })
481 : } else {
482 6 : let opts = options!(rumqttc::MqttOptions, id, dial, addr.port(), ep);
483 6 : let (client, mut eventloop) = rumqttc::AsyncClient::new(opts, 16);
484 6 : tokio::time::timeout(self.timeout, async {
485 : loop {
486 6 : match eventloop.poll().await {
487 6 : Ok(rumqttc::Event::Incoming(rumqttc::Packet::ConnAck(_))) => return Ok(()),
488 0 : Ok(_) => {}
489 0 : Err(e) => return Err(e.to_string()),
490 : }
491 : }
492 6 : })
493 6 : .await
494 6 : .map_err(|_| refused("connect timeout".into()))?
495 6 : .map_err(refused)?;
496 6 : let (ack, acks) = tokio::sync::mpsc::channel(16);
497 6 : let pump = tokio::spawn(async move {
498 14 : while let Ok(event) = eventloop.poll().await {
499 6 : if matches!(
500 2 : event,
501 : rumqttc::Event::Incoming(
502 : rumqttc::Packet::PubAck(_) | rumqttc::Packet::PubComp(_)
503 : )
504 2 : ) {
505 2 : let _ = ack.try_send(());
506 6 : }
507 : }
508 0 : });
509 6 : Ok(Conn {
510 6 : client: Client::V3(client),
511 6 : pump,
512 6 : acks,
513 6 : last_used: Instant::now(),
514 6 : })
515 : }
516 8 : }
517 : }
518 :
519 : /// One message an inbound subscription received: the topic it was
520 : /// published on and its payload, as sent.
521 : #[derive(Debug, Clone, PartialEq)]
522 : pub struct Inbound {
523 : pub topic: String,
524 : pub payload: Vec<u8>,
525 : }
526 :
527 : /// Subscribe to `ep.topic` and hand every message published there to `tx`,
528 : /// for an inbound bridge Channel. MQTT 5 only, at QoS 1, under the
529 : /// shared subscription `$share/{group}/{topic}` (MQTT 5.0 4.8.2): every
530 : /// broker replica binds the same Channels, and the server hands each message
531 : /// to one of them, so a sample is written once whatever the replica count.
532 : ///
533 : /// The destination is judged by the egress policy before the first connect
534 : /// and the checked address is dialled, as for a notification. The event loop
535 : /// runs on the returned task: it reconnects after a dropped connection and
536 : /// subscribes again on every ConnAck (a clean start keeps no subscription).
537 : /// A full `tx` drops the message rather than stalling the event loop's
538 : /// keep-alive. Aborting the task closes the subscription.
539 : ///
540 : /// # Errors
541 : /// The egress denial, or a broker that does not ConnAck within `timeout`.
542 0 : pub async fn listen(
543 0 : ep: &MqttEndpoint,
544 0 : group: &str,
545 0 : tx: tokio::sync::mpsc::Sender<Inbound>,
546 0 : timeout: Duration,
547 0 : ) -> Result<tokio::task::JoinHandle<()>, NgsiError> {
548 : use rumqttc::v5::mqttbytes::v5::Packet;
549 : use rumqttc::v5::mqttbytes::QoS;
550 : use rumqttc::v5::Event;
551 : static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
552 0 : let id = format!(
553 : "antares-in-{}-{}",
554 0 : std::process::id(),
555 0 : SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
556 : );
557 0 : let addr = checked_addr(&ep.host, ep.port, egress_policy(), timeout).await?;
558 0 : let opts = options!(
559 : rumqttc::v5::MqttOptions,
560 0 : id,
561 0 : dial_host(ep, addr),
562 0 : addr.port(),
563 : ep
564 : );
565 0 : let (client, mut eventloop) = rumqttc::v5::AsyncClient::new(opts, 16);
566 0 : let filter = format!("$share/{group}/{}", ep.topic);
567 0 : let refused =
568 0 : |e: String| NgsiError::InternalError(format!("mqtt connect {}:{}: {e}", ep.host, ep.port));
569 0 : tokio::time::timeout(timeout, async {
570 : loop {
571 0 : match eventloop.poll().await {
572 0 : Ok(Event::Incoming(Packet::ConnAck(_))) => return Ok(()),
573 0 : Ok(_) => {}
574 0 : Err(e) => return Err(e.to_string()),
575 : }
576 : }
577 0 : })
578 0 : .await
579 0 : .map_err(|_| refused("connect timeout".into()))?
580 0 : .map_err(refused)?;
581 0 : client
582 0 : .subscribe(filter.clone(), QoS::AtLeastOnce)
583 0 : .await
584 0 : .map_err(|e| refused(e.to_string()))?;
585 0 : let (host, port) = (ep.host.clone(), ep.port);
586 0 : Ok(tokio::spawn(async move {
587 : loop {
588 0 : match eventloop.poll().await {
589 0 : Ok(Event::Incoming(Packet::Publish(p))) => {
590 0 : let msg = Inbound {
591 0 : topic: String::from_utf8_lossy(&p.topic).into_owned(),
592 0 : payload: p.payload.to_vec(),
593 0 : };
594 0 : match tx.try_send(msg) {
595 0 : Ok(()) => {}
596 0 : Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => return,
597 : Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
598 0 : tracing::debug!(
599 : "mqtt inbound {host}:{port}: queue full, message dropped"
600 : );
601 : }
602 : }
603 : }
604 : Ok(Event::Incoming(Packet::ConnAck(_))) => {
605 0 : if client
606 0 : .subscribe(filter.clone(), QoS::AtLeastOnce)
607 0 : .await
608 0 : .is_err()
609 : {
610 0 : return;
611 0 : }
612 : }
613 0 : Ok(_) => {}
614 0 : Err(e) => {
615 0 : tracing::warn!("mqtt inbound {host}:{port}: {e}; reconnecting in 5 s");
616 0 : tokio::time::sleep(Duration::from_secs(5)).await;
617 : }
618 : }
619 : }
620 0 : }))
621 0 : }
622 :
623 : /// Clause 7: the mqtt(s) binding. `parse_endpoint` holds Table 7.2-1 (the
624 : /// endpoint URI syntax and the notifier parameters) at subscription
625 : /// creation; `deliver` publishes the 7.2 message — `{"metadata": …,
626 : /// "body": <Notification>}` — on the endpoint's topic.
627 : impl crate::NotificationSink for MqttSink {
628 662 : fn schemes(&self) -> &'static [&'static str] {
629 662 : &["mqtt", "mqtts"]
630 662 : }
631 :
632 0 : fn parse_endpoint(&self, uri: &str, notifier_info: &[(&str, &str)]) -> Result<(), NgsiError> {
633 0 : MqttEndpoint::parse(uri)?;
634 0 : MqttParams::from_notifier_info(notifier_info.iter().copied())?;
635 0 : Ok(())
636 0 : }
637 :
638 0 : fn deliver<'a>(
639 0 : &'a self,
640 0 : uri: &'a str,
641 0 : out: &'a crate::Outbound,
642 0 : _timeout: Duration,
643 0 : ) -> crate::DeliveryFuture<'a> {
644 0 : Box::pin(async move {
645 : // Creation validated the endpoint (7.2 / Table 7.2-1); a parse
646 : // failure here means a hand-edited row. The message names only
647 : // the redacted URI: 7.2 permits credentials in the userinfo.
648 0 : let Ok(ep) = MqttEndpoint::parse(uri) else {
649 0 : return Err(crate::DeliveryError::failed(format!(
650 0 : "mqtt endpoint {:?} unusable",
651 0 : redacted(uri)
652 0 : )));
653 : };
654 0 : let Ok(params) = MqttParams::from_notifier_info(out.notifier_pairs()) else {
655 0 : return Err(crate::DeliveryError::failed(format!(
656 0 : "mqtt notifierInfo of {:?} unusable",
657 0 : redacted(uri)
658 0 : )));
659 : };
660 0 : let msg = build_message(&out.body, &out.accept, Some(&out.link), &out.receiver_info);
661 0 : let bytes = antares_model::ordered_vec(&msg);
662 0 : self.publish_message(&ep, params, &bytes)
663 0 : .await
664 : // broker/socket-level failure — keep the timeout guard
665 0 : .map_err(|e| crate::DeliveryError::timeout(e.to_string()))
666 0 : })
667 0 : }
668 : }
669 :
670 : #[cfg(test)]
671 : mod tests {
672 : use super::*;
673 : use serde_json::json;
674 :
675 : #[test]
676 2 : fn debug_never_prints_the_password() {
677 2 : let e = MqttEndpoint::parse("mqtt://u:s3cret@host:9001/t").expect("user+pass");
678 2 : let shown = format!("{e:?}");
679 2 : assert!(!shown.contains("s3cret"), "{shown}");
680 2 : assert!(
681 2 : shown.contains("[redacted]") && shown.contains("\"u\"") && shown.contains("host"),
682 : "{shown}"
683 : );
684 2 : let none = format!(
685 : "{:?}",
686 2 : MqttEndpoint::parse("mqtt://u@host/t").expect("user")
687 : );
688 2 : assert!(none.contains("password: None"), "{none}");
689 2 : }
690 :
691 : #[test]
692 2 : fn parses_endpoint_variants() {
693 2 : let e = MqttEndpoint::parse("mqtt://host/topic").expect("plain");
694 2 : assert_eq!(
695 : e,
696 2 : MqttEndpoint {
697 2 : secure: false,
698 2 : username: None,
699 2 : password: None,
700 2 : host: "host".into(),
701 2 : port: 1883,
702 2 : topic: "topic".into()
703 2 : }
704 : );
705 2 : let e = MqttEndpoint::parse("mqtt://host:8085/a/b/c").expect("port+subtopics");
706 2 : assert_eq!(e.port, 8085);
707 2 : assert_eq!(e.topic, "a/b/c");
708 2 : let e = MqttEndpoint::parse("mqtt://user@host/t").expect("user");
709 2 : assert_eq!(e.username.as_deref(), Some("user"));
710 2 : assert_eq!(e.password, None);
711 2 : let e = MqttEndpoint::parse("mqtt://u:p@host:9001/t").expect("user+pass+port");
712 2 : assert_eq!(e.username.as_deref(), Some("u"));
713 2 : assert_eq!(e.password.as_deref(), Some("p"));
714 2 : assert_eq!(e.port, 9001);
715 2 : let e = MqttEndpoint::parse("mqtts://host/t").expect("tls");
716 2 : assert!(e.secure);
717 2 : assert_eq!(e.port, 8883, "mqtts default port");
718 2 : }
719 :
720 : /// IETF RFC 3986 3.1: scheme names are case-insensitive.
721 : /// `SinkRegistry::scheme_of` lowercases before it picks this sink, so an
722 : /// uppercase-scheme endpoint is routed here and this parser decides —
723 : /// rejecting it would 400 an endpoint URI the registry says it serves.
724 : #[test]
725 2 : fn the_scheme_is_matched_case_insensitively() {
726 2 : assert!(
727 2 : MqttEndpoint::parse("MQTT://host/topic")
728 2 : .expect("upper")
729 2 : .port
730 2 : == 1883
731 : );
732 2 : let e = MqttEndpoint::parse("MQTTS://host/t").expect("upper tls");
733 2 : assert!(e.secure);
734 2 : assert_eq!(e.port, 8883);
735 2 : assert!(MqttEndpoint::parse("MqTt://host/t").is_ok());
736 2 : }
737 :
738 : #[test]
739 2 : fn rejects_bad_endpoints() {
740 10 : for uri in [
741 2 : "http://host/topic",
742 2 : "mqtt://host",
743 2 : "mqtt://host/",
744 2 : "mqtt:///topic",
745 2 : "mqtt://host:notaport/t",
746 2 : ] {
747 10 : assert!(MqttEndpoint::parse(uri).is_err(), "{uri} must be rejected");
748 : }
749 2 : }
750 :
751 : /// 7.2 endpoint URIs may carry credentials in the userinfo
752 : /// (`mqtt[s]://<username>:<password>@host`). A parse failure is answered
753 : /// to the client as BadRequestData (5.8.1.4, 400 per Table 6.3.2-1) and
754 : /// the message becomes the 5.5.3 ProblemDetails `detail`, so no password
755 : /// may appear in it.
756 : #[test]
757 2 : fn parse_errors_redact_endpoint_userinfo() {
758 10 : for uri in [
759 2 : "mqtt://user:hunter2@host", // no topic
760 2 : "mqtt://user:hunter2@host/", // empty topic
761 2 : "mqtt://user:hunter2@host:notaport/t", // bad port
762 2 : "mqtts://user:hunter2@/t", // no host
763 2 : "http://user:hunter2@host/t", // wrong scheme
764 2 : ] {
765 10 : let NgsiError::BadRequestData(msg) =
766 10 : MqttEndpoint::parse(uri).expect_err(&format!("{uri} must be rejected"))
767 : else {
768 0 : panic!("{uri} must be BadRequestData (400, Table 6.3.2-1)");
769 : };
770 10 : assert!(
771 10 : !msg.contains("hunter2"),
772 : "the password leaked into the 400 detail: {msg}"
773 : );
774 10 : assert!(
775 10 : !msg.contains("user:"),
776 : "the userinfo leaked into the 400 detail: {msg}"
777 : );
778 : // the detail must still be useful: scheme and host survive
779 10 : assert!(
780 10 : msg.contains("host") || uri.contains("@/"),
781 : "the redacted detail lost the host: {msg}"
782 : );
783 : }
784 2 : }
785 :
786 : #[test]
787 2 : fn notifier_info_defaults_and_validation() {
788 2 : let p = MqttParams::from_notifier_info([]).expect("defaults");
789 2 : assert_eq!(p, MqttParams { qos: 0, v5: true });
790 2 : let p = MqttParams::from_notifier_info([("MQTT-QoS", "2"), ("MQTT-Version", "mqtt3.1.1")])
791 2 : .expect("explicit");
792 2 : assert_eq!(p, MqttParams { qos: 2, v5: false });
793 2 : assert!(MqttParams::from_notifier_info([("MQTT-QoS", "3")]).is_err());
794 2 : assert!(MqttParams::from_notifier_info([("MQTT-Version", "mqtt4")]).is_err());
795 2 : }
796 :
797 : /// The pool key must separate sessions by credentials: same user with
798 : /// two different passwords = two different authenticated principals,
799 : /// which must never share one session. The plaintext password itself
800 : /// must not appear in the key.
801 : #[test]
802 2 : fn pool_key_separates_credentials() {
803 2 : let p = MqttParams::default();
804 2 : let a = MqttEndpoint::parse("mqtt://u:secret-one@host/t").expect("a");
805 2 : let b = MqttEndpoint::parse("mqtt://u:secret-two@host/t").expect("b");
806 2 : let c = MqttEndpoint::parse("mqtt://u:secret-one@host/t").expect("c");
807 2 : assert_ne!(
808 2 : pool_key(&a, p),
809 2 : pool_key(&b, p),
810 : "different passwords must not share an authenticated session"
811 : );
812 2 : assert_eq!(
813 2 : pool_key(&a, p),
814 2 : pool_key(&c, p),
815 : "identical endpoints must keep pooling"
816 : );
817 2 : assert!(
818 2 : !pool_key(&a, p).contains("secret-one"),
819 : "the plaintext password must never appear in the key"
820 : );
821 : // no password vs some password are different principals too
822 2 : let none = MqttEndpoint::parse("mqtt://u@host/t").expect("none");
823 2 : assert_ne!(pool_key(&a, p), pool_key(&none, p));
824 2 : }
825 :
826 : /// The mqtts trust store is loaded once and the same shared rustls
827 : /// config is handed to every connect. (The failure path — an unreadable
828 : /// platform store — cannot be simulated in a unit test; the helper's
829 : /// error contract covers it.)
830 : #[test]
831 2 : fn tls_config_is_built_once_and_shared() {
832 2 : let a = shared_tls_config().expect("first load");
833 2 : let b = shared_tls_config().expect("second load");
834 2 : let (rumqttc::TlsConfiguration::Rustls(a), rumqttc::TlsConfiguration::Rustls(b)) = (a, b)
835 : else {
836 0 : panic!("expected the injected-rustls variant");
837 : };
838 2 : assert!(
839 2 : std::sync::Arc::ptr_eq(&a, &b),
840 : "each call built a fresh trust store instead of sharing one"
841 : );
842 2 : }
843 :
844 : /// A stub MQTT 3.1.1 broker on loopback: it accepts every CONNECT and
845 : /// answers a QoS 1 PUBLISH with its PUBACK only when `ack` holds.
846 4 : async fn stub_broker(ack: bool) -> u16 {
847 : use tokio::io::{AsyncReadExt, AsyncWriteExt};
848 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
849 4 : .await
850 4 : .expect("bind");
851 4 : let port = listener.local_addr().expect("addr").port();
852 4 : tokio::spawn(async move {
853 10 : while let Ok((mut s, _)) = listener.accept().await {
854 6 : tokio::spawn(async move {
855 6 : let mut buf = [0u8; 4096];
856 18 : while let Ok(n) = s.read(&mut buf).await {
857 16 : let Some(packet) = buf.get(..n).filter(|p| !p.is_empty()) else {
858 4 : return;
859 : };
860 12 : let reply: Vec<u8> = match packet.first().map(|b| b >> 4) {
861 6 : Some(1) => vec![0x20, 0x02, 0x00, 0x00],
862 : // header, one-byte length, topic length + topic, pkid
863 2 : Some(3) if ack => {
864 2 : let topic = usize::from(packet.get(3).copied().unwrap_or(0));
865 2 : let at = 4usize.saturating_add(topic);
866 2 : let pkid = packet.get(at..at.saturating_add(2)).unwrap_or(&[0, 1]);
867 2 : [&[0x40, 0x02][..], pkid].concat()
868 : }
869 4 : _ => continue,
870 : };
871 8 : if s.write_all(&reply).await.is_err() {
872 0 : return;
873 8 : }
874 : }
875 4 : });
876 : }
877 0 : });
878 4 : port
879 4 : }
880 :
881 : /// a QoS 1 publish is delivered once the broker sent its
882 : /// PUBACK; a broker that never acknowledges fails it within the timeout.
883 : #[tokio::test]
884 2 : async fn a_qos1_publish_waits_for_the_puback() {
885 2 : let params = MqttParams { qos: 1, v5: false };
886 2 : let sink = MqttSink::new(2, Duration::from_millis(300));
887 2 : let silent =
888 2 : MqttEndpoint::parse(&format!("mqtt://127.0.0.1:{}/t", stub_broker(false).await))
889 2 : .expect("parse");
890 2 : let started = Instant::now();
891 2 : let err = sink
892 2 : .publish_message(&silent, params, b"{}")
893 2 : .await
894 2 : .expect_err("no PUBACK, no delivery");
895 2 : assert!(err.to_string().contains("timed out"), "{err}");
896 2 : assert!(
897 2 : started.elapsed() < Duration::from_secs(3),
898 : "{:?}",
899 0 : started.elapsed()
900 : );
901 2 : let acking =
902 2 : MqttEndpoint::parse(&format!("mqtt://127.0.0.1:{}/t", stub_broker(true).await))
903 2 : .expect("parse");
904 2 : sink.publish_message(&acking, params, b"{}")
905 2 : .await
906 2 : .expect("the PUBACK delivers");
907 2 : }
908 :
909 : /// The MQTT destination is client-supplied (`notification.endpoint.uri`,
910 : /// 7.2), so it is an egress target: the cloud instance-metadata range is
911 : /// refused before any socket is opened, whatever the private-egress
912 : /// switch says, and the refusal must not echo the endpoint credentials.
913 : #[tokio::test]
914 2 : async fn deliver_refuses_instance_metadata_endpoint() {
915 2 : let ep = MqttEndpoint::parse("mqtt://user:hunter2@169.254.169.254/t").expect("parse");
916 2 : let sink = MqttSink::new(2, Duration::from_millis(250));
917 2 : let err = sink
918 2 : .publish_message(&ep, MqttParams::default(), b"{}")
919 2 : .await
920 2 : .expect_err("the metadata range must never be dialled");
921 2 : let msg = err.to_string();
922 2 : assert!(
923 2 : msg.contains("denied"),
924 : "expected an egress denial, got: {msg}"
925 : );
926 2 : assert!(
927 2 : !msg.contains("hunter2"),
928 2 : "the endpoint password leaked into the delivery error: {msg}"
929 2 : );
930 2 : }
931 :
932 : /// Egress classification, restated from the HTTP side's policy: the
933 : /// metadata range is refused unconditionally, private ranges only when
934 : /// the deployment switched private egress off, and IPv4-mapped IPv6
935 : /// spellings are judged as their IPv4 selves.
936 : const OPEN: EgressPolicy = EgressPolicy {
937 : allow_private: true,
938 : private_hosts: &[],
939 : };
940 : const CLOSED: EgressPolicy = EgressPolicy {
941 : allow_private: false,
942 : private_hosts: &[],
943 : };
944 :
945 : #[tokio::test]
946 2 : async fn checked_addr_applies_the_egress_rules_to_resolved_addresses() {
947 2 : let d = Duration::from_secs(2);
948 : // metadata: denied with private egress ALLOWED (the default)
949 : // the 6to4 spelling too (IETF RFC 3056: the two segments after the
950 : // 2002:: prefix ARE the IPv4 destination), which proves this binding
951 : // goes through the shared classifier rather than its own check
952 8 : for host in [
953 2 : "169.254.169.254",
954 2 : "::ffff:169.254.169.254",
955 2 : "fd00:ec2::254",
956 2 : "2002:a9fe:a9fe::",
957 2 : ] {
958 8 : let e = checked_addr(host, 1883, OPEN, d)
959 8 : .await
960 8 : .expect_err("metadata range must be refused whatever the switch says");
961 8 : assert!(e.to_string().contains("denied"), "{host}: {e}");
962 : }
963 : // loopback and RFC 1918: allowed by default, refused when the
964 : // deployment turns private egress off
965 10 : for host in [
966 2 : "127.0.0.1",
967 2 : "::ffff:127.0.0.1",
968 2 : "10.1.2.3",
969 2 : "::1",
970 2 : "localhost",
971 2 : ] {
972 10 : let ok = checked_addr(host, 1883, OPEN, d)
973 10 : .await
974 10 : .unwrap_or_else(|e| panic!("{host} must be reachable by default: {e}"));
975 10 : assert_eq!(ok.port(), 1883);
976 10 : assert!(
977 10 : checked_addr(host, 1883, CLOSED, d).await.is_err(),
978 : "{host} must be refused with private egress off"
979 : );
980 : }
981 : // a public literal clears the strict policy and comes back as the
982 : // ADDRESS to dial — the resolution the check judged, pinned
983 2 : let a = checked_addr("93.184.216.34", 8883, CLOSED, d)
984 2 : .await
985 2 : .expect("public address allowed");
986 2 : assert_eq!(a.to_string(), "93.184.216.34:8883");
987 : // a name that cannot be resolved is a denial, not a pass-through
988 2 : assert!(
989 2 : checked_addr("no-such-host.invalid", 1883, OPEN, d)
990 2 : .await
991 2 : .is_err(),
992 2 : "an unresolvable destination must not be dialled"
993 2 : );
994 2 : }
995 :
996 : /// The address the policy judged is the address dialled — except for
997 : /// mqtts, where the certificate is verified against the host NAME.
998 : #[test]
999 2 : fn dial_host_pins_the_address_and_keeps_the_tls_name() {
1000 2 : let plain = MqttEndpoint::parse("mqtt://broker.example/t").expect("plain");
1001 2 : let addr = "203.0.113.7:1883".parse().expect("addr");
1002 2 : assert_eq!(dial_host(&plain, addr), "203.0.113.7");
1003 2 : let secure = MqttEndpoint::parse("mqtts://broker.example/t").expect("secure");
1004 2 : assert_eq!(
1005 2 : dial_host(&secure, addr),
1006 : "broker.example",
1007 : "mqtts must dial the name so the certificate name check still applies"
1008 : );
1009 : // IPv6 comes back unbracketed, which is what rumqttc resolves
1010 2 : let v6 = "[2001:db8::1]:1883".parse().expect("v6 addr");
1011 2 : assert_eq!(dial_host(&plain, v6), "2001:db8::1");
1012 2 : }
1013 :
1014 : #[test]
1015 2 : fn message_wrapper_shape() {
1016 2 : let body = json!({"id": "urn:n:1", "type": "Notification"});
1017 2 : let m = build_message(
1018 2 : &body,
1019 2 : "application/json",
1020 2 : Some("<https://ctx>; rel=\"http://www.w3.org/ns/json-ld#context\""),
1021 2 : &[("MyKey".into(), "MyValue".into())],
1022 : );
1023 2 : assert_eq!(m["body"], body);
1024 2 : assert_eq!(m["metadata"]["Content-Type"], "application/json");
1025 2 : assert!(m["metadata"]["Link"]
1026 2 : .as_str()
1027 2 : .expect("link present")
1028 2 : .contains("json-ld#context"));
1029 2 : assert_eq!(m["metadata"]["MyKey"], "MyValue");
1030 :
1031 : // ld+json: @context is in the body, no Link in metadata (Table 7.2-2)
1032 2 : let m = build_message(&body, "application/ld+json", Some("<x>"), &[]);
1033 2 : assert_eq!(m["metadata"]["Content-Type"], "application/ld+json");
1034 2 : assert!(m["metadata"].get("Link").is_none());
1035 2 : }
1036 :
1037 : /// Table 7.2-2 names the source of two metadata keys: `Content-Type`
1038 : /// comes from `endpoint.accept` and `Link` from the served @context.
1039 : /// receiverInfo entries are added "additionally", so one keyed like a
1040 : /// table parameter cannot become that parameter's value — a subscriber
1041 : /// would otherwise decide the MIME type of its own notifications
1042 : /// against 6.3.8's "changed by means of the endpoint.accept member",
1043 : /// and could strip the @context reference the clause requires.
1044 : #[test]
1045 2 : fn receiver_info_cannot_take_over_a_table_7_2_2_parameter() {
1046 2 : let body = json!({"id": "urn:n:1", "type": "Notification"});
1047 2 : let link = "<https://ctx>; rel=\"http://www.w3.org/ns/json-ld#context\"";
1048 2 : let m = build_message(
1049 2 : &body,
1050 2 : "application/json",
1051 2 : Some(link),
1052 2 : &[
1053 2 : ("Content-Type".into(), "text/plain".into()),
1054 2 : ("Link".into(), "<https://evil>".into()),
1055 2 : ("NGSILD-Tenant".into(), "acme".into()),
1056 2 : ],
1057 : );
1058 2 : assert_eq!(m["metadata"]["Content-Type"], "application/json");
1059 2 : assert_eq!(m["metadata"]["Link"], link);
1060 : // every other pair is still added, this one by the caller (6.3.22)
1061 2 : assert_eq!(m["metadata"]["NGSILD-Tenant"], "acme");
1062 2 : }
1063 : }
|
