Denial of Service (DoS)

@@ -273,6 +273,18 @@ impl ClientBuilder {
         self.config.session_name = session_name.into();
         self
     }
+
+    /// Set the maximum message size
+    pub fn max_message_size(mut self, max_message_size: usize) -> Self {
+        self.config.decoding_options.max_message_size = max_message_size;
+        self
+    }
+
+    /// Set the max chunk count
+    pub fn max_chunk_count(mut self, max_chunk_count: usize) -> Self {
+        self.config.decoding_options.max_chunk_count = max_chunk_count;
+        self
+    }
 }
 
 #[test]

@@ -427,10 +427,12 @@ impl Client {
         let decoding_options = &self.config.decoding_options;
         DecodingOptions {
             max_chunk_count: decoding_options.max_chunk_count,
+            max_message_size: decoding_options.max_message_size,
             max_string_length: decoding_options.max_string_length,
             max_byte_string_length: decoding_options.max_byte_string_length,
             max_array_length: decoding_options.max_array_length,
             client_offset: Duration::zero(),
+            ..Default::default()
         }
     }
 

@@ -132,6 +132,7 @@ impl ReadState {
                 let chunks_len = self.chunks.len();
                 if self.max_chunk_count > 0 && chunks_len > self.max_chunk_count {
                     error!("too many chunks {}> {}", chunks_len, self.max_chunk_count);
+                    // TODO this code should return an error to be safe
                     //remove first
                     let first_req_id = *self.chunks.iter().next().unwrap().0;
                     self.chunks.remove(&first_req_id);
@@ -312,7 +313,7 @@ impl TcpTransport {
     pub fn connect(&self, endpoint_url: &str) -> Result<(), StatusCode> {
         debug_assert!(!self.is_connected(), "Should not try to connect when already connected");
         let (host, port) =
-            hostname_port_from_url(endpoint_url, constants::DEFAULT_OPC_UA_SERVER_PORT)?;
+            hostname_port_from_url(endpoint_url, crate::core::constants::DEFAULT_OPC_UA_SERVER_PORT)?;
 
         // Resolve the host name into a socket address
         let addr = {

@@ -136,6 +136,8 @@ impl ClientEndpoint {
 #[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
 pub struct DecodingOptions {
     /// Maximum size of a message chunk in bytes. 0 means no limit
+    pub max_message_size: usize,
+    /// Maximum number of chunks in a message. 0 means no limit
     pub max_chunk_count: usize,
     /// Maximum length in bytes (not chars!) of a string. 0 actually means 0, i.e. no string permitted
     pub max_string_length: usize,
@@ -311,7 +313,6 @@ impl ClientConfig {
         pki_dir.push(Self::PKI_DIR);
 
         let decoding_options = crate::types::DecodingOptions::default();
-
         ClientConfig {
             application_name: application_name.into(),
             application_uri: application_uri.into(),
@@ -334,6 +335,7 @@ impl ClientConfig {
                 max_string_length: decoding_options.max_string_length,
                 max_byte_string_length: decoding_options.max_byte_string_length,
                 max_chunk_count: decoding_options.max_chunk_count,
+                max_message_size: decoding_options.max_message_size,
             },
             performance: Performance {
                 ignore_clock_skew: false,

@@ -157,7 +157,6 @@ impl SessionState {
     const SEND_BUFFER_SIZE: usize = 65535;
     const RECEIVE_BUFFER_SIZE: usize = 65535;
     const MAX_BUFFER_SIZE: usize = 65535;
-    const MAX_CHUNK_COUNT: usize = 0;
 
     pub fn new(
         ignore_clock_skew: bool,
@@ -176,7 +175,7 @@ impl SessionState {
             send_buffer_size: Self::SEND_BUFFER_SIZE,
             receive_buffer_size: Self::RECEIVE_BUFFER_SIZE,
             max_message_size: Self::MAX_BUFFER_SIZE,
-            max_chunk_count: Self::MAX_CHUNK_COUNT,
+            max_chunk_count: constants::MAX_CHUNK_COUNT,
             request_handle: Handle::new(Self::FIRST_REQUEST_HANDLE),
             session_id: NodeId::null(),
             authentication_token: NodeId::null(),

@@ -158,7 +158,8 @@ impl BinaryEncoder<MessageChunk> for MessageChunk {
             })?;
 
         let message_size = chunk_header.message_size as usize;
-        if decoding_options.max_chunk_count > 0 && message_size > decoding_options.max_chunk_count {
+        if decoding_options.max_message_size > 0 && message_size > decoding_options.max_message_size
+        {
             // Message_size should be sanity checked and rejected if too large.
             Err(StatusCode::BadTcpMessageTooLarge)
         } else {

@@ -66,25 +66,37 @@ impl MessageWriter {
             &message,
         )?;
 
-        // Sequence number monotonically increases per chunk
-        self.last_sent_sequence_number += chunks.len() as u32;
+        if self.max_chunk_count > 0 && chunks.len() > self.max_chunk_count {
+            error!(
+                "Cannot write message since {} chunks exceeds {} chunk limit",
+                chunks.len(),
+                self.max_chunk_count
+            );
+            Err(StatusCode::BadCommunicationError)
+        } else {
+            // Sequence number monotonically increases per chunk
+            self.last_sent_sequence_number += chunks.len() as u32;
 
-        // Send chunks
+            // Send chunks
 
-        // This max chunk size allows the message to be encoded to a chunk with header + encoding
-        // which is just slightly larger in size (up to 1024 bytes).
-        let max_chunk_count = self.buffer.get_ref().len() + 1024;
-        let mut data = vec![0u8; max_chunk_count];
-        for chunk in chunks {
-            trace!("Sending chunk {:?}", chunk);
-            let size = secure_channel.apply_security(&chunk, &mut data)?;
-            self.buffer.write(&data[..size]).map_err(|error| {
-                error!("Error while writing bytes to stream, connection broken, check error {:?}", error);
-                StatusCode::BadCommunicationError
-            })?;
+            // This max chunk size allows the message to be encoded to a chunk with header + encoding
+            // which is just slightly larger in size (up to 1024 bytes).
+            let data_buffer_size = self.buffer.get_ref().len() + 1024;
+            let mut data = vec![0u8; data_buffer_size];
+            for chunk in chunks {
+                trace!("Sending chunk {:?}", chunk);
+                let size = secure_channel.apply_security(&chunk, &mut data)?;
+                self.buffer.write(&data[..size]).map_err(|error| {
+                    error!(
+                        "Error while writing bytes to stream, connection broken, check error {:?}",
+                        error
+                    );
+                    StatusCode::BadCommunicationError
+                })?;
+            }
+            trace!("Message written");
+            Ok(request_id)
         }
-        trace!("Message written");
-        Ok(request_id)
     }
 
     pub fn next_request_id(&mut self) -> u32 {

@@ -70,12 +70,14 @@ pub struct SecureChannel {
     decoding_options: DecodingOptions,
 }
 
-impl From<(SecurityPolicy, MessageSecurityMode)> for SecureChannel {
-    fn from(v: (SecurityPolicy, MessageSecurityMode)) -> Self {
+impl SecureChannel {
+    /// For testing purposes only
+    #[cfg(test)]
+    pub fn new_no_certificate_store() -> SecureChannel {
         SecureChannel {
             role: Role::Unknown,
-            security_policy: v.0,
-            security_mode: v.1,
+            security_policy: SecurityPolicy::None,
+            security_mode: MessageSecurityMode::None,
             secure_channel_id: 0,
             token_id: 0,
             token_created_at: DateTime::now(),
@@ -90,14 +92,6 @@ impl From<(SecurityPolicy, MessageSecurityMode)> for SecureChannel {
             decoding_options: DecodingOptions::default(),
         }
     }
-}
-
-impl SecureChannel {
-    /// For testing purposes only
-    #[cfg(test)]
-    pub fn new_no_certificate_store() -> SecureChannel {
-        (SecurityPolicy::None, MessageSecurityMode::None).into()
-    }
 
     pub fn new(
         certificate_store: Arc<RwLock<CertificateStore>>,
@@ -222,7 +216,7 @@ impl SecureChannel {
     }
 
     pub fn decoding_options(&self) -> DecodingOptions {
-        self.decoding_options
+        self.decoding_options.clone()
     }
 
     /// Test if the secure channel token needs to be renewed. The algorithm determines it needs

@@ -88,8 +88,7 @@ impl BinaryEncoder<AsymmetricSecurityHeader> for AsymmetricSecurityHeader {
 
         // validate sender_certificate_length < MaxCertificateSize
         if sender_certificate.value.is_some()
-            && sender_certificate.value.as_ref().unwrap().len()
-                >= constants::MAX_CERTIFICATE_LENGTH as usize
+            && sender_certificate.value.as_ref().unwrap().len() >= constants::MAX_CERTIFICATE_LENGTH
         {
             error!("Sender certificate exceeds max certificate size");
             Err(StatusCode::BadDecodingError)

@@ -6,7 +6,7 @@
 
 use url::Url;
 
-use crate::types::{constants::DEFAULT_OPC_UA_SERVER_PORT, status_code::StatusCode};
+use crate::types::status_code::StatusCode;
 
 pub const OPC_TCP_SCHEME: &str = "opc.tcp";
 
@@ -16,7 +16,7 @@ fn opc_url_from_str(s: &str) -> Result<Url, ()> {
         .map(|mut url| {
             if url.port().is_none() {
                 // If no port is supplied, then treat it as the default port 4840
-                let _ = url.set_port(Some(DEFAULT_OPC_UA_SERVER_PORT));
+                let _ = url.set_port(Some(crate::core::constants::DEFAULT_OPC_UA_SERVER_PORT));
             }
             url
         })
@@ -57,7 +57,7 @@ pub fn server_url_from_endpoint_url(endpoint_url: &str) -> std::result::Result<S
         url.set_query(None);
         if let Some(port) = url.port() {
             // If the port is the default, strip it so the url string omits it.
-            if port == DEFAULT_OPC_UA_SERVER_PORT {
+            if port == crate::core::constants::DEFAULT_OPC_UA_SERVER_PORT {
                 let _ = url.set_port(None);
             }
         }

@@ -93,6 +93,13 @@ pub mod debug {
 #[cfg(test)]
 pub mod tests;
 
+pub mod constants {
+    /// Default OPC UA port number. Used by a discovery server. Other servers would normally run
+    /// on a different port. So OPC UA for Rust does not use this nr by default but it is used
+    /// implicitly in opc.tcp:// urls and elsewhere.
+    pub const DEFAULT_OPC_UA_SERVER_PORT: u16 = 4840;
+}
+
 pub mod comms;
 pub mod config;
 pub mod handle;

@@ -36,7 +36,7 @@ fn sample_secure_channel_request_data_security_none() -> MessageChunk {
 
     // Decode chunk from stream
     stream.set_position(0);
-    let decoding_options = DecodingOptions::default();
+    let decoding_options = DecodingOptions::test();
     let chunk = MessageChunk::decode(&mut stream, &decoding_options).unwrap();
 
     println!(
@@ -102,6 +102,7 @@ fn chunk_multi_encode_decode() {
         max_string_length: 65535,
         max_byte_string_length: 65535,
         max_array_length: 20000, // Need to bump this up because large response uses a large array
+        ..Default::default()
     });
 
     let response = make_large_read_response();
@@ -152,7 +153,7 @@ fn chunk_multi_chunk_intermediate_final() {
     .unwrap();
     assert!(chunks.len() > 1);
 
-    let decoding_options = DecodingOptions::default();
+    let decoding_options = DecodingOptions::test();
 
     // All chunks except the last should be intermediate, the last should be final
     for (i, chunk) in chunks.iter().enumerate() {

@@ -24,7 +24,7 @@ fn ack_data() -> Vec<u8> {
 #[test]
 pub fn hello() {
     let mut stream = Cursor::new(hello_data());
-    let decoding_options = DecodingOptions::default();
+    let decoding_options = DecodingOptions::test();
     let hello = HelloMessage::decode(&mut stream, &decoding_options).unwrap();
     println!("hello = {:?}", hello);
     assert_eq!(hello.message_header.message_type, MessageType::Hello);
@@ -43,7 +43,7 @@ pub fn hello() {
 #[test]
 pub fn acknowledge() {
     let mut stream = Cursor::new(ack_data());
-    let decoding_options = DecodingOptions::default();
+    let decoding_options = DecodingOptions::test();
     let ack = AcknowledgeMessage::decode(&mut stream, &decoding_options).unwrap();
     println!("ack = {:?}", ack);
     assert_eq!(ack.message_header.message_type, MessageType::Acknowledge);

@@ -36,7 +36,7 @@ where
     println!("encoded bytes = {:?}", actual);
     let mut stream = Cursor::new(actual);
 
-    let decoding_options = DecodingOptions::default();
+    let decoding_options = DecodingOptions::test();
     let new_value: T = T::decode(&mut stream, &decoding_options).unwrap();
     println!("new value = {:?}", new_value);
     assert_eq!(value, new_value);

@@ -308,36 +308,60 @@ impl ServerBuilder {
     }
 
     /// Set the maximum number of subscriptions in a session
-    pub fn max_subscriptions(mut self, max_subscriptions: u32) -> Self {
+    pub fn max_subscriptions(mut self, max_subscriptions: usize) -> Self {
         self.config.limits.max_subscriptions = max_subscriptions;
         self
     }
 
     /// Set the maximum number of monitored items per subscription
-    pub fn max_monitored_items_per_sub(mut self, max_monitored_items_per_sub: u32) -> Self {
+    pub fn max_monitored_items_per_sub(mut self, max_monitored_items_per_sub: usize) -> Self {
         self.config.limits.max_monitored_items_per_sub = max_monitored_items_per_sub;
         self
     }
 
     /// Set the max array length in elements
-    pub fn max_array_length(mut self, max_array_length: u32) -> Self {
+    pub fn max_array_length(mut self, max_array_length: usize) -> Self {
         self.config.limits.max_array_length = max_array_length;
         self
     }
 
     /// Set the max string length in characters, i.e. if you set max to 1000 characters, then with
     /// UTF-8 encoding potentially that's 4000 bytes.
-    pub fn max_string_length(mut self, max_string_length: u32) -> Self {
+    pub fn max_string_length(mut self, max_string_length: usize) -> Self {
         self.config.limits.max_string_length = max_string_length;
         self
     }
 
     /// Set the max bytestring length in bytes
-    pub fn max_byte_string_length(mut self, max_byte_string_length: u32) -> Self {
+    pub fn max_byte_string_length(mut self, max_byte_string_length: usize) -> Self {
         self.config.limits.max_byte_string_length = max_byte_string_length;
         self
     }
 
+    /// Set the maximum message size
+    pub fn max_message_size(mut self, max_message_size: usize) -> Self {
+        self.config.limits.max_message_size = max_message_size;
+        self
+    }
+
+    /// Set the max chunk count
+    pub fn max_chunk_count(mut self, max_chunk_count: usize) -> Self {
+        self.config.limits.max_chunk_count = max_chunk_count;
+        self
+    }
+
+    // Set the send buffer size
+    pub fn send_buffer_size(mut self, send_buffer_size: usize) -> Self {
+        self.config.limits.send_buffer_size = send_buffer_size;
+        self
+    }
+
+    // Set the receive buffer size
+    pub fn receive_buffer_size(mut self, receive_buffer_size: usize) -> Self {
+        self.config.limits.receive_buffer_size = receive_buffer_size;
+        self
+    }
+
     /// Sets the server to automatically trust client certs. This subverts the
     /// authentication during handshake, so only do this if you understand the risks.
     pub fn trust_client_certs(mut self) -> Self {

@@ -9,9 +9,9 @@
 //! responses. i.e. the client is expected to call and wait for a response to their request.
 //! Publish requests are sent based on the number of subscriptions and the responses / handling are
 //! left to asynchronous event handlers.
-use std::{net::SocketAddr, sync::Arc};
 use chrono::{self, Utc};
 use futures::StreamExt;
+use std::{net::SocketAddr, sync::Arc};
 use tokio::{
     self,
     io::AsyncWriteExt,
@@ -44,15 +44,9 @@ use crate::server::{
     services::message_handler::MessageHandler,
     session::SessionManager,
     state::ServerState,
-    subscriptions::{subscription::TickReason},
+    subscriptions::subscription::TickReason,
 };
 
-// TODO these need to go, and use session settings
-const RECEIVE_BUFFER_SIZE: usize = std::u16::MAX as usize;
-const SEND_BUFFER_SIZE: usize = std::u16::MAX as usize;
-const MAX_MESSAGE_SIZE: usize = std::u16::MAX as usize;
-const MAX_CHUNK_COUNT: usize = 1;
-
 /// Messages that may be sent to the writer.
 #[derive(Debug)]
 enum Message {
@@ -224,14 +218,25 @@ impl TcpTransport {
         );
 
         // Store the address of the client
-        {
+        let (send_buffer_size, receive_buffer_size) = {
             let mut connection = trace_write_lock!(connection);
             connection.client_address = Some(socket.peer_addr().unwrap());
             connection.transport_state = TransportState::WaitingHello;
-        }
+            let server_state = trace_read_lock!(connection.server_state);
+            (
+                server_state.send_buffer_size,
+                server_state.receive_buffer_size,
+            )
+        };
 
         // Spawn the tasks we need to run
-        tokio::spawn(Self::spawn_session_handler_task(connection, socket, looping_interval_ms));
+        tokio::spawn(Self::spawn_session_handler_task(
+            connection,
+            socket,
+            looping_interval_ms,
+            send_buffer_size,
+            receive_buffer_size,
+        ));
     }
 
     async fn write_bytes_task(mut write_state: WriteState) -> WriteState {
@@ -252,10 +257,9 @@ impl TcpTransport {
         transport: Arc<RwLock<TcpTransport>>,
         socket: TcpStream,
         looping_interval_ms: f64,
+        send_buffer_size: usize,
+        receive_buffer_size: usize,
     ) {
-        // These should really come from the session
-        let send_buffer_size = SEND_BUFFER_SIZE;
-
         // The reader task will send responses, the writer task will receive responses
         let (tx, rx) = unbounded_channel();
         let send_buffer = Arc::new(Mutex::new(MessageWriter::new(send_buffer_size, 0, 0)));
@@ -265,8 +269,15 @@ impl TcpTransport {
             let transport = trace_read_lock!(transport);
             let server_state = trace_read_lock!(transport.server_state);
             let server_config = trace_read_lock!(server_state.config);
-            info!("Session transport {} started at {}", transport.transport_id, Utc::now());
-            (server_config.tcp_config.hello_timeout, transport.secure_channel.clone())
+            info!(
+                "Session transport {} started at {}",
+                transport.transport_id,
+                Utc::now()
+            );
+            (
+                server_config.tcp_config.hello_timeout,
+                transport.secure_channel.clone(),
+            )
         };
 
         let read_state = ReadState {
@@ -287,7 +298,7 @@ impl TcpTransport {
                 log::trace!("Closing connection after the write task ended");
                 status
             }
-            status = Self::spawn_reading_loop_task(read_state) => {
+            status = Self::spawn_reading_loop_task(read_state, send_buffer_size, receive_buffer_size) => {
                 log::trace!("Closing connection after the read task ended");
                 status
             }
@@ -372,7 +383,10 @@ impl TcpTransport {
                 Err(StatusCode::BadCommunicationError)
             }
             Ok(Some(Err(communication_err))) => {
-                error!("Communication error while waiting for Hello message: {}", communication_err);
+                error!(
+                    "Communication error while waiting for Hello message: {}",
+                    communication_err
+                );
                 Err(StatusCode::BadCommunicationError)
             }
             Ok(None) => Err(StatusCode::BadConnectionClosed),
@@ -381,7 +395,11 @@ impl TcpTransport {
 
     /// Spawns the reading loop where a reader task continuously reads messages, chunks from the
     /// input and process them. The reading task will terminate upon error.
-    async fn spawn_reading_loop_task(read_state: ReadState) -> Result<(), StatusCode> {
+    async fn spawn_reading_loop_task(
+        read_state: ReadState,
+        send_buffer_size: usize,
+        receive_buffer_size: usize,
+    ) -> Result<(), StatusCode> {
         let (transport, mut sender) = { (read_state.transport.clone(), read_state.sender.clone()) };
 
         let decoding_options = {
@@ -392,10 +410,16 @@ impl TcpTransport {
 
         // The reader reads frames from the codec, which are messages
         let mut framed_read =
-            FramedRead::new(read_state.reader, TcpCodec::new(decoding_options));
+            FramedRead::new(read_state.reader, TcpCodec::new(decoding_options.clone()));
 
         let hello = Self::wait_for_hello(&mut framed_read, read_state.hello_timeout).await?;
-        trace_write_lock!(transport).process_hello(hello, &mut sender)?;
+        trace_write_lock!(transport).process_hello(
+            hello,
+            &mut sender,
+            &decoding_options,
+            send_buffer_size,
+            receive_buffer_size,
+        )?;
 
         while let Some(next_msg) = framed_read.next().await {
             match next_msg {
@@ -450,24 +474,31 @@ impl TcpTransport {
                 session.tick_subscriptions(&now, &address_space, TickReason::TickTimerFired)?;
 
                 // Check if there are publish responses to send for transmission
-                if let Some(publish_responses) = session.subscriptions_mut().take_publish_responses() {
+                if let Some(publish_responses) =
+                    session.subscriptions_mut().take_publish_responses()
+                {
                     for publish_response in publish_responses {
-                        trace!("<-- Sending a Publish Response{}, {:?}", publish_response.request_id, &publish_response.response);
-                        // Messages will be sent by the writing task
-                        sender.send(Message::Message(
+                        trace!(
+                            "<-- Sending a Publish Response{}, {:?}",
                             publish_response.request_id,
-                            publish_response.response,
-                        )).map_err(|e| {
-                            error!("Unable to send publish response to writer task: {}", e);
-                            StatusCode::BadUnexpectedError
-                        })?;
+                            &publish_response.response
+                        );
+                        // Messages will be sent by the writing task
+                        sender
+                            .send(Message::Message(
+                                publish_response.request_id,
+                                publish_response.response,
+                            ))
+                            .map_err(|e| {
+                                error!("Unable to send publish response to writer task: {}", e);
+                                StatusCode::BadUnexpectedError
+                            })?;
                     }
                 }
             }
         }
     }
 
-
     /// Test if the connection should abort
     pub fn is_server_abort(&self) -> bool {
         let server_state = trace_read_lock!(self.server_state);
@@ -478,13 +509,16 @@ impl TcpTransport {
         &mut self,
         hello: HelloMessage,
         sender: &mut UnboundedSender<Message>,
+        decoding_options: &DecodingOptions,
+        send_buffer_size: usize,
+        receive_buffer_size: usize,
     ) -> std::result::Result<(), StatusCode> {
         let server_protocol_version = 0;
         let endpoints = {
             let server_state = trace_read_lock!(self.server_state);
             server_state.endpoints(&hello.endpoint_url, &None)
         }
-            .unwrap();
+        .unwrap();
 
         trace!("Server received HELLO {:?}", hello);
         if !hello.is_endpoint_url_valid(&endpoints) {
@@ -507,10 +541,10 @@ impl TcpTransport {
         let mut acknowledge = AcknowledgeMessage {
             message_header: MessageHeader::new(MessageType::Acknowledge),
             protocol_version: server_protocol_version,
-            receive_buffer_size: RECEIVE_BUFFER_SIZE as u32,
-            send_buffer_size: SEND_BUFFER_SIZE as u32,
-            max_message_size: MAX_MESSAGE_SIZE as u32,
-            max_chunk_count: MAX_CHUNK_COUNT as u32,
+            receive_buffer_size: receive_buffer_size as u32,
+            send_buffer_size: send_buffer_size as u32,
+            max_message_size: decoding_options.max_message_size as u32,
+            max_chunk_count: decoding_options.max_chunk_count as u32,
         };
         acknowledge.message_header.message_size = acknowledge.byte_len() as u32;
         let acknowledge: SupportedMessage = acknowledge.into();
@@ -651,4 +685,4 @@ impl TcpTransport {
             .handle_message(request_id, request, sender)?;
         Ok(())
     }
-}
\ No newline at end of file
+}

@@ -10,16 +10,16 @@ use std::str::FromStr;
 use crate::{
     core::{comms::url::url_matches_except_host, config::Config},
     crypto::{CertificateStore, SecurityPolicy, Thumbprint},
-    types::{
-        constants as opcua_types_constants, service_types::ApplicationType, DecodingOptions,
-        MessageSecurityMode, UAString,
-    },
+    types::{service_types::ApplicationType, DecodingOptions, MessageSecurityMode, UAString},
 };
 
 use super::constants;
 
 pub const ANONYMOUS_USER_TOKEN_ID: &str = "ANONYMOUS";
 
+const RECEIVE_BUFFER_SIZE: usize = std::u16::MAX as usize;
+const SEND_BUFFER_SIZE: usize = std::u16::MAX as usize;
+
 #[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
 pub struct TcpConfig {
     /// Timeout for hello on a session in seconds
@@ -132,35 +132,48 @@ pub struct Limits {
     /// in a later revision. By default, this value is `false`
     pub clients_can_modify_address_space: bool,
     /// Maximum number of subscriptions in a session, 0 for no limit
-    pub max_subscriptions: u32,
+    pub max_subscriptions: usize,
     /// Maximum number of monitored items per subscription, 0 for no limit
-    pub max_monitored_items_per_sub: u32,
+    pub max_monitored_items_per_sub: usize,
     /// Maximum number of values in a monitored item queue
-    pub max_monitored_item_queue_size: u32,
+    pub max_monitored_item_queue_size: usize,
     /// Max array length in elements
-    pub max_array_length: u32,
+    pub max_array_length: usize,
     /// Max string length in characters
-    pub max_string_length: u32,
+    pub max_string_length: usize,
     /// Max bytestring length in bytes
-    pub max_byte_string_length: u32,
+    pub max_byte_string_length: usize,
     /// Specifies the minimum sampling interval for this server in seconds.
     pub min_sampling_interval: f64,
     /// Specifies the minimum publishing interval for this server in seconds.
     pub min_publishing_interval: f64,
+    /// Maximum message length in bytes
+    pub max_message_size: usize,
+    /// Maximum chunk count
+    pub max_chunk_count: usize,
+    /// Send buffer size in bytes
+    pub send_buffer_size: usize,
+    /// Receive buffer size in bytes
+    pub receive_buffer_size: usize,
 }
 
 impl Default for Limits {
     fn default() -> Self {
+        let decoding_options = DecodingOptions::default();
         Self {
-            max_array_length: opcua_types_constants::MAX_ARRAY_LENGTH as u32,
-            max_string_length: opcua_types_constants::MAX_STRING_LENGTH as u32,
-            max_byte_string_length: opcua_types_constants::MAX_BYTE_STRING_LENGTH as u32,
+            max_array_length: decoding_options.max_array_length,
+            max_string_length: decoding_options.max_string_length,
+            max_byte_string_length: decoding_options.max_byte_string_length,
             max_subscriptions: constants::DEFAULT_MAX_SUBSCRIPTIONS,
             max_monitored_items_per_sub: constants::DEFAULT_MAX_MONITORED_ITEMS_PER_SUB,
-            max_monitored_item_queue_size: constants::MAX_DATA_CHANGE_QUEUE_SIZE as u32,
+            max_monitored_item_queue_size: constants::MAX_DATA_CHANGE_QUEUE_SIZE,
+            max_message_size: decoding_options.max_message_size,
+            max_chunk_count: decoding_options.max_chunk_count,
             clients_can_modify_address_space: false,
             min_sampling_interval: constants::MIN_SAMPLING_INTERVAL,
             min_publishing_interval: constants::MIN_PUBLISHING_INTERVAL,
+            send_buffer_size: SEND_BUFFER_SIZE,
+            receive_buffer_size: RECEIVE_BUFFER_SIZE,
         }
     }
 }
@@ -708,10 +721,12 @@ impl ServerConfig {
     pub fn decoding_options(&self) -> DecodingOptions {
         DecodingOptions {
             client_offset: chrono::Duration::zero(),
-            max_chunk_count: 0,
-            max_string_length: self.limits.max_string_length as usize,
-            max_byte_string_length: self.limits.max_byte_string_length as usize,
-            max_array_length: self.limits.max_array_length as usize,
+            max_message_size: self.limits.max_message_size,
+            max_chunk_count: self.limits.max_chunk_count,
+            max_string_length: self.limits.max_string_length,
+            max_byte_string_length: self.limits.max_byte_string_length,
+            max_array_length: self.limits.max_array_length,
+            ..Default::default()
         }
     }
 

@@ -93,9 +93,9 @@ pub mod constants {
     /// Default OPC UA server port for this implementation
     pub const DEFAULT_RUST_OPC_UA_SERVER_PORT: u16 = 4855;
     /// Default maximum number of subscriptions in a session
-    pub const DEFAULT_MAX_SUBSCRIPTIONS: u32 = 100;
+    pub const DEFAULT_MAX_SUBSCRIPTIONS: usize = 100;
     /// Default maximum number of monitored items per subscription
-    pub const DEFAULT_MAX_MONITORED_ITEMS_PER_SUB: u32 = 1000;
+    pub const DEFAULT_MAX_MONITORED_ITEMS_PER_SUB: usize = 1000;
     /// Default, well known address for TCP discovery server
     pub const DEFAULT_DISCOVERY_SERVER_URL: &str = "opc.tcp://localhost:4840/UADiscovery";
 

@@ -112,6 +112,8 @@ impl Server {
         let diagnostics = Arc::new(RwLock::new(ServerDiagnostics::default()));
         let min_publishing_interval_ms = config.limits.min_publishing_interval * 1000.0;
         let min_sampling_interval_ms = config.limits.min_sampling_interval * 1000.0;
+        let send_buffer_size = config.limits.send_buffer_size;
+        let receive_buffer_size = config.limits.receive_buffer_size;
 
         // TODO max string, byte string and array lengths
 
@@ -182,6 +184,8 @@ impl Server {
             historical_data_provider: None,
             historical_event_provider: None,
             operational_limits: OperationalLimits::default(),
+            send_buffer_size,
+            receive_buffer_size,
         };
         let server_state = Arc::new(RwLock::new(server_state));
 

@@ -488,6 +488,7 @@ impl MessageHandler {
 
     /// Test if the session is activated
     fn is_session_activated(
+        &self,
         session: Arc<RwLock<Session>>,
         request_header: &RequestHeader,
     ) -> Result<(), SupportedMessage> {
@@ -496,7 +497,19 @@ impl MessageHandler {
             error!("Session is not activated so request fails");
             Err(ServiceFault::new(request_header, StatusCode::BadSessionNotActivated).into())
         } else {
-            Ok(())
+            // Ensure the session's secure channel
+            let secure_channel_id = {
+                let secure_channel = trace_read_lock!(self.secure_channel);
+                secure_channel.secure_channel_id()
+            };
+            if secure_channel_id != session.secure_channel_id() {
+                error!(
+                    "service call rejected as secure channel id does not match that on the session"
+                );
+                Err(ServiceFault::new(request_header, StatusCode::BadSessionIdInvalid).into())
+            } else {
+                Ok(())
+            }
         }
     }
 
@@ -559,20 +572,19 @@ impl MessageHandler {
             session_manager.find_session_by_token(&request_header.authentication_token)
         };
         if let Some(session) = session {
-            let (response, authorized) = if let Err(response) =
-                Self::is_session_activated(session.clone(), request_header)
-            {
-                (Some(response), false)
-            } else if let Err(response) =
-                Self::is_session_timed_out(session.clone(), request_header, now)
-            {
-                (Some(response), false)
-            } else {
-                let response = action(session.clone(), session_manager);
-                let mut session = trace_write_lock!(session);
-                session.set_last_service_request_timestamp(now);
-                (response, true)
-            };
+            let (response, authorized) =
+                if let Err(response) = self.is_session_activated(session.clone(), request_header) {
+                    (Some(response), false)
+                } else if let Err(response) =
+                    Self::is_session_timed_out(session.clone(), request_header, now)
+                {
+                    (Some(response), false)
+                } else {
+                    let response = action(session.clone(), session_manager);
+                    let mut session = trace_write_lock!(session);
+                    session.set_last_service_request_timestamp(now);
+                    (response, true)
+                };
             // Async calls may not return a response here
             response.map(|response| {
                 Self::diag_service_response(session, authorized, &response, diagnostic_key);

@@ -112,7 +112,7 @@ pub struct ServerState {
     pub max_lifetime_count: u32,
     /// Operational limits
     pub(crate) operational_limits: OperationalLimits,
-    //// Current state
+    /// Current state
     pub state: ServerStateType,
     /// Sets the abort flag that terminates the associated server
     pub abort: bool,
@@ -128,6 +128,10 @@ pub struct ServerState {
     pub(crate) historical_data_provider: Option<Box<dyn HistoricalDataProvider + Send + Sync>>,
     /// Callback for historical events
     pub(crate) historical_event_provider: Option<Box<dyn HistoricalEventProvider + Send + Sync>>,
+    /// Size of the send buffer in bytes
+    pub send_buffer_size: usize,
+    /// Size of the receive buffer in bytes
+    pub receive_buffer_size: usize,
 }
 
 impl ServerState {

@@ -47,7 +47,10 @@ pub(crate) enum FilterType {
 }
 
 impl FilterType {
-    pub fn from_filter(filter: &ExtensionObject) -> Result<FilterType, StatusCode> {
+    pub fn from_filter(
+        filter: &ExtensionObject,
+        decoding_options: &DecodingOptions,
+    ) -> Result<FilterType, StatusCode> {
         // Check if the filter is a supported filter type
         let filter_type_id = &filter.node_id;
         if filter_type_id.is_null() {
@@ -56,17 +59,13 @@ impl FilterType {
         } else if let Ok(filter_type_id) = filter_type_id.as_object_id() {
             match filter_type_id {
                 ObjectId::DataChangeFilter_Encoding_DefaultBinary => {
-                    let decoding_options = DecodingOptions::minimal();
                     Ok(FilterType::DataChangeFilter(
-                        filter.decode_inner::<DataChangeFilter>(&decoding_options)?,
-                    ))
-                }
-                ObjectId::EventFilter_Encoding_DefaultBinary => {
-                    let decoding_options = DecodingOptions::default();
-                    Ok(FilterType::EventFilter(
-                        filter.decode_inner::<EventFilter>(&decoding_options)?,
+                        filter.decode_inner::<DataChangeFilter>(decoding_options)?,
                     ))
                 }
+                ObjectId::EventFilter_Encoding_DefaultBinary => Ok(FilterType::EventFilter(
+                    filter.decode_inner::<EventFilter>(decoding_options)?,
+                )),
                 _ => {
                     error!(
                         "Requested data filter type is not supported, {:?}",
@@ -125,7 +124,10 @@ impl MonitoredItem {
         server_state: &ServerState,
         request: &MonitoredItemCreateRequest,
     ) -> Result<MonitoredItem, StatusCode> {
-        let filter = FilterType::from_filter(&request.requested_parameters.filter)?;
+        let filter = FilterType::from_filter(
+            &request.requested_parameters.filter,
+            &server_state.decoding_options(),
+        )?;
         let sampling_interval = Self::sanitize_sampling_interval(
             server_state,
             request.requested_parameters.sampling_interval,
@@ -162,7 +164,10 @@ impl MonitoredItem {
         request: &MonitoredItemModifyRequest,
     ) -> Result<ExtensionObject, StatusCode> {
         self.timestamps_to_return = timestamps_to_return;
-        self.filter = FilterType::from_filter(&request.requested_parameters.filter)?;
+        self.filter = FilterType::from_filter(
+            &request.requested_parameters.filter,
+            &server_state.decoding_options(),
+        )?;
         self.sampling_interval = Self::sanitize_sampling_interval(
             server_state,
             request.requested_parameters.sampling_interval,

@@ -708,7 +708,7 @@ fn method_builder() {
             let v = v.get(0).unwrap().clone();
             if let Variant::ExtensionObject(v) = v {
                 // deserialize the Argument here
-                let decoding_options = DecodingOptions::default();
+                let decoding_options = DecodingOptions::test();
                 let argument = v.decode_inner::<Argument>(&decoding_options).unwrap();
                 assert_eq!(argument.name, UAString::from("Result"));
                 assert_eq!(argument.data_type, DataTypeId::String.into());

@@ -790,6 +790,8 @@ fn test_bitwise_and() {
 
 #[test]
 fn test_where_clause() {
+    crate::console_logging::init();
+
     let address_space = address_space();
 
     let object_id = NodeId::root_folder_id();

@@ -802,7 +802,7 @@ fn monitored_item_triggers() {
                 |response| {
                     let (notifications, events) = response
                         .notification_message
-                        .notifications(&DecodingOptions::default())
+                        .notifications(&DecodingOptions::test())
                         .unwrap();
                     assert_eq!(notifications.len(), 1);
                     assert!(events.is_empty());
@@ -875,7 +875,7 @@ fn monitored_item_triggers() {
                 |response| {
                     let (notifications, events) = response
                         .notification_message
-                        .notifications(&DecodingOptions::default())
+                        .notifications(&DecodingOptions::test())
                         .unwrap();
                     assert_eq!(notifications.len(), 1);
                     assert!(events.is_empty());
@@ -942,7 +942,7 @@ fn monitored_item_triggers() {
                     // expect only 1 data change corresponding to sampling triggered item
                     let (notifications, events) = response
                         .notification_message
-                        .notifications(&DecodingOptions::default())
+                        .notifications(&DecodingOptions::test())
                         .unwrap();
                     assert_eq!(notifications.len(), 1);
                     assert!(events.is_empty());

@@ -72,7 +72,7 @@ fn republish_request(subscription_id: u32, retransmit_sequence_number: u32) -> R
 
 #[test]
 fn create_modify_destroy_subscription() {
-    do_subscription_service_test(|server_state, session, _, ss, _| {
+    do_subscription_service_test(|server_state, _session, _, _ss, _| {
         // TODO Create a subscription, modify it, destroy it
         //unimplemented!();
     })
@@ -238,7 +238,7 @@ fn publish_response_subscription() {
 
         // We expect the notification to contain one data change notification referring to
         // the monitored item.
-        let decoding_options = DecodingOptions::default();
+        let decoding_options = DecodingOptions::test();
         let data_change = notification_data[0]
             .decode_inner::<DataChangeNotification>(&decoding_options)
             .unwrap();
@@ -355,8 +355,8 @@ fn publish_keep_alive() {
 
 #[test]
 fn multiple_publish_response_subscription() {
-    do_subscription_service_test(|server_state, session, address_space, ss, mis| {
-        let subscription_id = create_subscription(server_state, session.clone(), &ss);
+    do_subscription_service_test(|server_state, session, address_space, ss, _mis| {
+        let _subscription_id = create_subscription(server_state, session.clone(), &ss);
 
         let now = Utc::now();
         let request_id = 1001;
@@ -380,7 +380,7 @@ fn multiple_publish_response_subscription() {
 
 #[test]
 fn acknowledge_unknown_sequence_nr() {
-    do_subscription_service_test(|server_state, session, address_space, ss, mis| {
+    do_subscription_service_test(|server_state, session, address_space, ss, _mis| {
         let subscription_id = create_subscription(server_state, session.clone(), &ss);
 
         let now = Utc::now();

@@ -129,7 +129,7 @@ impl BinaryEncoder<DataValue> for DataValue {
             // The source timestamp should never be adjusted, not even when ignoring clock skew
             let decoding_options = DecodingOptions {
                 client_offset: chrono::Duration::zero(),
-                ..*decoding_options
+                ..decoding_options.clone()
             };
             Some(DateTime::decode(stream, &decoding_options)?)
         } else {

@@ -8,54 +8,144 @@
 use std::{
     fmt::Debug,
     io::{Cursor, Read, Result, Write},
+    sync::Arc,
 };
 
 use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
 use chrono::Duration;
 
-use crate::types::{constants, status_codes::StatusCode};
+use crate::{
+    sync::Mutex,
+    types::{constants, status_codes::StatusCode},
+};
 
 pub type EncodingResult<T> = std::result::Result<T, StatusCode>;
 
-#[derive(Clone, Copy, Debug)]
+/// Depth lock holds a reference on the depth gauge. The drop ensures impl that the reference is
+/// decremented even if there is a panic unwind.
+#[derive(Debug)]
+pub struct DepthLock {
+    depth_gauge: Arc<Mutex<DepthGauge>>,
+}
+
+impl Drop for DepthLock {
+    fn drop(&mut self) {
+        let mut dg = trace_lock!(self.depth_gauge);
+        if dg.current_depth > 0 {
+            dg.current_depth -= 1;
+        }
+        // panic if current_depth == 0 is probably overkill and might have issues when drop
+        // is called from a panic.
+    }
+}
+
+impl DepthLock {
+    /// The depth lock tests if the depth can increment and then obtains a lock on it.
+    /// The lock will decrement the depth when it drops to ensure proper behaviour during unwinding.
+    pub fn obtain(
+        depth_gauge: Arc<Mutex<DepthGauge>>,
+    ) -> core::result::Result<DepthLock, StatusCode> {
+        let mut dg = trace_lock!(depth_gauge);
+        if dg.current_depth >= dg.max_depth {
+            warn!("Decoding in stream aborted due maximum recursion depth being reached");
+            Err(StatusCode::BadDecodingError)
+        } else {
+            dg.current_depth += 1;
+            drop(dg);
+            Ok(Self { depth_gauge })
+        }
+    }
+}
+
+/// Depth gauge is used on potentially recursive structures like Variant & ExtensionObject during
+/// decoding to limit the depth the decoder will go before giving up.
+#[derive(Debug)]
+pub struct DepthGauge {
+    /// Maximum decoding depth for recursive elements. Triggers when current depth equals max depth.
+    pub(self) max_depth: usize,
+    /// Current decoding depth for recursive elements.
+    pub(self) current_depth: usize,
+}
+
+impl Default for DepthGauge {
+    fn default() -> Self {
+        Self {
+            max_depth: constants::MAX_DECODING_DEPTH,
+            current_depth: 0,
+        }
+    }
+}
+
+impl DepthGauge {
+    pub fn minimal() -> Self {
+        Self {
+            max_depth: 1,
+            ..Default::default()
+        }
+    }
+    pub fn max_depth(&self) -> usize {
+        self.max_depth
+    }
+    pub fn current_depth(&self) -> usize {
+        self.current_depth
+    }
+}
+
+#[derive(Clone, Debug)]
 pub struct DecodingOptions {
     /// Time offset between the client and the server, only used by the client when it's configured
     /// to ignore time skew.
     pub client_offset: Duration,
-    /// Maximum size of a message chunk in bytes. 0 means no limit
+    /// Maximum size of a message in bytes. 0 means no limit.
+    pub max_message_size: usize,
+    /// Maximum number of chunks. 0 means no limit.
     pub max_chunk_count: usize,
     /// Maximum length in bytes (not chars!) of a string. 0 actually means 0, i.e. no string permitted
     pub max_string_length: usize,
     /// Maximum length in bytes of a byte string. 0 actually means 0, i.e. no byte string permitted
     pub max_byte_string_length: usize,
     /// Maximum number of array elements. 0 actually means 0, i.e. no array permitted
     pub max_array_length: usize,
+    /// Decoding depth gauge is used to check for recursion
+    pub decoding_depth_gauge: Arc<Mutex<DepthGauge>>,
 }
 
 impl Default for DecodingOptions {
     fn default() -> Self {
         DecodingOptions {
             client_offset: Duration::zero(),
-            max_chunk_count: 0,
+            max_message_size: constants::MAX_MESSAGE_SIZE,
+            max_chunk_count: constants::MAX_CHUNK_COUNT,
             max_string_length: constants::MAX_STRING_LENGTH,
             max_byte_string_length: constants::MAX_BYTE_STRING_LENGTH,
             max_array_length: constants::MAX_ARRAY_LENGTH,
+            decoding_depth_gauge: Arc::new(Mutex::new(DepthGauge::default())),
         }
     }
 }
 
 impl DecodingOptions {
     /// This can be useful for decoding extension objects where the payload is not expected to contain
-    /// any string or array.
+    /// a large value.
     pub fn minimal() -> Self {
         DecodingOptions {
-            client_offset: Duration::zero(),
-            max_chunk_count: 0,
-            max_string_length: 0,
-            max_byte_string_length: 0,
-            max_array_length: 0,
+            max_string_length: 8192,
+            max_byte_string_length: 8192,
+            max_array_length: 8192,
+            decoding_depth_gauge: Arc::new(Mutex::new(DepthGauge::minimal())),
+            ..Default::default()
         }
     }
+
+    /// For test only. Having a separate function makes it easier to control calls to DecodingOptions::default().
+    #[cfg(test)]
+    pub fn test() -> Self {
+        Self::default()
+    }
+
+    pub fn depth_lock(&self) -> core::result::Result<DepthLock, StatusCode> {
+        DepthLock::obtain(self.decoding_depth_gauge.clone())
+    }
 }
 
 /// OPC UA Binary Encoding interface. Anything that encodes to binary must implement this. It provides

@@ -84,6 +84,8 @@ impl BinaryEncoder<ExtensionObject> for ExtensionObject {
     }
 
     fn decode<S: Read>(stream: &mut S, decoding_options: &DecodingOptions) -> EncodingResult<Self> {
+        // Extension object is depth checked to prevent deep recursion
+        let _depth_lock = decoding_options.depth_lock()?;
         let node_id = NodeId::decode(stream, decoding_options)?;
         let encoding_type = u8::decode(stream, decoding_options)?;
         let body = match encoding_type {

@@ -17,27 +17,34 @@
 pub mod profiles {
     pub const TRANSPORT_PROFILE_URI_BINARY: &str =
         "http://opcfoundation.org/UA-Profile/Transport/uatcp-uasc-uabinary";
-
     pub const SECURITY_USER_TOKEN_POLICY_ANONYMOUS: &str =
         "http://opcfoundation.org/UA-Profile/Security/UserToken/Anonymous";
     pub const SECURITY_USER_TOKEN_POLICY_USERPASS: &str =
         "http://opcfoundation.org/UA-Profile/ Security/UserToken-Server/UserNamePassword";
 }
 
 pub mod constants {
-    /// Default OPC UA port number. Used by a discovery server. Other servers would normally run
-    /// on a different port. So OPC UA for Rust does not use this nr by default but it is used
-    /// implicitly in opc.tcp:// urls and elsewhere.
-    pub const DEFAULT_OPC_UA_SERVER_PORT: u16 = 4840;
-    /// Maximum number of elements in an array
+    /// Default maximum number of elements in an array
     pub const MAX_ARRAY_LENGTH: usize = 1000;
-    /// Maximum size of a string in chars
+    /// Default maximum size of a string in chars
     pub const MAX_STRING_LENGTH: usize = 65535;
-    /// Maximum size of a byte string in bytes
+    /// Default maximum size of a byte string in bytes
     pub const MAX_BYTE_STRING_LENGTH: usize = 65535;
-    /// Maximum size of a certificate to send
-    pub const MAX_CERTIFICATE_LENGTH: u32 = 32767;
-
+    /// Default maximum size of a certificate to send
+    pub const MAX_CERTIFICATE_LENGTH: usize = 32767;
+    /// Default maximum size of a message in bytes. 0 is any length, i.e. the other end can send a message of any size which is
+    /// not recommended in a server configuration. Override in the client / server config.
+    /// In clients, max message size is only preferred size since it can be adjusted by the server during the handshake.
+    pub const MAX_MESSAGE_SIZE: usize = 65535 * MAX_CHUNK_COUNT;
+    /// Default maximum number of chunks in a single message. 0 is any number but this is not recommended
+    /// as the default since server memory could be exhausted. Default number can be overridden
+    /// by client / server config which is where it should happen if you want a different figure. In clients
+    /// chunk size is a preferred value since the server can modify it during the handshake.
+    pub const MAX_CHUNK_COUNT: usize = 5;
+    /// Default maximum decoding depth for recursive data structures, i.e. if data is nested deeper than this it is
+    /// an error during decoding. This is a security measure to stop deeply nested junk being sent to
+    /// a server / client.
+    pub const MAX_DECODING_DEPTH: usize = 10;
     /// URI supplied for the None security policy
     pub const SECURITY_POLICY_NONE_URI: &str = "http://opcfoundation.org/UA/SecurityPolicy#None";
     /// String used as shorthand in config files, debug etc.for `None` security policy

@@ -109,7 +109,7 @@ impl TryFrom<&ExtensionObject> for Operand {
         let object_id = v
             .object_id()
             .map_err(|_| StatusCode::BadFilterOperandInvalid)?;
-        let decoding_options = DecodingOptions::default();
+        let decoding_options = DecodingOptions::minimal();
         let operand = match object_id {
             ObjectId::ElementOperand_Encoding_DefaultBinary => {
                 Operand::ElementOperand(v.decode_inner::<ElementOperand>(&decoding_options)?)

@@ -1,3 +1,5 @@
+use parking_lot::Mutex;
+use std::sync::Arc;
 use std::{io::Cursor, str::FromStr};
 
 use crate::types::{encoding::DecodingOptions, string::UAString, tests::*};
@@ -102,7 +104,7 @@ fn decode_string_malformed_utf8() {
     // Bytes below are a mangled 水Boy, missing a byte
     let bytes = [0x06, 0x00, 0x00, 0xE6, 0xB0, 0xB4, 0x42, 0x6F, 0x79];
     let mut stream = Cursor::new(bytes);
-    let decoding_options = DecodingOptions::default();
+    let decoding_options = DecodingOptions::test();
     assert_eq!(
         UAString::decode(&mut stream, &decoding_options).unwrap_err(),
         StatusCode::BadDecodingError
@@ -518,7 +520,7 @@ fn null_array() -> EncodingResult<()> {
     length.encode(&mut stream)?;
     let actual = stream.into_inner();
     let mut stream = Cursor::new(actual);
-    let arr = Variant::decode(&mut stream, &DecodingOptions::default())?;
+    let arr = Variant::decode(&mut stream, &DecodingOptions::test())?;
     assert_eq!(
         arr,
         Variant::Array(Box::new(Array {
@@ -529,3 +531,42 @@ fn null_array() -> EncodingResult<()> {
     );
     Ok(())
 }
+
+#[test]
+fn depth_gauge() {
+    let dg = Arc::new(Mutex::new(DepthGauge::default()));
+
+    let max_depth = {
+        let dg = trace_lock!(dg);
+        dg.max_depth()
+    };
+    assert_eq!(max_depth, constants::MAX_DECODING_DEPTH);
+
+    // Iterate the depth
+    {
+        let mut v = Vec::new();
+        for i in 0..max_depth {
+            v.push(DepthLock::obtain(dg.clone()).unwrap());
+        }
+
+        // Depth should now be MAX_DECODING_DEPTH
+        {
+            let dg = trace_lock!(dg);
+            assert_eq!(dg.current_depth(), max_depth);
+        }
+
+        // Next obtain should fail
+        assert_eq!(
+            DepthLock::obtain(dg.clone()).unwrap_err(),
+            StatusCode::BadDecodingError
+        );
+
+        // DepthLocks drop here
+    }
+
+    // Depth should be zero
+    {
+        let dg = trace_lock!(dg);
+        assert_eq!(dg.current_depth(), 0);
+    }
+}

@@ -42,7 +42,7 @@ where
     println!("encoded bytes = {:?}", actual);
     let mut stream = Cursor::new(actual);
 
-    let decoding_options = DecodingOptions::default();
+    let decoding_options = DecodingOptions::test();
     let new_value: T = T::decode(&mut stream, &decoding_options).unwrap();
     println!("new value = {:?}", new_value);
     assert_eq!(expected_value, new_value);

@@ -1012,8 +1012,11 @@ impl Variant {
         } else if Self::test_encoding_flag(encoding_mask, EncodingMask::LOCALIZED_TEXT) {
             Self::from(LocalizedText::decode(stream, decoding_options)?)
         } else if Self::test_encoding_flag(encoding_mask, EncodingMask::EXTENSION_OBJECT) {
+            // Extension object internally does depth checking to prevent deep recursion
             Self::from(ExtensionObject::decode(stream, decoding_options)?)
         } else if Self::test_encoding_flag(encoding_mask, EncodingMask::VARIANT) {
+            // Nested variant is depth checked to prevent deep recursion
+            let _depth_lock = decoding_options.depth_lock()?;
             Variant::Variant(Box::new(Variant::decode(stream, decoding_options)?))
         } else if Self::test_encoding_flag(encoding_mask, EncodingMask::DATA_VALUE) {
             Self::from(DataValue::decode(stream, decoding_options)?)

@@ -39,7 +39,8 @@ endpoints:
     security_mode: None
     user_token_id: ANONYMOUS
 decoding_options:
-  max_chunk_count: 0
+  max_message_size: 327675
+  max_chunk_count: 5
   max_string_length: 65535
   max_byte_string_length: 65535
   max_array_length: 1000

@@ -24,6 +24,10 @@ limits:
   max_byte_string_length: 65535
   min_sampling_interval: 0.1
   min_publishing_interval: 0.1
+  max_message_size: 327675
+  max_chunk_count: 5
+  send_buffer_size: 65535
+  receive_buffer_size: 65535
 performance:
   single_threaded_executor: false
 locale_ids: