Compliant cancel message (#29)

* Compliant cancel message

Make the cancel message more compliant with the SIP specification.

* Fix formatting

* Handle 487 response

Send an ACK for "487 Request Terminated" response to Cancel message.

Author: jaminh

tests/test_sip.py

@@ -1,7 +1,9 @@
 """Test voip_utils SIP functionality."""
 
-from voip_utils.sip import SipEndpoint, get_sip_endpoint
+from voip_utils.sip import CallInfo, SdpInfo, SipDatagramProtocol, SipEndpoint, SipMessage, get_sip_endpoint
+from unittest.mock import Mock
 
+_CRLF = "\r\n"
 
 def test_parse_header_for_uri():
     endpoint = SipEndpoint('"Test Name" <sip:[email protected]>')
@@ -84,3 +86,44 @@ def test_get_sip_endpoint_with_scheme():
     assert endpoint.description is None
     assert endpoint.username is None
     assert endpoint.uri == "sips:example.com"
+
+class MockSipDatagramProtocol(SipDatagramProtocol):
+    def on_call(self, call_info: CallInfo):
+        pass
+
+def test_cancel():
+    protocol = MockSipDatagramProtocol(SdpInfo("username", 5, "session", "version"))
+    source = get_sip_endpoint("testsource")
+    destination = get_sip_endpoint("destination")
+    invite_lines = [
+        f"INVITE {destination.uri} SIP/2.0",
+        f"Via: SIP/2.0/UDP {source.host}:{source.port}",
+        f"From: {source.sip_header}",
+        f"Contact: {source.sip_header}",
+        f"To: {destination.sip_header}",
+        f"Call-ID: 100",
+        "CSeq: 50 INVITE",
+        f"User-Agent: test-agent 1.0",
+        "Allow: INVITE, ACK, OPTIONS, CANCEL, BYE, SUBSCRIBE, NOTIFY, INFO, REFER, UPDATE",
+        "Accept: application/sdp, application/dtmf-relay",
+        "Content-Type: application/sdp",
+        "Content-Length: 0",
+        "",
+    ]
+    invite_text = _CRLF.join(invite_lines) + _CRLF
+    invite_msg = SipMessage.parse_sip(invite_text, False)
+
+    call_info = CallInfo(
+        caller_endpoint=destination,
+        local_endpoint=source,
+        caller_rtp_port=12345,
+        server_ip=source.host,
+        headers=invite_msg.headers,
+    )
+
+    transport = Mock()
+    protocol.connection_made(transport)
+    protocol.cancel_call(call_info)
+
+    transport.sendto.assert_called_once_with(b'CANCEL sip:destination SIP/2.0\r\nVia: SIP/2.0/UDP testsource:5060\r\nFrom: sip:testsource\r\nTo: sip:destination\r\nCall-ID: 100\r\nCSeq: 50 CANCEL\r\nUser-Agent: voip-utils 1.0\r\nContent-Length: 0\r\n\r\n', ('destination', 5060))
+

voip_utils/sip.py

@@ -85,7 +85,7 @@ class SipMessage:
     body: str
 
     @staticmethod
-    def parse_sip(message: str) -> SipMessage:
+    def parse_sip(message: str, header_lowercase: bool = True) -> SipMessage:
         """Parse a SIP message into a SipMessage object."""
         lines = message.splitlines()
 
@@ -115,7 +115,7 @@ def parse_sip(message: str) -> SipMessage:
                 break
             else:
                 key, value = line.split(":", maxsplit=1)
-                headers[key.lower()] = value.strip()
+                headers[key.lower() if header_lowercase else key] = value.strip()
 
         body = message[offset:]
         return SipMessage(protocol, method, request_uri, code, reason, headers, body)
@@ -220,6 +220,11 @@ def get_rtp_info(body: str) -> RtpInfo:
     return RtpInfo(rtp_ip=rtp_ip, rtp_port=rtp_port, payload_type=opus_payload_type)
 
 
+def get_header(headers: dict[str, str], name: str) -> tuple[str, str] | None:
+    """Get a header entry using a case insensitive key comparison."""
+    return next(((k, v) for k, v in headers.items() if k.lower() == name.lower()), None)
+
+
 class SipDatagramProtocol(asyncio.DatagramProtocol, ABC):
     """UDP server for the Session Initiation Protocol (SIP)."""
 
@@ -289,25 +294,28 @@ def outgoing_call(
             (destination.host, destination.port),
         )
 
+        invite_msg = SipMessage.parse_sip(invite_text, False)
+
         return CallInfo(
             caller_endpoint=destination,
             local_endpoint=source,
             caller_rtp_port=rtp_port,
             server_ip=source.host,
-            headers={"call-id": call_id},
+            headers=invite_msg.headers,
         )
 
     def hang_up(self, call_info: CallInfo):
         """Hang up the call when finished"""
         if self.transport is None:
             raise RuntimeError("No transport available for sending hangup.")
 
+        call_id = get_header(call_info.headers, "call-id")[1]
         bye_lines = [
             f"BYE {call_info.caller_endpoint.uri} SIP/2.0",
             f"Via: SIP/2.0/UDP {call_info.local_endpoint.host}:{call_info.local_endpoint.port}",
             f"From: {call_info.local_endpoint.sip_header}",
             f"To: {call_info.caller_endpoint.sip_header}",
-            f"Call-ID: {call_info.headers['call-id']}",
+            f"Call-ID: {call_id}",
             "CSeq: 51 BYE",
             f"User-Agent: {VOIP_UTILS_AGENT} 1.0",
             "Content-Length: 0",
@@ -328,17 +336,27 @@ def cancel_call(self, call_info: CallInfo):
         if self.transport is None:
             raise RuntimeError("No transport available for sending cancel.")
 
-        cancel_lines = [
-            f"CANCEL {call_info.caller_endpoint.uri} SIP/2.0",
-            f"Via: SIP/2.0/UDP {call_info.local_endpoint.host}:{call_info.local_endpoint.port}",
-            f"From: {call_info.local_endpoint.sip_header}",
-            f"To: {call_info.caller_endpoint.sip_header}",
-            f"Call-ID: {call_info.headers['call-id']}",
-            "CSeq: 51 CANCEL",
-            f"User-Agent: {VOIP_UTILS_AGENT} 1.0",
-            "Content-Length: 0",
-            "",
+        required_headers = ("via", "from", "to", "call-id")
+
+        cancel_headers = [
+            f"{k}: {v}"
+            for k, v in call_info.headers.items()
+            if k.lower() in required_headers
         ]
+
+        cseq_header, cseq_value = get_header(call_info.headers, "cseq")
+        cseq_num = cseq_value.split()[0]
+
+        cancel_lines = (
+            [f"CANCEL {call_info.caller_endpoint.uri} SIP/2.0"]
+            + cancel_headers
+            + [
+                f"{cseq_header}: {cseq_num} CANCEL",
+                f"User-Agent: {VOIP_UTILS_AGENT} 1.0",
+                "Content-Length: 0",
+                "",
+            ]
+        )
         _LOGGER.debug("Canceling call...")
         cancel_text = _CRLF.join(cancel_lines) + _CRLF
         cancel_bytes = cancel_text.encode("utf-8")
@@ -347,7 +365,7 @@ def cancel_call(self, call_info: CallInfo):
             (call_info.caller_endpoint.host, call_info.caller_endpoint.port),
         )
 
-        self._end_outgoing_call(call_info.headers["call-id"])
+        self._end_outgoing_call(get_header(call_info.headers, "call-id")[1])
         self.on_hangup(call_info)
 
     def _register_outgoing_call(self, call_id: str, rtp_port: int):
@@ -482,6 +500,31 @@ def datagram_received(self, data: bytes, addr):
                 # TODO: Verify that the call / sequence IDs match our outgoing INVITE
                 _LOGGER.debug("Received response [%s]", message)
                 is_ok = smsg.code == "200" and smsg.reason == "OK"
+                if smsg.code == "487":
+                    # A 487 Request Terminated will be sent in response to a Cancel message
+                    _LOGGER.debug("Got 487 Request Terminated")
+                    caller_endpoint = None
+                    if smsg.headers.get("to") is not None:
+                        caller_endpoint = SipEndpoint(smsg.headers.get("to", ""))
+                    else:
+                        caller_endpoint = get_sip_endpoint(
+                            caller_ip, port=caller_sip_port
+                        )
+                    cseq_num = get_header(smsg.headers, "cseq")[1].split()[0]
+                    ack_lines = [
+                        f"ACK {caller_endpoint.uri} SIP/2.0",
+                        f"Via: {smsg.headers['via']}",
+                        f"From: {smsg.headers['from']}",
+                        f"To: {smsg.headers['to']}",
+                        f"Call-ID: {smsg.headers['call-id']}",
+                        f"CSeq: {cseq_num} ACK",
+                        f"User-Agent: {VOIP_UTILS_AGENT} 1.0",
+                        "Content-Length: 0",
+                    ]
+                    ack_text = _CRLF.join(ack_lines) + _CRLF
+                    ack_bytes = ack_text.encode("utf-8")
+                    self.transport.sendto(ack_bytes, (caller_ip, caller_sip_port))
+                    return
                 if not is_ok:
                     _LOGGER.debug("Received non-OK response [%s]", message)
                     return