feat: handle messages with incorrect BeginString and comp ID (#181)

* Add test case for handling message with invalid begin string

* Fix comp ID checks and add test for incorrect target comp ID handling

* Add test case for handling message with invalid sender comp ID

Author: davidsteiner

crates/hotfix/src/error.rs

@@ -16,8 +16,18 @@ pub enum MessageVerificationError {
 
     /// The comp ID is different from our expectations.
     #[allow(dead_code)]
-    #[error("incorrect comp id {0}")]
-    IncorrectCompId(String),
+    #[error("incorrect comp id {comp_id} ({comp_id_type:?})")]
+    IncorrectCompId {
+        comp_id: String,
+        comp_id_type: CompIdType,
+        msg_seq_num: u64,
+    },
+}
+
+#[derive(Debug)]
+pub enum CompIdType {
+    Sender,
+    Target,
 }
 
 #[derive(Debug, Error)]

crates/hotfix/src/session.rs

@@ -25,7 +25,7 @@ use crate::message::parser::RawFixMessage;
 use crate::store::MessageStore;
 use crate::transport::writer::WriterRef;
 
-use crate::error::MessageVerificationError;
+use crate::error::{CompIdType, MessageVerificationError};
 use crate::message::logout::Logout;
 use crate::message::resend_request::ResendRequest;
 use crate::message::sequence_reset::SequenceReset;
@@ -318,15 +318,15 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
         &self,
         message: &Message,
     ) -> std::result::Result<(), MessageVerificationError> {
-        let begin_string: &str = message.header().get(fix44::BEGIN_STRING).unwrap();
+        let begin_string: &str = message.header().get(fix44::BEGIN_STRING).unwrap_or("");
         if begin_string != self.config.begin_string.as_str() {
             return Err(MessageVerificationError::IncorrectBeginString(
                 begin_string.to_string(),
             ));
         }
 
         let expected_seq_number = self.store.next_target_seq_number();
-        let actual_seq_number: u64 = message.header().get(fix44::MSG_SEQ_NUM).unwrap();
+        let actual_seq_number: u64 = message.header().get(fix44::MSG_SEQ_NUM).unwrap_or_default();
 
         match actual_seq_number.cmp(&expected_seq_number) {
             Ordering::Greater => {
@@ -344,6 +344,28 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
             _ => {}
         }
 
+        // our TargetCompId is always the same as the expected SenderCompId for them
+        let expected_sender_comp_id: &str = self.config.target_comp_id.as_str();
+        let actual_sender_comp_id: &str = message.header().get(fix44::SENDER_COMP_ID).unwrap_or("");
+        if expected_sender_comp_id != actual_sender_comp_id {
+            return Err(MessageVerificationError::IncorrectCompId {
+                comp_id: actual_sender_comp_id.to_string(),
+                comp_id_type: CompIdType::Sender,
+                msg_seq_num: actual_seq_number,
+            });
+        }
+
+        // our SenderCompId is always the same as the expected TargetCompId for them
+        let expected_target_comp_id: &str = self.config.sender_comp_id.as_str();
+        let actual_target_comp_id: &str = message.header().get(fix44::TARGET_COMP_ID).unwrap_or("");
+        if expected_target_comp_id != actual_target_comp_id {
+            return Err(MessageVerificationError::IncorrectCompId {
+                comp_id: actual_target_comp_id.to_string(),
+                comp_id_type: CompIdType::Target,
+                msg_seq_num: actual_seq_number,
+            });
+        }
+
         Ok(())
     }
 
@@ -508,21 +530,40 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
             MessageVerificationError::IncorrectBeginString(begin_string) => {
                 self.handle_incorrect_begin_string(begin_string).await;
             }
-            MessageVerificationError::IncorrectCompId(comp_id) => {
-                self.handle_incorrect_comp_id(comp_id).await;
+            MessageVerificationError::IncorrectCompId {
+                comp_id,
+                comp_id_type,
+                msg_seq_num,
+            } => {
+                self.handle_incorrect_comp_id(comp_id, comp_id_type, msg_seq_num)
+                    .await;
             }
         }
     }
 
     async fn handle_incorrect_begin_string(&mut self, received_begin_string: String) {
-        // TODO: this should be a disconnect (and maybe a reject first?)
-        // see: https://www.fixtrading.org/standards/fix-session-layer-online/#when-to-terminate-a-fix-connection-by-terminating-the-transport-layer-connection-instead-of-sending-a-logout355
-        panic!("incorrect begin string received: {received_begin_string}");
-    }
+        self.logout_and_terminate(&format!(
+            "beginString={received_begin_string} is not supported"
+        ))
+        .await;
+    }
+
+    async fn handle_incorrect_comp_id(
+        &mut self,
+        received_comp_id: String,
+        comp_id_type: CompIdType,
+        msg_seq_num: u64,
+    ) {
+        error!(
+            "rejecting message with incorrect comp ID: {received_comp_id} (type: {comp_id_type:?})"
+        );
+        let reject = Reject::new(msg_seq_num)
+            .session_reject_reason(SessionRejectReason::ValueIsIncorrect)
+            .text(&format!("invalid comp ID {received_comp_id}"));
+        self.send_message(reject).await;
 
-    async fn handle_incorrect_comp_id(&mut self, received_comp_id: String) {
-        // TODO: this should also be a disconnect I think (and maybe a reject first?)
-        panic!("incorrect comp ID received: {received_comp_id}");
+        self.logout_and_terminate("incorrect comp ID received")
+            .await;
     }
 
     async fn handle_sequence_number_too_low(&mut self, actual: u64, expected: u64) {

crates/hotfix/tests/common/setup.rs

@@ -13,6 +13,9 @@ use hotfix_message::fix44::MSG_TYPE;
 pub const HEARTBEAT_INTERVAL: u64 = 30;
 pub const LOGON_TIMEOUT: u64 = 10;
 
+pub const COUNTERPARTY_COMP_ID: &str = "dummy-acceptor";
+pub const OUR_COMP_ID: &str = "dummy-initiator";
+
 pub async fn given_a_connected_session() -> (SessionRef<TestMessage>, MockCounterparty<TestMessage>)
 {
     let message_store = InMemoryMessageStore::default();
@@ -48,8 +51,8 @@ pub async fn given_an_active_session() -> (SessionRef<TestMessage>, MockCounterp
 pub fn create_session_config() -> SessionConfig {
     SessionConfig {
         begin_string: "FIX.4.4".to_string(),
-        sender_comp_id: "dummy-initiator".to_string(),
-        target_comp_id: "dummy-acceptor".to_string(),
+        sender_comp_id: OUR_COMP_ID.to_string(),
+        target_comp_id: COUNTERPARTY_COMP_ID.to_string(),
         data_dictionary_path: None,
         connection_host: "".to_string(),
         connection_port: 0,

crates/hotfix/tests/session_test_cases/invalid_message_tests.rs

@@ -1,12 +1,13 @@
 use crate::common::actions::when;
 use crate::common::assertions::then;
-use crate::common::setup::given_an_active_session;
+use crate::common::setup::{COUNTERPARTY_COMP_ID, OUR_COMP_ID, given_an_active_session};
 use crate::common::test_messages::{TestMessage, replace_field_value};
 use hotfix::message::{FixMessage, generate_message};
 use hotfix::session::Status;
 use hotfix_message::dict::{FieldLocation, FixDatatype};
+use hotfix_message::field_types::Timestamp;
 use hotfix_message::fix44::MSG_TYPE;
-use hotfix_message::message::Message;
+use hotfix_message::message::{Config, Message};
 use hotfix_message::{HardCodedFixFieldDefinition, Part, fix44};
 
 /// Tests that when a counterparty sends a message containing an invalid/unrecognised field,
@@ -34,8 +35,6 @@ async fn test_garbled_message_with_invalid_target_comp_id_gets_ignored() {
 
     // counterparty sends a message with invalid body length, which constitutes a garbled message
     let garbled_message = build_execution_report_with_incorrect_body_length(
-        "dummy-acceptor",
-        "dummy-initiator",
         mock_counterparty.next_target_sequence_number(),
     );
     when(&mut mock_counterparty)
@@ -59,6 +58,79 @@ async fn test_garbled_message_with_invalid_target_comp_id_gets_ignored() {
     then(&mut mock_counterparty).gets_disconnected().await;
 }
 
+/// Tests that when a counterparty sends a message with an invalid BeginString,
+/// the session logs out and disconnects.
+#[tokio::test]
+async fn test_message_with_invalid_begin_string() {
+    let (_session, mut mock_counterparty) = given_an_active_session().await;
+
+    // a message with invalid BeginString is sent by the counterparty
+    let invalid_message = build_execution_report_with_incorrect_begin_string(
+        mock_counterparty.next_target_sequence_number(),
+    );
+    when(&mut mock_counterparty)
+        .sends_raw_message(invalid_message)
+        .await;
+
+    // then we log out and disconnect
+    then(&mut mock_counterparty)
+        .receives(|msg| assert_eq!(msg.header().get::<&str>(MSG_TYPE).unwrap(), "5"))
+        .await;
+    then(&mut mock_counterparty).gets_disconnected().await;
+}
+
+/// Tests that when a counterparty sends a message with an invalid TargetCompId,
+/// the session sends a Reject (MsgType=3) and logs out and disconnects.
+#[tokio::test]
+async fn test_message_with_invalid_target_comp_id() {
+    let (_session, mut mock_counterparty) = given_an_active_session().await;
+
+    // a message with incorrect TargetCompId is sent by the counterparty
+    let invalid_message = build_execution_report_with_comp_id(
+        mock_counterparty.next_target_sequence_number(),
+        COUNTERPARTY_COMP_ID,
+        "WRONG_COMP_ID",
+    );
+    when(&mut mock_counterparty)
+        .sends_raw_message(invalid_message)
+        .await;
+
+    // then we send a reject, log out and disconnect
+    then(&mut mock_counterparty)
+        .receives(|msg| assert_eq!(msg.header().get::<&str>(MSG_TYPE).unwrap(), "3"))
+        .await;
+    then(&mut mock_counterparty)
+        .receives(|msg| assert_eq!(msg.header().get::<&str>(MSG_TYPE).unwrap(), "5"))
+        .await;
+    then(&mut mock_counterparty).gets_disconnected().await;
+}
+
+/// Tests that when a counterparty sends a message with an invalid SenderCompId,
+/// the session sends a Reject (MsgType=3) and logs out and disconnects.
+#[tokio::test]
+async fn test_message_with_invalid_sender_comp_id() {
+    let (_session, mut mock_counterparty) = given_an_active_session().await;
+
+    // a message with incorrect SenderCompId is sent by the counterparty
+    let invalid_message = build_execution_report_with_comp_id(
+        mock_counterparty.next_target_sequence_number(),
+        "WRONG_COMP_ID",
+        OUR_COMP_ID,
+    );
+    when(&mut mock_counterparty)
+        .sends_raw_message(invalid_message)
+        .await;
+
+    // then we send a reject, log out and disconnect
+    then(&mut mock_counterparty)
+        .receives(|msg| assert_eq!(msg.header().get::<&str>(MSG_TYPE).unwrap(), "3"))
+        .await;
+    then(&mut mock_counterparty)
+        .receives(|msg| assert_eq!(msg.header().get::<&str>(MSG_TYPE).unwrap(), "5"))
+        .await;
+    then(&mut mock_counterparty).gets_disconnected().await;
+}
+
 /// A new order message with an extra, invalid field.
 #[derive(Clone, Debug)]
 struct ExecutionReportWithInvalidField {
@@ -121,16 +193,45 @@ pub const CUSTOM_FIELD: &HardCodedFixFieldDefinition = &HardCodedFixFieldDefinit
     location: FieldLocation::Body,
 };
 
-fn build_execution_report_with_incorrect_body_length(
-    sender_comp_id: &str,
-    target_comp_id: &str,
-    msg_seq_num: usize,
-) -> Vec<u8> {
+fn build_execution_report_with_incorrect_body_length(msg_seq_num: usize) -> Vec<u8> {
     let report = TestMessage::dummy_execution_report();
     let mut raw_message =
-        generate_message(sender_comp_id, target_comp_id, msg_seq_num, report).unwrap();
+        generate_message(COUNTERPARTY_COMP_ID, OUR_COMP_ID, msg_seq_num, report).unwrap();
 
     replace_field_value(&mut raw_message, 9, b"999");
 
     raw_message
 }
+
+fn build_execution_report_with_incorrect_begin_string(msg_seq_num: usize) -> Vec<u8> {
+    let report = TestMessage::dummy_execution_report();
+
+    // we expect BeginString FIX.4.4 but this message contains FIX.4.2
+    let mut msg = Message::new("FIX.4.2", report.message_type());
+    msg.set(fix44::SENDER_COMP_ID, COUNTERPARTY_COMP_ID);
+    msg.set(fix44::TARGET_COMP_ID, OUR_COMP_ID);
+    msg.set(fix44::MSG_SEQ_NUM, msg_seq_num);
+    msg.set(fix44::SENDING_TIME, Timestamp::utc_now());
+
+    report.write(&mut msg);
+
+    msg.encode(&Config::default()).unwrap()
+}
+
+fn build_execution_report_with_comp_id(
+    msg_seq_num: usize,
+    sender_comp_id: &str,
+    target_comp_id: &str,
+) -> Vec<u8> {
+    let report = TestMessage::dummy_execution_report();
+
+    let mut msg = Message::new("FIX.4.4", report.message_type());
+    msg.set(fix44::SENDER_COMP_ID, sender_comp_id);
+    msg.set(fix44::TARGET_COMP_ID, target_comp_id);
+    msg.set(fix44::MSG_SEQ_NUM, msg_seq_num);
+    msg.set(fix44::SENDING_TIME, Timestamp::utc_now());
+
+    report.write(&mut msg);
+
+    msg.encode(&Config::default()).unwrap()
+}