Try sips: first, fall back to sip:;transport=tls on 403 per RFC 5630

Copilot · codingjoe · Copilot · commit 265152bf60be · 2026-03-13T16:43:18.000Z
Co-authored-by: codingjoe &lt;1772890+codingjoe@users.noreply.github.com&gt;
diff --git a/tests/sip/test_protocol.py b/tests/sip/test_protocol.py
@@ -682,15 +682,16 @@ async def test_answer__no_address_logs_error(self, caplog):
 
     @pytest.mark.asyncio
     async def test_answer__includes_contact_header(self, fake_rtp_transport):
-        """Include a Contact header with the local SIP address in 200 OK."""
+        """Include a Contact header with the local SIPS address in 200 OK."""
         protocol = FakeProtocol()
         addr = ("192.0.2.1", 5060)
         invite = self._make_invite("answer-contact-1")
         protocol.request_received(invite, addr)
         await self._run_answer(protocol, invite, fake_rtp_transport)
         response, _ = protocol._sent_responses[-1]
         assert "Contact" in response.headers
-        assert response.headers["Contact"].startswith("<sip:")
+        # FakeProtocol uses FakeTransport (TLS, not downgraded) → sips:
+        assert response.headers["Contact"].startswith("<sips:")
 
     @pytest.mark.asyncio
     async def test_answer__includes_allow_header(self, fake_rtp_transport):
@@ -1631,16 +1632,31 @@ class TestRegistration:
     def test_registrar_uri__strips_user_from_aor(self):
         """Derive registrar URI from AOR by stripping the user part."""
         p = make_register_session(aor="sip:alice@example.com")
+        # _is_tls=False by default (no connection_made called)
         assert p.registrar_uri == "sip:example.com"
 
     def test_registrar_uri__preserves_port(self):
         """Preserve a non-default port in the derived registrar URI."""
         p = make_register_session(aor="sip:alice@example.com:5080")
         assert p.registrar_uri == "sip:example.com:5080"
 
-    def test_registrar_uri__normalises_sips_to_sip(self):
-        """sips: AOR is normalised to sip: in the registrar URI (RFC 5630 §4)."""
+    def test_registrar_uri__uses_sips_when_tls(self):
+        """Prefer sips: scheme over TLS before any downgrade (RFC 5630 §4)."""
+        p = make_register_session(aor="sip:alice@example.com")
+        p._is_tls = True
+        assert p.registrar_uri == "sips:example.com"
+
+    def test_registrar_uri__uses_sip_when_downgraded(self):
+        """Fall back to sip: after a sips: rejection (RFC 5630 §4 downgrade)."""
+        p = make_register_session(aor="sip:alice@example.com")
+        p._is_tls = True
+        p._sips_downgraded = True
+        assert p.registrar_uri == "sip:example.com"
+
+    def test_registrar_uri__normalises_sips_aor_when_not_tls(self):
+        """sips: AOR normalised to sip: in the registrar URI when not on TLS."""
         p = make_register_session(aor="sips:alice@example.com")
+        # _is_tls=False → always sip:
         assert p.registrar_uri == "sip:example.com"
 
     async def test_connection_made__sends_register(self):
@@ -1662,7 +1678,8 @@ async def _initialize(self):
         await asyncio.sleep(0.05)
         transport.write.assert_called()
         (data,) = transport.write.call_args[0]
-        assert b"REGISTER sip:example.com SIP/2.0" in data
+        # make_mock_transport() simulates TLS → sips: is preferred first attempt.
+        assert b"REGISTER sips:example.com SIP/2.0" in data
 
     async def test_register__includes_required_headers(self):
         """REGISTER request includes From, To, Call-ID, CSeq, Contact and Expires."""
@@ -1675,7 +1692,8 @@ async def test_register__includes_required_headers(self):
         (data,) = transport.write.call_args[0]
         assert b"From: sip:alice@example.com" in data
         assert b"To: sip:alice@example.com" in data
-        assert b"Contact: <sip:alice@127.0.0.1:5061;transport=tls>" in data
+        # TLS and not downgraded → sips: Contact
+        assert b"Contact: <sips:alice@127.0.0.1:5061>" in data
         assert b"Expires: 3600" in data
 
     async def test_register__increments_cseq(self):
@@ -1855,16 +1873,68 @@ async def test_register__via_branch_is_unique_per_request(self):
         assert branch1 != branch2
 
     async def test_register__contact_uses_local_addr(self):
-        """Contact header uses sip: URI with transport=tls and the local socket address."""
+        """Contact header uses sips: URI when TLS is active (RFC 5630 §4, first attempt)."""
+        p = make_register_session()
+        p.local_address = "10.0.0.5", 5061
+        transport = make_mock_transport("10.0.0.5", 5061)
+        p.transport = transport
+        p._is_tls = True
+        await p.register()
+        (data,) = transport.write.call_args[0]
+        assert b"Contact: <sips:alice@10.0.0.5:5061>" in data
+
+    async def test_register__contact_uses_sip_transport_tls_when_downgraded(self):
+        """Contact header falls back to sip:;transport=tls after a 403 downgrade."""
         p = make_register_session()
         p.local_address = "10.0.0.5", 5061
         transport = make_mock_transport("10.0.0.5", 5061)
         p.transport = transport
         p._is_tls = True
+        p._sips_downgraded = True
         await p.register()
         (data,) = transport.write.call_args[0]
         assert b"Contact: <sip:alice@10.0.0.5:5061;transport=tls>" in data
 
+    async def test_response_received__403_for_sips_retries_with_sip(self):
+        """403 Forbidden on REGISTER when using sips: triggers a sip:;transport=tls retry."""
+        p = make_register_session(username="alice", password="secret")  # noqa: S106
+        transport = make_mock_transport()
+        p.transport = transport
+        p.local_address = ("127.0.0.1", 5061)
+        p._is_tls = True
+        p.response_received(
+            Response(
+                status_code=403,
+                reason="Forbidden (sips URI unsupported)",
+                headers={"CSeq": "1 REGISTER"},
+            ),
+            ("192.0.2.2", 5061),
+        )
+        await asyncio.sleep(0.05)
+        (data,) = transport.write.call_args[0]
+        # After downgrade, REGISTER request-URI must use sip: scheme.
+        assert b"REGISTER sip:" in data
+        # Contact must use sip: with transport=tls, not sips:.
+        assert b"Contact: <sip:alice@127.0.0.1:5061;transport=tls>" in data
+
+    async def test_response_received__403_not_retriggered_after_downgrade(self):
+        """A second 403 after downgrading does not loop — raises NotImplementedError."""
+        p = make_register_session()
+        transport = make_mock_transport()
+        p.transport = transport
+        p.local_address = ("127.0.0.1", 5061)
+        p._is_tls = True
+        p._sips_downgraded = True  # already downgraded
+        with pytest.raises(NotImplementedError):
+            p.response_received(
+                Response(
+                    status_code=403,
+                    reason="Forbidden",
+                    headers={"CSeq": "1 REGISTER"},
+                ),
+                ("192.0.2.2", 5061),
+            )
+
     async def test_data_received__sip_response__calls_response_received(self):
         """data_received routes SIP messages to response_received via TCP stream."""
         received = []
diff --git a/tests/test_main.py b/tests/test_main.py
@@ -88,7 +88,7 @@ def test_transcribe__missing_server_exits_with_error(self):
         assert "server" in result.output.lower() or "SIP_SERVER" in result.output
 
     def test_transcribe__missing_aor_defaults_to_username_at_host(self):
-        """Default AOR to sips:{username}@{server_host} when SIP_AOR is not provided."""
+        """Default AOR to sip:{username}@{server_host} when SIP_AOR is not provided."""
         from voip.__main__ import voip
 
         runner = make_runner()
diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py
@@ -126,6 +126,9 @@ def call_received(self, request: Request) -> None:
     transport: asyncio.Transport | None = dataclasses.field(init=False, default=None)
     #: True when the underlying transport is TLS-wrapped; False for plain TCP.
     _is_tls: bool = dataclasses.field(init=False, default=False)
+    #: True once a 403 Forbidden forces us to fall back from ``sips:`` to
+    #: ``sip:`` with ``transport=tls`` (RFC 5630 §4).
+    _sips_downgraded: bool = dataclasses.field(init=False, default=False)
 
     def __post_init__(self):
         self.call_id = f"{uuid.uuid4()}@{socket.gethostname()}"
