diff --git a/docs/feature_roadmap.md b/docs/feature_roadmap.md index 283dee7..f84a25c 100644 --- a/docs/feature_roadmap.md +++ b/docs/feature_roadmap.md @@ -8,6 +8,9 @@ SIP User Agent Client (UAC) over TLS/TCP ([RFC 3261]). Handles incoming `INVITE`, `BYE`, `ACK`, `CANCEL`, and `OPTIONS` requests, carrier `REGISTER` with digest authentication ([RFC 8760]: MD5, SHA-256, SHA-512/256), and double-CRLF keepalive ping/pong ([RFC 5626 §4.4.1]). +Client-initiated keepalive pings, `Supported: outbound` and `;ob` Contact +parameter ([RFC 5626 §5]), and automatic reconnection with exponential +back-off ensure robust long-running sessions. ### Media Transport (RTP/SRTP) diff --git a/docs/rfc_status.md b/docs/rfc_status.md index 7ff7c72..d2b1dea 100644 --- a/docs/rfc_status.md +++ b/docs/rfc_status.md @@ -2,14 +2,14 @@ ## SIP Signaling -| RFC | Title | Status | Notes | -| --------------------------------------------------------- | --------------------------------------------- | -------- | ----------------------------------------------------------------------- | -| [RFC 3261](https://datatracker.ietf.org/doc/html/rfc3261) | SIP: Session Initiation Protocol | Partial | UAC only; REGISTER, INVITE, BYE, and digest authentication over TLS/TCP | -| [RFC 5626](https://datatracker.ietf.org/doc/html/rfc5626) | Managing Client-Initiated Connections in SIP | Partial | Double-CRLF keepalive ping/pong (§4.4.1) | -| [RFC 8760](https://datatracker.ietf.org/doc/html/rfc8760) | SIP Digest Authentication Using AES-HMAC-SHA2 | Complete | MD5, SHA-256, and SHA-512/256 digest responses | -| [RFC 3824](https://datatracker.ietf.org/doc/html/rfc3824) | Using E.164 Numbers with SIP | Planned | Phone number mapping into SIP/ENUM | -| [RFC 3966](https://datatracker.ietf.org/doc/html/rfc3966) | The tel URI for Telephone Numbers | Planned | Canonical `tel:` URI scheme | -| [RFC 6116](https://datatracker.ietf.org/doc/html/rfc6116) | The E.164 to URI DDDS Application (ENUM) | Planned | DNS-based E.164 number-to-URI mapping | +| RFC | Title | Status | Notes | +| --------------------------------------------------------- | --------------------------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| [RFC 3261](https://datatracker.ietf.org/doc/html/rfc3261) | SIP: Session Initiation Protocol | Partial | UAC only; REGISTER, INVITE, BYE, and digest authentication over TLS/TCP | +| [RFC 5626](https://datatracker.ietf.org/doc/html/rfc5626) | Managing Client-Initiated Connections in SIP | Complete | Double-CRLF keepalive ping/pong (§4.4.1); client keepalive task; `Supported: outbound` and `;ob` Contact parameter (§5); reconnect with exponential back-off | +| [RFC 8760](https://datatracker.ietf.org/doc/html/rfc8760) | SIP Digest Authentication Using AES-HMAC-SHA2 | Complete | MD5, SHA-256, and SHA-512/256 digest responses | +| [RFC 3824](https://datatracker.ietf.org/doc/html/rfc3824) | Using E.164 Numbers with SIP | Planned | Phone number mapping into SIP/ENUM | +| [RFC 3966](https://datatracker.ietf.org/doc/html/rfc3966) | The tel URI for Telephone Numbers | Planned | Canonical `tel:` URI scheme | +| [RFC 6116](https://datatracker.ietf.org/doc/html/rfc6116) | The E.164 to URI DDDS Application (ENUM) | Planned | DNS-based E.164 number-to-URI mapping | ## Media Transport diff --git a/tests/sip/test_protocol.py b/tests/sip/test_protocol.py index 2cae877..c63f9ff 100644 --- a/tests/sip/test_protocol.py +++ b/tests/sip/test_protocol.py @@ -2,6 +2,7 @@ import asyncio import dataclasses +import datetime import hashlib import ipaddress import re @@ -1651,6 +1652,75 @@ def call_received(self, request): await asyncio.sleep(0.05) assert len(protocol._sent) == 1 + async def test_run_keepalive__sends_double_crlf(self): + """Keep-alive task sends double-CRLF after the configured interval.""" + protocol = SIP( + outbound_proxy=("127.0.0.1", 5061), + aor="sip:test@example.com", + rtp_stun_server_address=None, + keepalive_interval=datetime.timedelta(seconds=0.01), + ) + transport = make_mock_transport() + protocol.connection_made(transport) + transport.write.reset_mock() + await asyncio.sleep(0.05) + transport.write.assert_any_call(b"\r\n\r\n") + protocol._keepalive_task.cancel() + protocol._initialize_task.cancel() + + async def test_run_keepalive__stops_when_transport_cleared(self): + """Keep-alive loop exits cleanly when the transport is set to None.""" + protocol = SIP( + outbound_proxy=("127.0.0.1", 5061), + aor="sip:test@example.com", + rtp_stun_server_address=None, + keepalive_interval=datetime.timedelta(seconds=0.01), + ) + transport = make_mock_transport() + protocol.connection_made(transport) + transport.write.reset_mock() + protocol.transport = None + await asyncio.sleep(0.05) + transport.write.assert_not_called() + protocol._initialize_task.cancel() + + async def test_connection_lost__cancels_and_clears_keepalive_task(self): + """connection_lost cancels the keepalive task and clears _keepalive_task.""" + protocol = SIP( + outbound_proxy=("127.0.0.1", 5061), + aor="sip:test@example.com", + rtp_stun_server_address=None, + ) + + async def _long_running() -> None: + await asyncio.sleep(100) + + task = asyncio.get_running_loop().create_task(_long_running()) + protocol._keepalive_task = task + protocol.connection_lost(None) + assert protocol._keepalive_task is None + await asyncio.sleep(0) + assert task.done() + + async def test_connection_lost__sets_disconnected_event(self): + """connection_lost sets the disconnected_event.""" + protocol = SIP(outbound_proxy=("127.0.0.1", 5061), aor="sip:test@example.com") + assert not protocol.disconnected_event.is_set() + protocol.connection_lost(None) + assert protocol.disconnected_event.is_set() + + async def test_disconnected_event__resolves_after_connection_lost(self): + """disconnected_event resolves once connection_lost is called.""" + protocol = SIP(outbound_proxy=("127.0.0.1", 5061), aor="sip:test@example.com") + + async def lose_connection() -> None: + await asyncio.sleep(0.01) + protocol.connection_lost(None) + + asyncio.create_task(lose_connection()) + await asyncio.wait_for(protocol.disconnected_event.wait(), timeout=1.0) + assert protocol.disconnected_event.is_set() + # --------------------------------------------------------------------------- # Tests for SIP REGISTER / digest-auth / response handling @@ -1759,9 +1829,11 @@ 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 - # sip: AOR over TLS → sip:;transport=tls Contact - assert b"Contact: " in data + # sip: AOR over TLS → sip:;transport=tls Contact with RFC 5626 ;ob parameter + assert b"Contact: " in data assert b"Expires: 3600" in data + # RFC 5626 §5 outbound keep-alive support advertised + assert b"Supported: outbound" in data async def test_register__increments_cseq(self): """CSeq increments with each REGISTER sent.""" @@ -1936,7 +2008,7 @@ 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:;transport=tls when AOR is sip: over TLS.""" + """Contact header uses sip:;transport=tls;ob when AOR is sip: over TLS.""" p = make_register_session() p.local_address = (ipaddress.IPv4Address("10.0.0.5"), 5061) transport = make_mock_transport("10.0.0.5", 5061) @@ -1944,10 +2016,10 @@ async def test_register__contact_uses_local_addr(self): p._is_tls = True await p.register() (data,) = transport.write.call_args[0] - assert b"Contact: " in data + assert b"Contact: " in data async def test_register__contact_uses_sips_when_aor_is_sips(self): - """Contact header uses sips: when AOR scheme is sips:.""" + """Contact header uses sips: with ;ob when AOR scheme is sips:.""" p = make_register_session(aor="sips:alice@example.com") p.local_address = (ipaddress.IPv4Address("10.0.0.5"), 5061) transport = make_mock_transport("10.0.0.5", 5061) @@ -1955,7 +2027,7 @@ async def test_register__contact_uses_sips_when_aor_is_sips(self): p._is_tls = True await p.register() (data,) = transport.write.call_args[0] - assert b"Contact: " in data + assert b"Contact: " in data async def test_register__contact_wraps_ipv6_in_brackets(self): """Contact header wraps an IPv6 local address in square brackets.""" @@ -1966,7 +2038,7 @@ async def test_register__contact_wraps_ipv6_in_brackets(self): p._is_tls = True await p.register() (data,) = transport.write.call_args[0] - assert b"Contact: " in data + assert b"Contact: " in data async def test_register__via_wraps_ipv6_in_brackets(self): """Via header wraps an IPv6 local address in square brackets.""" @@ -2085,6 +2157,30 @@ async def test_response_received__unexpected_status__raises_registration_error( ("192.0.2.2", 5060), ) + def test_build_contact__default__no_ob_param(self): + """Contact without ob=True has no ;ob parameter.""" + p = make_register_session() + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) + p._is_tls = True + contact = p._build_contact("alice") + assert ";ob" not in contact + + def test_build_contact__ob_true__includes_ob_uri_param(self): + """Contact with ob=True includes the ;ob URI parameter (RFC 5626 §5).""" + p = make_register_session() + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) + p._is_tls = True + assert p._build_contact("alice", ob=True) == ( + "" + ) + + def test_build_contact__sips_with_ob__includes_ob_before_closing_bracket(self): + """sips: Contact with ob=True places ;ob inside the angle brackets.""" + p = make_register_session(aor="sips:alice@example.com") + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) + p._is_tls = True + assert p._build_contact("alice", ob=True) == "" + # --------------------------------------------------------------------------- # Tests for digest_response (RFC 3261 §22, RFC 8760) diff --git a/tests/test_main.py b/tests/test_main.py index ee0d55e..9accfc5 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -496,6 +496,74 @@ async def _run_whisper(): asyncio.run(_run_whisper()) + def test_transcribe__retries_on_os_error(self): + """_connect_sip retries when the outbound connection fails with OSError.""" + from unittest.mock import AsyncMock + + call_count = 0 + + async def fake_connection(factory, *, host, port, ssl): + nonlocal call_count + call_count += 1 + if call_count < 3: + raise OSError("Connection refused") + raise KeyboardInterrupt + + with ( + patch.dict(sys.modules, _WHISPER_STUBS), + patch("asyncio.get_event_loop"), + patch("voip.__main__.asyncio.get_running_loop") as mock_loop, + patch("voip.__main__.asyncio.sleep", new=AsyncMock()), + ): + mock_loop.return_value.create_connection = fake_connection + make_runner().invoke( + voip, + [ + "sip", + "--password=p", + "--stun-server=none", + "sips:alice@example.com", + "transcribe", + ], + catch_exceptions=False, + ) + assert call_count == 3 + + def test_transcribe__reconnects_after_disconnect(self): + """_connect_sip reconnects when the disconnected_event is set.""" + from unittest.mock import AsyncMock + + call_count = 0 + + async def fake_connection(factory, *, host, port, ssl): + nonlocal call_count + call_count += 1 + protocol = factory() + if call_count == 1: + protocol.disconnected_event.set() + return MagicMock(), protocol + raise KeyboardInterrupt + + with ( + patch.dict(sys.modules, _WHISPER_STUBS), + patch("asyncio.get_event_loop"), + patch("voip.__main__.asyncio.get_running_loop") as mock_loop, + patch("voip.__main__.asyncio.sleep", new=AsyncMock()), + ): + mock_loop.return_value.create_connection = fake_connection + make_runner().invoke( + voip, + [ + "sip", + "--password=p", + "--stun-server=none", + "sips:alice@example.com", + "transcribe", + ], + catch_exceptions=False, + ) + assert call_count == 2 + class TestAgentCLI: def test_agent__sips_aor_uses_tls(self): diff --git a/voip/__main__.py b/voip/__main__.py index b0f4510..3115f8e 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -24,6 +24,9 @@ ) from e +logger = logging.getLogger("voip") + + #: Standard SIP/TCP port — plain text, no TLS (RFC 3261 §18.2). SIP_TCP_PORT = 5060 #: Standard SIP/TLS port (RFC 3261 §26.2.2). @@ -244,7 +247,12 @@ async def _connect_sip( use_tls: bool, no_verify_tls: bool, ) -> None: - """Connect to a SIP proxy and wait indefinitely.""" + """Connect to a SIP proxy with automatic reconnection on failure. + + Retries with exponential back-off (1 s → 2 s → … → 60 s) after each + failed connection or dropped session so the process stays running without + manual intervention. + """ loop = asyncio.get_running_loop() ssl_context: ssl.SSLContext | None = None if use_tls: @@ -252,13 +260,24 @@ async def _connect_sip( if no_verify_tls: ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE - await loop.create_connection( - session_factory, - host=str(proxy_addr[0]), - port=proxy_addr[1], - ssl=ssl_context, - ) - await asyncio.Future() + backoff_secs = 1 + while True: + try: + _, protocol = await loop.create_connection( + session_factory, + host=str(proxy_addr[0]), + port=proxy_addr[1], + ssl=ssl_context, + ) + backoff_secs = 1 + await protocol.disconnected_event.wait() + logger.info("SIP connection closed; reconnecting in %s s", backoff_secs) + except (OSError, ssl.SSLError) as exc: + logger.warning( + "SIP connection failed (%s); retrying in %s s", exc, backoff_secs + ) + await asyncio.sleep(backoff_secs) + backoff_secs = min(backoff_secs * 2, 60) @sip.command() diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index e9fdb1b..3428309 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -9,6 +9,7 @@ import asyncio import collections import dataclasses +import datetime import hashlib import ipaddress import json @@ -134,6 +135,9 @@ def call_received(self, request: Request) -> None: ALLOW: RFC 3261 §11 – methods supported by this UA (used in Allow header). + Args: + keepalive_interval: Keep-alive ping interval. Should be between 30 and 90 seconds. + """ #: RFC 3261 §8.1.1.7 Via branch magic cookie (indicates RFC 3261 compliance). @@ -155,10 +159,14 @@ def call_received(self, request: Request) -> None: init=False, default=None ) _initialize_task: asyncio.Task | None = dataclasses.field(init=False, default=None) + _keepalive_task: asyncio.Task | None = dataclasses.field(init=False, default=None) _call_rtp_addrs: dict[str, tuple[str, int] | None] = dataclasses.field( init=False, default_factory=dict ) _buffer: bytearray = dataclasses.field(init=False, default_factory=bytearray) + disconnected_event: asyncio.Event = dataclasses.field( + init=False, default_factory=asyncio.Event + ) #: RFC 3261 §8.1.2 — outbound SIP proxy address ``(host, port)``. #: When ``None`` the caller connects directly to the registrar server. #: The address may differ from the registrar domain derived from @@ -171,6 +179,7 @@ def call_received(self, request: Request) -> None: password: str | None = None #: STUN server used for RTP NAT traversal (SIP uses TLS/TCP; no STUN needed). rtp_stun_server_address: tuple[str, int] | None = ("stun.cloudflare.com", 3478) + keepalive_interval: datetime.timedelta = datetime.timedelta(seconds=30) call_id: str = dataclasses.field(init=False) cseq: int = dataclasses.field(init=False, default=0) #: Local TCP socket address (host, port) — set when connection is established. @@ -194,9 +203,9 @@ def connection_made(self, transport: asyncio.Transport) -> None: # type: ignore self.local_address = (ipaddress.ip_address(host), port) self._is_tls = transport.get_extra_info("ssl_object") is not None try: - self._initialize_task = asyncio.get_running_loop().create_task( - self._initialize() - ) + loop = asyncio.get_running_loop() + self._initialize_task = loop.create_task(self._initialize()) + self._keepalive_task = loop.create_task(self._run_keepalive()) except RuntimeError: pass # no running loop in synchronous test setups @@ -215,6 +224,14 @@ async def _initialize(self) -> None: ) await self.register() + async def _run_keepalive(self) -> None: + while True: + await asyncio.sleep(self.keepalive_interval.total_seconds()) + if self.transport is None: + return + logger.debug("Sending RFC 5626 §4.4.1 keep-alive ping") + self.transport.write(b"\r\n\r\n") + def data_received(self, data: bytes) -> None: self._buffer.extend(data) while True: @@ -694,7 +711,7 @@ 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: + def _build_contact(self, user: str | None = None, *, ob: bool = False) -> str: """Return a ``Contact:`` header value for this UA. The URI scheme mirrors `aor`: a ``sips:`` AOR produces a @@ -702,18 +719,27 @@ def _build_contact(self, user: str | None = None) -> str: TLS produces ``sip:`` with ``transport=tls``; plain TCP produces plain ``sip:``. + When *ob* is ``True`` the ``ob`` URI parameter ([RFC 5626 §5]) is + appended inside the angle brackets to advertise outbound keep-alive + support to the registrar. + + [RFC 5626 §5]: https://datatracker.ietf.org/doc/html/rfc5626#section-5 + Args: user: SIP user part (e.g. ``"alice"``). When provided the Contact is of the form ````; otherwise just ````. + ob: Include the ``ob`` URI parameter (RFC 5626 §5) to indicate + outbound keep-alive support. """ aor_scheme = self.aor.partition(":")[0] # "sip" or "sips" host_port = f"{_format_host(self.local_address[0])}:{self.local_address[1]}" addr = f"{user}@{host_port}" if user else host_port + ob_uri_param = ";ob" if ob else "" if aor_scheme == "sips": - return f"" + return f"" tls_param = ";transport=tls" if self._is_tls else "" - return f"" + return f"" def ringing(self, request: Request) -> None: """Send a 180 Ringing provisional response to the caller. @@ -861,9 +887,10 @@ async def register( "To": self.aor, "Call-ID": self.call_id, "CSeq": f"{self.cseq} REGISTER", - "Contact": self._build_contact(user), + "Contact": self._build_contact(user, ob=True), "Expires": "3600", # 1 hour "Max-Forwards": "70", + "Supported": "outbound", # RFC 5626 §5 — outbound keep-alive support } if authorization is not None: headers["Authorization"] = authorization @@ -945,7 +972,11 @@ def connection_lost(self, exc: Exception | None) -> None: """Handle a lost TLS/TCP connection.""" if exc is not None: logger.exception("Connection lost", exc_info=exc) + if self._keepalive_task is not None: + self._keepalive_task.cancel() + self._keepalive_task = None self.transport = None + self.disconnected_event.set() #: Short alias for `SessionInitiationProtocol`.