Merge pull request #34 from chadgates/fix/33

When address-type is not specified, only use provided value

pyas2lib/as2.py

@@ -985,7 +985,24 @@ def detect_mdn(self):
                 message_id = mdn.get("Original-Message-ID").strip("<>")
                 message_recipient = None
                 if "Original-Recipient" in mdn:
-                    message_recipient = mdn["Original-Recipient"].split(";")[1].strip()
+                    address_type, message_recipient = self._parse_message_recipient(
+                        mdn["Original-Recipient"]
+                    )
                 elif "Final-Recipient" in mdn:
-                    message_recipient = mdn["Final-Recipient"].split(";")[1].strip()
+                    address_type, message_recipient = self._parse_message_recipient(
+                        mdn["Final-Recipient"]
+                    )
         return message_id, message_recipient
+
+    @staticmethod
+    def _parse_message_recipient(recipient: str):
+        """
+        Function parses the recipient values "Original-Recipient: rfc822; 012345678000" into address_type and
+        message_recipient.
+        :param recipient: example: "rfc822; 012345678000"
+        :return: address_type: "rfc822", message_recipient: "012345678000"
+        """
+        if ";" in recipient:
+            return recipient.split(";")[0].strip(), recipient.split(";")[1].strip()
+        else:
+            return None, recipient.strip()

pyas2lib/tests/test_advanced.py

@@ -2,6 +2,7 @@
 import base64
 import os
 from email import message
+from email import message_from_bytes as parse_mime
 
 import pytest
 from pyas2lib import as2
@@ -448,6 +449,72 @@ def test_non_matching_mic(self):
         self.assertEqual(status, "processed/warning")
         self.assertEqual(detailed_status, "Message Integrity check failed.")
 
+    def test_missing_address_type(self):
+        """Test the case where the address type is missing and only a receiver is provided."""
+        self.partner.mdn_mode = as2.SYNCHRONOUS_MDN
+        self.partner.sign = True
+        self.out_message = as2.Message(self.org, self.partner)
+        self.out_message.build(self.test_data)
+
+        # Parse the generated AS2 message as the partner
+        raw_out_message = (
+            self.out_message.headers_str + b"\r\n" + self.out_message.content
+        )
+        in_message = as2.Message()
+        _, _, mdn = in_message.parse(
+            raw_out_message,
+            find_org_cb=self.find_org,
+            find_partner_cb=self.find_partner,
+            find_message_cb=lambda x, y: False,
+        )
+
+        # Remove the address type from the content.
+        patched_content = mdn.content.replace(
+            b"Original-Recipient: rfc822;", b"Original-Recipient:"
+        )
+        patched_content = patched_content.replace(
+            b"Final-Recipient: rfc822;", b"Final-Recipient:"
+        )
+
+        out_mdn = as2.Mdn()
+        out_mdn.payload = parse_mime(mdn.headers_str + b"\r\n" + patched_content)
+        message_id, message_recipient = out_mdn.detect_mdn()
+
+        self.assertEqual(message_recipient, self.partner.as2_name)
+
+    def test_final_recipient_fallback(self):
+        """Test the case where the original recipient is missing, but final recipient is provided."""
+        self.partner.mdn_mode = as2.SYNCHRONOUS_MDN
+        self.partner.sign = True
+        self.out_message = as2.Message(self.org, self.partner)
+        self.out_message.build(self.test_data)
+
+        # Parse the generated AS2 message as the partner
+        raw_out_message = (
+            self.out_message.headers_str + b"\r\n" + self.out_message.content
+        )
+        in_message = as2.Message()
+        _, _, mdn = in_message.parse(
+            raw_out_message,
+            find_org_cb=self.find_org,
+            find_partner_cb=self.find_partner,
+            find_message_cb=lambda x, y: False,
+        )
+
+        # Remove the address type from the content.
+        patched_content = mdn.content.replace(
+            b"Original-Recipient: rfc822; "
+            + self.partner.as2_name.encode("utf-8")
+            + b"\r\n",
+            b"",
+        )
+
+        out_mdn = as2.Mdn()
+        out_mdn.payload = parse_mime(mdn.headers_str + b"\r\n" + patched_content)
+        message_id, message_recipient = out_mdn.detect_mdn()
+
+        self.assertEqual(message_recipient, self.partner.as2_name)
+
     def find_org(self, headers):
         return self.org