@@ -345,8 +348,9 @@ def response_received(
 
         Only processes responses when registration parameters are configured.
         """
-        if response.status_code == Status["OK"] and "REGISTER" in response.headers.get(
-            "CSeq", ""
+        if (
+            response.status_code == Status["OK"]
+            and response.headers.get("CSeq", "").split()[-1:] == ["REGISTER"]
         ):
             logger.info("Registration successful")
             self.registered()
@@ -405,6 +409,22 @@ def response_received(
             else:
                 asyncio.create_task(self.register(authorization=auth_value))
             return
+        if (
+            response.status_code == Status["Forbidden"]
+            and response.headers.get("CSeq", "").split()[-1:] == ["REGISTER"]
+            and self._is_tls
+            and not self._sips_downgraded
+        ):
+            # RFC 5630 §4: some servers reject ``sips:`` URIs even over TLS.
+            # Downgrade to ``sip:`` with ``transport=tls`` and retry once.
+            logger.warning(
+                "REGISTER rejected with 403 Forbidden while using sips: URIs; "
+                "server may not support sips: — retrying with sip:;transport=tls "
+                "(RFC 5630 §4 fallback)"
+            )
+            self._sips_downgraded = True
+            asyncio.create_task(self.register())
+            return
         logger.warning(
             "Unexpected REGISTER response: %s %s", response.status_code, response.reason
         )
@@ -573,7 +593,7 @@ async def _answer(self, request: Request, call_class: type[Call]) -> None:
                         call_id,
                     ),
                     **({"Record-Route": record_route} if record_route else {}),
-                    "Contact": f"<sip:{self.local_address[0]}:{self.local_address[1]}{';transport=tls' if self._is_tls else ''}>",
+                    "Contact": self._build_contact(),
                     "Allow": self.ALLOW,
                     "Supported": "replaces",
                     "Content-Type": "application/sdp",
@@ -616,6 +636,26 @@ def _with_to_tag(self, headers: dict[str, str], call_id: str) -> dict[str, str]:
             "To": headers.get("To", "") + (f";tag={tag}" if tag else ""),
         }
 
+    def _build_contact(self, user: str | None = None) -> str:
+        """Return a ``Contact:`` header value for this UA.
+
+        Prefers the ``sips:`` URI scheme when connected over TLS and the server
+        has not yet rejected it (RFC 5630 §4 first attempt).  After a
+        403-driven downgrade (:attr:`_sips_downgraded`) the compatible
+        ``sip:`` form with ``transport=tls`` is used instead.
+
+        Args:
+            user: SIP user part (e.g. ``"alice"``).  When provided the Contact
+                is of the form ``<scheme:user@host:port>``; otherwise just
+                ``<scheme:host:port>``.
+        """
+        host_port = f"{self.local_address[0]}:{self.local_address[1]}"
+        addr = f"{user}@{host_port}" if user else host_port
+        if self._is_tls and not self._sips_downgraded:
+            return f"<sips:{addr}>"
+        tls_param = ";transport=tls" if self._is_tls else ""
+        return f"<sip:{addr}{tls_param}>"
+
     def ringing(self, request: Request) -> None:
         """Send a 180 Ringing provisional response to the caller.
 
@@ -707,19 +747,21 @@ def reject(
 
     @property
     def registrar_uri(self) -> str:
-        """Registrar Request-URI derived from the AOR (e.g. sip:example.com).
-
-        Per RFC 5630 §4, ``sips:`` URIs require end-to-end TLS on every hop and
-        are not supported by many carriers.  When the AOR uses the ``sips:``
-        scheme we normalise to ``sip:`` here — TLS transport is already signalled
-        by ``Via: SIP/2.0/TLS``, so the scheme conversion is safe and RFC-
-        compliant.
+        """Registrar Request-URI derived from the AOR.
+
+        Over a TLS connection the most secure form, ``sips:host``, is preferred
+        (RFC 5630 §4).  If the server rejects this with ``403 Forbidden``,
+        :meth:`response_received` sets :attr:`_sips_downgraded` to ``True`` and
+        retries; subsequent calls then return the compatible ``sip:host`` form.
+        TLS transport is already signalled by ``Via: SIP/2.0/TLS``, so this
+        downgrade is safe and interoperable.
         """
         if not self.aor:
             raise ValueError("AOR is not configured; cannot derive registrar URI")
         _, _, rest = self.aor.partition(":")
         _, _, hostport = rest.partition("@")
-        return f"sip:{hostport}"
+        scheme = "sips" if (self._is_tls and not self._sips_downgraded) else "sip"
+        return f"{scheme}:{hostport}"
 
     async def register(
         self,
@@ -745,7 +787,7 @@ async def register(
             "To": self.aor,
             "Call-ID": self.call_id,
             "CSeq": f"{self.cseq} REGISTER",
-            "Contact": f"<sip:{user}@{self.local_address[0]}:{self.local_address[1]}{';transport=tls' if self._is_tls else ''}>",
+            "Contact": self._build_contact(user),
             "Expires": "3600",  # 1 hour
             "Max-Forwards": "70",
         }
