Check max qos violations in server dispatcher

CHANGES.md

@@ -6,6 +6,8 @@
 
 * v3: Allow to specify max allowed qos for server publishes
 
+* v5: Check max qos violations in server dispatcher
+
 ## [0.8.10] - 2022-09-25
 
 * Add .into_inner() client's helper for publish control message

src/v3/dispatcher.rs

@@ -174,9 +174,10 @@ where
                     }
                 }
 
+                // check max allowed qos
                 if publish.qos > self.max_qos {
                     log::trace!(
-                        "Max allowed QoS is viaolated, max {:?} provided {:?}",
+                        "Max allowed QoS is violated, max {:?} provided {:?}",
                         self.max_qos,
                         publish.qos
                     );

src/v3/server.rs

@@ -117,10 +117,10 @@ where
         self
     }
 
-    /// Set max QoS allowed.
+    /// Set max allowed QoS.
     ///
     /// If peer sends publish with higher qos then ProtocolError::MaxQoSViolated(..)
-    /// By default max qos is not set.
+    /// By default max qos is set to `ExactlyOnce`.
     pub fn max_qos(mut self, qos: QoS) -> Self {
         self.max_qos = qos;
         self

src/v5/dispatcher.rs

@@ -14,12 +14,13 @@ use super::control::{ControlMessage, ControlResult};
 use super::publish::{Publish, PublishAck};
 use super::shared::{Ack, MqttShared};
 use super::sink::MqttSink;
-use super::{codec, codec::EncodeLtd, Session};
+use super::{codec, codec::EncodeLtd, QoS, Session};
 
 /// mqtt3 protocol dispatcher
 pub(super) fn factory<St, T, C, E>(
     publish: T,
     control: C,
+    max_qos: QoS,
     max_inflight_size: usize,
 ) -> impl ServiceFactory<
     DispatchItem<Rc<MqttShared>>,
@@ -60,6 +61,7 @@ where
                 max_inflight_size,
                 Dispatcher::<_, _, E>::new(
                     cfg.sink().clone(),
+                    max_qos,
                     max_receive as usize,
                     max_topic_alias,
                     publish,
@@ -85,6 +87,7 @@ pub(crate) struct Dispatcher<T, C: Service<ControlMessage<E>>, E> {
     sink: MqttSink,
     publish: T,
     shutdown: RefCell<Option<Pin<Box<C::Future>>>>,
+    max_qos: QoS,
     max_receive: usize,
     max_topic_alias: u16,
     inner: Rc<Inner<C>>,
@@ -111,13 +114,15 @@ where
 {
     fn new(
         sink: MqttSink,
+        max_qos: QoS,
         max_receive: usize,
         max_topic_alias: u16,
         publish: T,
         control: C,
     ) -> Self {
         Self {
             publish,
+            max_qos,
             max_receive,
             max_topic_alias,
             sink: sink.clone(),
@@ -205,6 +210,21 @@ where
                             )));
                         }
 
+                        // check max allowed qos
+                        if publish.qos > self.max_qos {
+                            log::trace!(
+                                "Max allowed QoS is violated, max {:?} provided {:?}",
+                                self.max_qos,
+                                publish.qos
+                            );
+                            return Either::Right(Either::Right(ControlResponse::new(
+                                ControlMessage::proto_error(ProtocolError::MaxQoSViolated(
+                                    publish.qos,
+                                )),
+                                &self.inner,
+                            )));
+                        }
+
                         // check for duplicated packet id
                         if !inner.inflight.insert(pid) {
                             self.sink.send(codec::Packet::PublishAck(codec::PublishAck {

src/v5/server.rs

@@ -120,7 +120,7 @@ where
 
     /// Set server max qos setting.
     ///
-    /// By default max qos is not set`
+    /// By default max qos is not set.
     pub fn max_qos(mut self, qos: QoS) -> Self {
         self.max_qos = Some(qos);
         self
@@ -233,7 +233,12 @@ where
                 pool: self.pool,
                 _t: PhantomData,
             },
-            factory(self.srv_publish, self.srv_control, self.max_inflight_size),
+            factory(
+                self.srv_publish,
+                self.srv_control,
+                self.max_qos.unwrap_or(QoS::ExactlyOnce),
+                self.max_inflight_size,
+            ),
             self.disconnect_timeout,
         )
     }
@@ -258,6 +263,7 @@ where
             handler: Rc::new(factory(
                 self.srv_publish,
                 self.srv_control,
+                self.max_qos.unwrap_or(QoS::ExactlyOnce),
                 self.max_inflight_size,
             )),
             max_size: self.max_size,

tests/test_server_v5.rs

@@ -6,7 +6,7 @@ use ntex::{server, service::fn_service, time::sleep};
 
 use ntex_mqtt::v5::{
     client, codec, error, ControlMessage, Handshake, HandshakeAck, MqttServer, Publish,
-    PublishAck, Session,
+    PublishAck, QoS, Session,
 };
 
 struct St;
@@ -833,3 +833,54 @@ async fn test_handle_incoming() -> std::io::Result<()> {
 
     Ok(())
 }
+
+#[ntex::test]
+async fn test_max_qos() -> std::io::Result<()> {
+    let violated = Arc::new(AtomicBool::new(false));
+    let violated2 = violated.clone();
+
+    let srv = server::test_server(move || {
+        let violated = violated2.clone();
+        MqttServer::new(handshake)
+            .max_qos(QoS::AtMostOnce)
+            .publish(|p: Publish| Ready::Ok::<_, TestError>(p.ack()))
+            .control(move |msg| {
+                let violated = violated.clone();
+                match msg {
+                    ControlMessage::ProtocolError(msg) => {
+                        match msg.get_ref() {
+                            error::ProtocolError::MaxQoSViolated(_) => {
+                                violated.store(true, Relaxed);
+                            }
+                            _ => (),
+                        }
+                        Ready::Ok::<_, TestError>(msg.ack())
+                    }
+                    _ => Ready::Ok(msg.disconnect()),
+                }
+            })
+            .finish()
+    });
+
+    let io = srv.connect().await.unwrap();
+    let codec = codec::Codec::default();
+    io.encode(
+        codec::Packet::Connect(Box::new(codec::Connect::default().client_id("user"))),
+        &codec,
+    )
+    .unwrap();
+    let _ = io.recv(&codec).await.unwrap().unwrap();
+
+    io.encode(pkt_publish().into(), &codec).unwrap();
+    let pkt = io.recv(&codec).await.unwrap().unwrap();
+    assert_eq!(
+        pkt,
+        codec::Packet::Disconnect(codec::Disconnect {
+            reason_code: codec::DisconnectReasonCode::ImplementationSpecificError,
+            ..Default::default()
+        })
+    );
+    assert!(violated.load(Relaxed));
+
+    Ok(())
+}