diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 0000000..42061c0 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +README.md \ No newline at end of file diff --git a/README.md b/README.md index 653300b..709a5e6 100644 --- a/README.md +++ b/README.md @@ -30,20 +30,21 @@ Check your ISP's documentation or router for details. You will need a SIP AOR (URI), which looks like this: ```INI -sip:USER:PASSWORD@SIP_SERVER;transport=TCP +sip:USER:PASSWORD@SIP_SERVER ``` > [!NOTE] -> This library uses secure defaults (TLS transport on port 5061). -> However, most SIP servers only support unencrypted connections. -> Therefore, you will need to provide an explict transport parameter. +> This library defaults to **UDP transport on port 5060** for `sip:` URIs, which +> is the most widely supported configuration. To use TCP or TLS, add an explicit +> `transport` parameter, e.g. `sip:user@host;transport=TCP` or +> `sips:user@host` (SIPS always uses TLS on port 5061). ### CLI A simple echo call can be started with: ```console -uvx 'voip[cli]' sip sips:alice:********@sip.example.com echo +uvx 'voip[cli]' sip sip:alice:********@sip.example.com echo ``` Each command supports an optional `--dial` argument to initiate an @@ -52,13 +53,13 @@ outbound call instead of waiting for an inbound one. To dial a number, say a message, and hang up automatically: ```console -uvx 'voip[cli]' sip sips:alice:********@sip.example.com say sip:+15551234567@sip.example.com "Your package has arrived." +uvx 'voip[cli]' sip sip:alice:********@sip.example.com say sip:+15551234567@sip.example.com "Your package has arrived." ``` You can also talk to a local agent (needs [Ollama]): ```console -uvx 'voip[cli]' sip sips:alice:********@sip.example.com agent --initial-prompt "Hi, I am looking for a Mr. Ron, first name Mo?" +uvx 'voip[cli]' sip sip:alice:********@sip.example.com agent --initial-prompt "Hi, I am looking for a Mr. Ron, first name Mo?" ``` ### MCP @@ -77,7 +78,7 @@ that exposes tools to make phone calls on your behalf to any MCP client. "mcp" ], "env": { - "SIP_AOR": "sip:****:****@example.com:5060?transport=tcp" + "SIP_AOR": "sip:number:password@example.com" } } } @@ -96,7 +97,7 @@ Pass it as `session_class` when answering an incoming call: ```python import asyncio import dataclasses -import ssl + from voip.ai import TranscribeCall from voip.sip.protocol import SIP from voip.sip.types import SipURI @@ -126,16 +127,14 @@ async def main(): RealtimeTransportProtocol, local_addr=("0.0.0.0", 0), ) - ssl_context = ssl.create_default_context() await loop.create_connection( lambda: SIP( rtp=rtp_protocol, - aor=SipURI.parse("sips:alice:********@example.com"), + aor=SipURI.parse("sip:alice:********@example.com"), transaction_class=TranscribeInviteTransaction, ), host="sip.example.com", - port=5061, - ssl=ssl_context, + port=5060, ) await asyncio.Future() diff --git a/docs/cookbook.md b/docs/cookbook.md index 0aed0cc..be33385 100644 --- a/docs/cookbook.md +++ b/docs/cookbook.md @@ -14,7 +14,6 @@ handle each utterance after a silence gap: ```python import asyncio -import ssl from voip.ai import TranscribeCall from voip.sip.dialog import Dialog @@ -40,12 +39,11 @@ async def main(): loop = asyncio.get_running_loop() await loop.create_connection( lambda: SIP( - aor="sips:alice@example.com", + aor="sip:alice@example.com", dialog_class=AutoAcceptDialog, ), host="sip.example.com", - port=5061, - ssl=ssl.create_default_context(), + port=5060, ) await asyncio.Future() @@ -80,7 +78,6 @@ Share both heavy models across calls to avoid reloading them per call: ```python import asyncio -import ssl from pocket_tts import TTSModel @@ -112,11 +109,10 @@ async def main(): loop = asyncio.get_running_loop() await loop.create_connection( lambda: MySession( - aor="sips:alice@example.com", username="alice", password="secret" + aor="sip:alice@example.com", username="alice", password="secret" ), host="sip.example.com", - port=5061, - ssl=ssl.create_default_context(), + port=5060, ) await asyncio.Future() @@ -193,10 +189,10 @@ domain. Pass `outbound_proxy` to route all signalling through it: from voip.sip.protocol import SIP session = SIP( - aor="sips:alice@carrier.com", + aor="sip:alice@carrier.com", username="alice", password="secret", - outbound_proxy=("proxy.carrier.com", 5061), + outbound_proxy=("proxy.carrier.com", 5060), ) ``` @@ -209,7 +205,7 @@ discovery round-trip by setting `rtp_stun_server_address=None`: from voip.sip.protocol import SIP session = SIP( - aor="sips:alice@example.com", + aor="sip:alice@example.com", username="alice", password="secret", rtp_stun_server_address=None, @@ -230,7 +226,6 @@ the call class when you want to terminate: ```python import asyncio -import ssl import numpy as np @@ -263,13 +258,12 @@ async def main(): loop = asyncio.get_running_loop() await loop.create_connection( lambda: MySession( - aor="sips:alice@example.com", + aor="sip:alice@example.com", username="alice", password="secret", ), host="sip.example.com", - port=5061, - ssl=ssl.create_default_context(), + port=5060, ) await asyncio.Future() @@ -292,7 +286,6 @@ Create a [Dialog][voip.sip.Dialog] subclass, set it as ```python import asyncio -import ssl from voip.audio import AudioCall from voip.sip import SipURI @@ -327,13 +320,12 @@ async def main(): loop = asyncio.get_running_loop() await loop.create_connection( lambda: MySession( - aor="sips:alice@carrier.com", + aor="sip:alice@carrier.com", username="alice", password="secret", ), host="sip.carrier.com", - port=5061, - ssl=ssl.create_default_context(), + port=5060, ) await asyncio.Future() diff --git a/docs/feature_roadmap.md b/docs/feature_roadmap.md index f84a25c..8c1073c 100644 --- a/docs/feature_roadmap.md +++ b/docs/feature_roadmap.md @@ -4,7 +4,7 @@ ### SIP Signalling -SIP User Agent Client (UAC) over TLS/TCP ([RFC 3261]). Handles incoming +SIP User Agent Client (UAC) over TLS/TCP/UDP ([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]). diff --git a/docs/rfc_status.md b/docs/rfc_status.md index d2b1dea..17ecbb0 100644 --- a/docs/rfc_status.md +++ b/docs/rfc_status.md @@ -4,7 +4,7 @@ | 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 3261](https://datatracker.ietf.org/doc/html/rfc3261) | SIP: Session Initiation Protocol | Partial | UAC only; REGISTER, INVITE, BYE, and digest authentication over TLS/TCP/UDP | | [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 | diff --git a/docs/sip.md b/docs/sip.md index fdb3b27..cea2dc3 100644 --- a/docs/sip.md +++ b/docs/sip.md @@ -15,7 +15,9 @@ ::: voip.sip.SessionInitiationProtocol options: heading_level: 2 - members: false + members: + - run + - serve ## Types diff --git a/tests/codecs/test_opus.py b/tests/codecs/test_opus.py index ca67dc7..67958e1 100644 --- a/tests/codecs/test_opus.py +++ b/tests/codecs/test_opus.py @@ -7,7 +7,7 @@ np = pytest.importorskip("numpy") av = pytest.importorskip("av") -from voip.codecs.opus import Opus # noqa: E402 +from voip.codecs.opus import Opus, OpusDecoder # noqa: E402 class TestOggCRC32: @@ -92,6 +92,18 @@ def test_decode__real_decode_returns_float32(self): result = Opus.decode(sample, 16000) assert result.dtype == np.float32 + def test_decode__real_decode_not_empty(self): + """Decode produces non-empty audio for a non-empty Opus packet. + + Regression test: a too-large OpusHead pre-skip combined with a zero + granule position previously discarded all decoded samples, yielding + an empty array and silent calls. + """ + rng = np.random.default_rng(0) + sample = Opus.encode(rng.uniform(-0.3, 0.3, 960).astype(np.float32)) + result = Opus.decode(sample, 16000) + assert result.size > 0 + class TestOpusEncode: def test_encode__returns_bytes(self): @@ -100,15 +112,57 @@ def test_encode__returns_bytes(self): assert isinstance(result, bytes) assert len(result) > 0 - def test_encode__uses_libopus_codec(self): - """Encode delegates to encode_pcm with libopus codec name.""" - with patch.object(Opus, "encode_pcm", return_value=b"encoded") as mock_enc: - Opus.encode(np.zeros(960, dtype=np.float32)) - mock_enc.assert_called_once_with( - pytest.approx(np.zeros(960, dtype=np.float32)), - "libopus", - Opus.sample_rate_hz, - ) + def test_encode__produces_single_opus_frame(self): + """Encode produces exactly one Code-0 Opus frame per 960-sample chunk. + + Regression test: the previous implementation concatenated two raw Opus + frames (one from `codec.encode(frame)` and one from the flush + `codec.encode(None)`) into a single RTP payload. A remote decoder + receiving such a payload sees Code-0 (single frame) in the TOC byte + and tries to decode the entire concatenated blob as one frame, which is + malformed — causing silence on outbound Opus echo calls. + """ + rng = np.random.default_rng(0) + result = Opus.encode(rng.uniform(-0.3, 0.3, 960).astype(np.float32)) + # Code-0 single-frame payload: TOC byte only, rest is frame data. + # Verify size is consistent with a single 20 ms Opus frame (not double). + assert (result[0] & 0x03) == 0 # TOC code bits: 0 = single frame + # A correctly encoded single 20 ms Opus frame is well under 1200 bytes. + # Two concatenated frames from the old code would be ~500+ bytes for noise. + # Silence is highly compressed; noise at 0.3 amplitude is a better bound. + assert len(result) < 1200 + + +class TestOpusPacketize: + def test_packetize__yields_single_frame_packets(self): + """Packetize yields only Code-0 (single-frame) Opus packets.""" + rng = np.random.default_rng(0) + audio = rng.uniform(-0.3, 0.3, 48000).astype(np.float32) + for pkt in Opus.packetize(audio): + assert (pkt[0] & 0x03) == 0 + + def test_packetize__frame_count(self): + """Packetize yields exactly one packet per 20 ms frame, no flush packet. + + Regression test: the previous implementation appended a flush packet + (`codec.encode(None)`) after all frames, producing N+1 RTP packets + for N frames of audio. `_dispatch_next_packet` sends every yielded + payload at a fixed 20 ms interval, so the extra packet shifted the + receiver's playback timeline by one ptime (20 ms), causing audible + timing glitches. + """ + # 5 full frames of 960 samples each → exactly 5 packets, no flush + audio = np.zeros(4800, dtype=np.float32) + assert len(list(Opus.packetize(audio))) == 5 + + def test_packetize__pads_partial_final_frame(self): + """Packetize zero-pads a partial last frame to a full 960-sample frame.""" + # 5 full frames + 100 extra samples → 6 frames (5 full + 1 padded), no flush + audio = np.zeros(4900, dtype=np.float32) + packets = list(Opus.packetize(audio)) + assert len(packets) == 6 # 6 frames (5 full + 1 padded), no flush + for pkt in packets: + assert (pkt[0] & 0x03) == 0 class TestOpusConstants: @@ -135,3 +189,59 @@ def test_frame_size(self): def test_timestamp_increment(self): """Opus timestamp increment is 960 ticks per frame.""" assert Opus.timestamp_increment == 960 + + +class TestOpusCreateDecoder: + def test_create_decoder__returns_opus_decoder(self): + """create_decoder returns an OpusDecoder instance.""" + decoder = Opus.create_decoder(16000) + assert isinstance(decoder, OpusDecoder) + + def test_create_decoder__ignores_input_rate_hz(self): + """create_decoder ignores input_rate_hz for API consistency.""" + decoder = Opus.create_decoder(16000, input_rate_hz=8000) + assert isinstance(decoder, OpusDecoder) + assert decoder.output_rate_hz == 16000 + + +class TestOpusDecoderDecode: + def test_decode__returns_float32(self): + """OpusDecoder.decode produces a float32 array.""" + decoder = Opus.create_decoder(16000) + payload = Opus.encode(np.zeros(960, dtype=np.float32)) + result = decoder.decode(payload) + assert result.dtype == np.float32 + + def test_decode__non_empty_for_real_packet(self): + """OpusDecoder.decode produces non-empty audio for a real Opus packet.""" + rng = np.random.default_rng(0) + decoder = Opus.create_decoder(16000) + payload = Opus.encode(rng.uniform(-0.3, 0.3, 960).astype(np.float32)) + result = decoder.decode(payload) + assert result.size > 0 + + def test_decode__preserves_state_across_packets(self): + """OpusDecoder.decode produces consistent per-packet output for sequential packets. + + Regression test: the previous per-packet Ogg-container decode reset the + `libopus` CELT MDCT overlap window every 20 ms, producing 50 Hz + window-boundary discontinuities heard as choppiness on echo calls. + A persistent decoder context preserves overlap state, so packets after + the first warm-up packet each produce exactly `frame_size / 3` samples + at the 16 kHz output rate. + """ + rng = np.random.default_rng(42) + decoder = Opus.create_decoder(16000) + counts = [] + for _ in range(10): + payload = Opus.encode(rng.uniform(-0.3, 0.3, 960).astype(np.float32)) + result = decoder.decode(payload) + counts.append(len(result)) + # After the first warm-up packet all packets must produce 320 samples. + assert all(c == 320 for c in counts[1:]), f"Inconsistent counts: {counts}" + + def test_decode__empty_payload_returns_empty(self): + """OpusDecoder.decode returns an empty array for an empty payload.""" + decoder = Opus.create_decoder(16000) + result = decoder.decode(b"") + assert result.size == 0 diff --git a/tests/sip/conftest.py b/tests/sip/conftest.py index 8423d69..ac1c1d2 100644 --- a/tests/sip/conftest.py +++ b/tests/sip/conftest.py @@ -1,5 +1,6 @@ """Shared fixtures for SIP tests.""" +import asyncio import dataclasses import ipaddress @@ -64,10 +65,13 @@ def fake_transport() -> FakeTransport: @pytest.fixture -def rtp() -> RealtimeTransportProtocol: - """Return a RealtimeTransportProtocol with a pre-set public address.""" +async def rtp() -> RealtimeTransportProtocol: + """Return a RealtimeTransportProtocol with a pre-resolved public address.""" mux = RealtimeTransportProtocol() - mux.public_address = NetworkAddress(ipaddress.ip_address("192.0.2.1"), 5004) + mux.public_address = asyncio.get_running_loop().create_future() + mux.public_address.set_result( + NetworkAddress(ipaddress.ip_address("192.0.2.1"), 5004) + ) return mux diff --git a/tests/sip/test_messages.py b/tests/sip/test_messages.py index 32e795d..da81e27 100644 --- a/tests/sip/test_messages.py +++ b/tests/sip/test_messages.py @@ -290,8 +290,8 @@ def test___bytes____with_sdp_body__auto_content_length(self): parsed = messages.Message.parse(serialized) assert parsed.body is None - def test_from_request__with_dialog_remote_tag(self): - """Include dialog remote_tag in To header when dialog has a remote_tag.""" + def test_from_request__with_dialog_local_tag(self): + """Include dialog local_tag in To header when dialog has a remote_tag.""" data = ( b"INVITE sip:bob@biloxi.com SIP/2.0\r\n" b"Via: SIP/2.0/UDP pc33.atlanta.com;branch=z9hG4bKabc\r\n" @@ -304,7 +304,7 @@ def test_from_request__with_dialog_remote_tag(self): request = messages.Message.parse(data) dialog = Dialog( uac=SipURI.parse("sip:alice@atlanta.com"), - remote_tag="server-tag", + local_tag="server-tag", ) response = messages.Response.from_request( request, dialog=dialog, status_code=200, phrase="OK" diff --git a/tests/sip/test_types.py b/tests/sip/test_types.py index f996c20..4603182 100644 --- a/tests/sip/test_types.py +++ b/tests/sip/test_types.py @@ -295,10 +295,10 @@ def test_transport__sips_returns_tls(self): uri = SipURI.parse("sips:alice@example.com") assert uri.transport == "TLS" - def test_transport__sip_without_parameter_returns_none(self): - """Return None for a plain sip: URI without transport parameter.""" + def test_transport__sip_without_parameter_returns_udp(self): + """Return 'UDP' for a plain sip: URI without transport parameter.""" uri = SipURI.parse("sip:alice@example.com") - assert uri.transport == "TLS" + assert uri.transport == "UDP" def test_transport__explicit_parameter(self): """Return explicit transport parameter value.""" diff --git a/tests/test_mcp.py b/tests/test_mcp.py index 0294297..d8de9e1 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -392,7 +392,6 @@ async def test_run__sets_connection_pool_sip(self) -> None: aor = SipURI.parse("sip:alice@example.com") mock_protocol = MagicMock(spec=SessionInitiationProtocol) - fn = MagicMock() with patch.object( SessionInitiationProtocol, "run", @@ -400,7 +399,7 @@ async def test_run__sets_connection_pool_sip(self) -> None: return_value=mock_protocol, ): with patch.object(voip.mcp.mcp, "run_async", new_callable=AsyncMock): - await run(fn, aor) + await run(aor) assert connection_pool.sip is mock_protocol @@ -418,7 +417,7 @@ async def test_run__calls_mcp_run_async_with_transport(self) -> None: with patch.object( voip.mcp.mcp, "run_async", new_callable=AsyncMock ) as mock_run: - await run(lambda: None, aor, transport="stdio") + await run(aor, transport="stdio") mock_run.assert_awaited_once_with(transport="stdio") @@ -434,7 +433,7 @@ async def test_run__passes_no_verify_tls(self) -> None: return_value=mock_protocol, ) as mock_sip_run: with patch.object(voip.mcp.mcp, "run_async", new_callable=AsyncMock): - await run(lambda: None, aor, no_verify_tls=True) + await run(aor, no_verify_tls=True) _, kwargs = mock_sip_run.call_args assert kwargs["no_verify_tls"] is True @@ -452,7 +451,7 @@ async def test_run__passes_stun_server(self) -> None: return_value=mock_protocol, ) as mock_sip_run: with patch.object(voip.mcp.mcp, "run_async", new_callable=AsyncMock): - await run(lambda: None, aor, stun_server=stun) + await run(aor, stun_server=stun) _, kwargs = mock_sip_run.call_args assert kwargs["stun_server"] is stun @@ -468,22 +467,7 @@ def test_registered_event__set_by_on_registered(self) -> None: """on_registered() sets registered_event so run() can unblock.""" protocol = SessionInitiationProtocol.__new__(SessionInitiationProtocol) protocol.registered_event = asyncio.Event() - protocol.ready_callback = None assert not protocol.registered_event.is_set() protocol.on_registered() assert protocol.registered_event.is_set() - - def test_registered_event__ready_callback_called_after_event(self) -> None: - """ready_callback is invoked after registered_event is set.""" - call_order: list[str] = [] - protocol = SessionInitiationProtocol.__new__(SessionInitiationProtocol) - protocol.registered_event = asyncio.Event() - - def _cb() -> None: - call_order.append("cb" if protocol.registered_event.is_set() else "early") - - protocol.ready_callback = _cb - protocol.on_registered() - - assert call_order == ["cb"] diff --git a/voip/__main__.py b/voip/__main__.py index 7813899..e6cdcd1 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -1,15 +1,11 @@ #!/usr/bin/env python3 import asyncio -import collections.abc import dataclasses -import ipaddress import logging import socket -import ssl import time from voip.ai import SayCall -from voip.rtp import RealtimeTransportProtocol, Session from voip.sip import dialog, messages from voip.sip.protocol import SessionInitiationProtocol from voip.sip.types import SipURI, parse_uri @@ -39,6 +35,16 @@ def _parse_sip_uri(ctx, param, value) -> SipURI: raise click.BadParameter(str(e)) from e +@dataclasses.dataclass(kw_only=True, slots=True) +class OutboundDialog(dialog.Dialog): + """A dialog that closes the SIP connection when the remote party hangs up.""" + + def hangup_received(self) -> None: + """Close the SIP connection so the process can exit cleanly.""" + if self.sip is not None: + self.sip.close() + + @dataclasses.dataclass(kw_only=True, slots=True) class ConsoleMessageProtocol(SessionInitiationProtocol): """Pretty print SIP messages to stdout using pygments.""" @@ -135,7 +141,6 @@ def mcp(aor: SipURI, stun_server: NetworkAddress, no_verify_tls: bool, transport asyncio.run( run( - lambda: None, aor, stun_server=stun_server, no_verify_tls=no_verify_tls, @@ -179,114 +184,6 @@ def sip(ctx, aor, stun_server, no_verify_tls): ) -async def _connect_rtp( - proxy_addr: NetworkAddress, - rtp_stun_server_address: NetworkAddress | None, -) -> tuple[asyncio.DatagramTransport, RealtimeTransportProtocol]: - loop = asyncio.get_running_loop() - rtp_bind = ( - "::" if isinstance(proxy_addr[0], ipaddress.IPv6Address) else "0.0.0.0" # noqa: S104 - ) - return await loop.create_datagram_endpoint( - lambda: RealtimeTransportProtocol(stun_server_address=rtp_stun_server_address), - local_addr=(rtp_bind, 0), - ) - - -async def _connect_sip( - session_factory, - proxy_addr: NetworkAddress, - use_tls: bool, - no_verify_tls: bool, -) -> None: - loop = asyncio.get_running_loop() - ssl_context: ssl.SSLContext | None = None - if use_tls: - ssl_context = ssl.create_default_context() - if no_verify_tls: - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - 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) - - -async def _connect_sip_once( - session_factory: collections.abc.Callable[[], SessionInitiationProtocol], - proxy_addr: NetworkAddress, - use_tls: bool, - no_verify_tls: bool, -) -> None: - loop = asyncio.get_running_loop() - ssl_context: ssl.SSLContext | None = None - if use_tls: - ssl_context = ssl.create_default_context() - if no_verify_tls: - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - _, protocol = await loop.create_connection( - session_factory, - host=str(proxy_addr[0]), - port=proxy_addr[1], - ssl=ssl_context, - ) - await protocol.disconnected_event.wait() - - -def _make_outbound_factory( - *, - verbose: int, - aor: SipURI, - rtp_protocol: RealtimeTransportProtocol, - target_uri: SipURI, - session_class: type[Session], - session_kwargs: dict, -) -> collections.abc.Callable[[], ConsoleMessageProtocol]: - - class OutboundDialog(dialog.Dialog): - def hangup_received(self) -> None: - if self.sip is not None: - self.sip.close() - - @dataclasses.dataclass(kw_only=True, slots=True) - class OutboundProtocol(ConsoleMessageProtocol): - dial_target: SipURI - - def on_registered(self) -> None: - dialog = OutboundDialog(sip=self) - asyncio.create_task( - dialog.dial( - self.dial_target, session_class=session_class, **session_kwargs - ) - ) - - def factory() -> ConsoleMessageProtocol: - return OutboundProtocol( - verbose=verbose, - dialog_class=OutboundDialog, - aor=aor, - rtp=rtp_protocol, - dial_target=target_uri, - ) - - return factory - - @sip.command() @click.option( "--dial", @@ -308,36 +205,26 @@ def call_received(self) -> None: self.answer(session_class=EchoCall) async def run(): - _, rtp_protocol = await _connect_rtp( - aor.maddr, - obj["stun_server"], - ) if dial is None: - await _connect_sip( - lambda: ConsoleMessageProtocol( - verbose=obj.get("verbose", 0), - dialog_class=EchoDialog, - aor=aor, - rtp=rtp_protocol, - ), - aor.maddr, - aor.transport == "TLS", - obj["no_verify_tls"], + await ConsoleMessageProtocol.serve( + aor, + EchoDialog, + verbose=obj.get("verbose", 0), + no_verify_tls=obj["no_verify_tls"], + stun_server=obj["stun_server"], ) else: - await _connect_sip_once( - _make_outbound_factory( - verbose=obj.get("verbose", 0), - aor=aor, - rtp_protocol=rtp_protocol, - target_uri=parse_uri(dial, aor), - session_class=EchoCall, - session_kwargs={}, - ), - aor.maddr, - aor.transport == "TLS", - obj["no_verify_tls"], + protocol = await ConsoleMessageProtocol.run( + aor, + OutboundDialog, + verbose=obj.get("verbose", 0), + no_verify_tls=obj["no_verify_tls"], + stun_server=obj["stun_server"], + ) + await OutboundDialog(sip=protocol).dial( + parse_uri(dial, aor), session_class=EchoCall ) + await protocol.disconnected_event.wait() try: asyncio.run(run()) @@ -385,36 +272,28 @@ def call_received(self) -> None: ) async def run(): - _, rtp_protocol = await _connect_rtp( - aor.maddr, - obj["stun_server"], - ) if dial is None: - await _connect_sip( - lambda: ConsoleMessageProtocol( - verbose=obj.get("verbose", 0), - dialog_class=TranscribeDialog, - aor=aor, - rtp=rtp_protocol, - ), - aor.maddr, - aor.transport == "TLS", - obj["no_verify_tls"], + await ConsoleMessageProtocol.serve( + aor, + TranscribeDialog, + verbose=obj.get("verbose", 0), + no_verify_tls=obj["no_verify_tls"], + stun_server=obj["stun_server"], ) else: - await _connect_sip_once( - _make_outbound_factory( - verbose=obj.get("verbose", 0), - aor=aor, - rtp_protocol=rtp_protocol, - target_uri=parse_uri(dial, aor), - session_class=TranscribingCall, - session_kwargs={"stt_model": WhisperModel(stt_model)}, - ), - aor.maddr, - aor.transport == "TLS", - obj["no_verify_tls"], + protocol = await ConsoleMessageProtocol.run( + aor, + OutboundDialog, + verbose=obj.get("verbose", 0), + no_verify_tls=obj["no_verify_tls"], + stun_server=obj["stun_server"], + ) + await OutboundDialog(sip=protocol).dial( + parse_uri(dial, aor), + session_class=TranscribingCall, + stt_model=WhisperModel(stt_model), ) + await protocol.disconnected_event.wait() try: asyncio.run(run()) @@ -519,42 +398,32 @@ def call_received(self) -> None: ) async def run(): - _, rtp_protocol = await _connect_rtp( - aor.maddr, - obj["stun_server"], - ) if dial is None: - await _connect_sip( - lambda: ConsoleMessageProtocol( - verbose=obj.get("verbose", 0), - dialog_class=AgentDialog, - aor=aor, - rtp=rtp_protocol, - ), - aor.maddr, - aor.transport == "TLS", - obj["no_verify_tls"], + await ConsoleMessageProtocol.serve( + aor, + AgentDialog, + verbose=obj.get("verbose", 0), + no_verify_tls=obj["no_verify_tls"], + stun_server=obj["stun_server"], ) else: - await _connect_sip_once( - _make_outbound_factory( - verbose=obj.get("verbose", 0), - aor=aor, - rtp_protocol=rtp_protocol, - target_uri=parse_uri(dial, aor), - session_class=AgentCallWithOutput, - session_kwargs={ - "stt_model": WhisperModel(stt_model), - "llm_model": llm_model, - "voice": voice, - "system_prompt": system_prompt, - "salutation": salutation, - }, - ), - aor.maddr, - aor.transport == "TLS", - obj["no_verify_tls"], + protocol = await ConsoleMessageProtocol.run( + aor, + OutboundDialog, + verbose=obj.get("verbose", 0), + no_verify_tls=obj["no_verify_tls"], + stun_server=obj["stun_server"], + ) + await OutboundDialog(sip=protocol).dial( + parse_uri(dial, aor), + session_class=AgentCallWithOutput, + stt_model=WhisperModel(stt_model), + llm_model=llm_model, + voice=voice, + system_prompt=system_prompt, + salutation=salutation, ) + await protocol.disconnected_event.wait() try: asyncio.run(run()) @@ -579,23 +448,20 @@ def say(ctx, target: str, prompt: str, voice: str): aor = obj["aor"] async def run(): - _, rtp_protocol = await _connect_rtp( - aor.maddr, - obj["stun_server"], + protocol = await ConsoleMessageProtocol.run( + aor, + OutboundDialog, + verbose=obj.get("verbose", 0), + no_verify_tls=obj["no_verify_tls"], + stun_server=obj["stun_server"], ) - await _connect_sip_once( - _make_outbound_factory( - verbose=obj.get("verbose", 0), - aor=aor, - rtp_protocol=rtp_protocol, - target_uri=parse_uri(target, aor), - session_class=SayCall, - session_kwargs={"text": prompt, "voice": voice}, - ), - aor.maddr, - aor.transport == "TLS", - obj["no_verify_tls"], + await OutboundDialog(sip=protocol).dial( + parse_uri(target, aor), + session_class=SayCall, + text=prompt, + voice=voice, ) + await protocol.disconnected_event.wait() try: asyncio.run(run()) diff --git a/voip/ai.py b/voip/ai.py index b0857de..dd900b7 100644 --- a/voip/ai.py +++ b/voip/ai.py @@ -4,7 +4,7 @@ with faster-whisper, and [AgentCall][voip.ai.AgentCall], which extends it with an Ollama-powered response loop and Pocket TTS voice synthesis. -Requires the ``ai`` extra: ``pip install voip[ai]``. +Requires the `ai` extra: `pip install voip[ai]`. """ import asyncio diff --git a/voip/audio.py b/voip/audio.py index 8c39d1e..630e11f 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -4,9 +4,9 @@ packets, negotiates codecs, and decodes/encodes audio using the codec implementations in [voip.codecs][voip.codecs]. -Requires the ``audio`` extra: ``pip install voip[audio]``. +Requires the `audio` extra: `pip install voip[audio]`. AI-powered subclasses (Whisper transcription, Ollama agent) live in -[voip.ai][voip.ai] and require the ``ai`` extra. +[voip.ai][voip.ai] and require the `ai` extra. """ import asyncio @@ -86,6 +86,14 @@ def __post_init__(self) -> None: if fmt.encoding_name is None: raise ValueError(f"No encoding name for payload type {fmt.payload_type}") self.codec = codecs.get(fmt.encoding_name) + if fmt.payload_type != self.codec.payload_type: + logger.warning( + "negotiated payload type %d differs from codec default %d for %s " + "(dynamic PT assignment by remote SDP)", + fmt.payload_type, + self.codec.payload_type, + self.codec.encoding_name, + ) self.payload_decoder = self.codec.create_decoder( self.sampling_rate_hz, input_rate_hz=self.sample_rate ) @@ -93,7 +101,7 @@ def __post_init__(self) -> None: @property def payload_type(self) -> int: """Negotiated RTP payload type number.""" - return self.codec.payload_type + return self.media.fmt[0].payload_type @property def sample_rate(self) -> int: @@ -161,13 +169,18 @@ async def emit_audio(self, packet: RTPPacket) -> None: audio = self.decode_payload(packet.payload) if audio.size > 0: self.audio_received(audio=audio, rms=self.rms(audio)) + elif packet.payload: + logger.warning( + "Decoded audio is empty for non-empty RTP payload (size %d bytes)", + len(packet.payload), + ) def decode_payload(self, payload: bytes) -> np.ndarray: return self.payload_decoder.decode(payload) def next_rtp_packet(self, payload: bytes) -> RTPPacket: packet = RTPPacket( - payload_type=self.codec.payload_type, + payload_type=self.payload_type, sequence_number=self.rtp_sequence_number, timestamp=self.rtp_timestamp, ssrc=self.rtp_ssrc, @@ -209,7 +222,7 @@ def on_audio_sent(self) -> None: """Handle completion of an outbound audio stream. Called once the last RTP packet of an outbound stream has been - dispatched (i.e. `outbound_handle` transitions to ``None``). + dispatched (i.e. `outbound_handle` transitions to `None`). The base implementation is a no-op. Override in subclasses to trigger post-audio actions, for example hanging up after [SayCall][voip.ai.SayCall] finishes speaking. @@ -246,18 +259,16 @@ async def send_audio(self, audio: np.ndarray) -> None: Args: audio: Float32 mono PCM at `codec.sample_rate_hz` Hz. """ - remote_addr = next( - (addr for addr, call in self.rtp.calls.items() if call is self), - None, - ) - match remote_addr: - case None: - logger.warning( - "No remote RTP address for this call; dropping audio", - ) - return - case _: - pass + if not ( + remote_addr := next( + (addr for addr, call in self.rtp.calls.items() if call is self), + None, + ) + ): + logger.warning( + "No remote RTP address for this call; dropping audio", + ) + return async with self.send_audio_lock: self.cancel_outbound_audio() loop = asyncio.get_running_loop() diff --git a/voip/codecs/__init__.py b/voip/codecs/__init__.py index 1665a08..46ae7fb 100644 --- a/voip/codecs/__init__.py +++ b/voip/codecs/__init__.py @@ -5,13 +5,13 @@ - [PCMA][voip.codecs.PCMA] — G.711 A-law (RFC 3551), PT 8 *(pure NumPy)* - [PCMU][voip.codecs.PCMU] — G.711 mu-law (RFC 3551), PT 0 *(pure NumPy)* -- [G722][voip.codecs.G722] — G.722 (RFC 3551), PT 9 *(requires* ``pyav`` *extra)* -- [Opus][voip.codecs.Opus] — Opus (RFC 7587), PT 111 *(requires* ``pyav`` *extra)* +- [G722][voip.codecs.G722] — G.722 (RFC 3551), PT 9 *(requires* `pyav` *extra)* +- [Opus][voip.codecs.Opus] — Opus (RFC 7587), PT 111 *(requires* `pyav` *extra)* Use [get][voip.codecs.get] to look up a codec class by its SDP encoding name (case-insensitive). -When the ``pyav`` extra is not installed only PCMA and PCMU are registered. +When the `pyav` extra is not installed only PCMA and PCMU are registered. """ from voip.codecs.base import RTPCodec diff --git a/voip/codecs/av.py b/voip/codecs/av.py index b0d75c9..d50c9c8 100644 --- a/voip/codecs/av.py +++ b/voip/codecs/av.py @@ -6,7 +6,7 @@ [encode_pcm][voip.codecs.av.PyAVCodec.encode_pcm] helpers that use [PyAV][] for container-aware decode and codec-aware encode. -Requires the ``pyav`` extra: ``pip install voip[pyav]``. +Requires the `pyav` extra: `pip install voip[pyav]`. Concrete subclasses: [Opus][voip.codecs.Opus], [G722][voip.codecs.G722]. diff --git a/voip/codecs/base.py b/voip/codecs/base.py index 27a5d7d..4effd25 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -176,10 +176,6 @@ def create_decoder( ) -> PayloadDecoder: """Create a stateless per-call payload decoder for this codec. - Override in subclasses that require stateful decoding across RTP - packets (e.g. G.722 ADPCM — see - [G722.create_decoder][voip.codecs.g722.G722.create_decoder]). - Args: output_rate_hz: Target PCM sample rate in Hz for decoded audio. input_rate_hz: Input clock rate override, or `None` to use the diff --git a/voip/codecs/g722.py b/voip/codecs/g722.py index f75f547..36541f3 100644 --- a/voip/codecs/g722.py +++ b/voip/codecs/g722.py @@ -8,7 +8,7 @@ stateful decoding that preserves the ADPCM predictor state across consecutive RTP packets. -Requires the ``hd-audio`` extra: ``pip install voip[hd-audio]``. +Requires the `hd-audio` extra: `pip install voip[hd-audio]`. """ import dataclasses @@ -81,26 +81,6 @@ def packetize(cls, audio: np.ndarray) -> Iterator[bytes]: def create_decoder( cls, output_rate_hz: int, *, input_rate_hz: int | None = None ) -> G722Decoder: - """Create a stateful per-call G.722 decoder. - - Returns a [G722Decoder][voip.codecs.g722.G722Decoder] that preserves - the ADPCM predictor state across consecutive RTP packets. Pass the - returned decoder to - [AudioCall][voip.audio.AudioCall] (via the `create_decoder` - factory) to avoid the per-packet state reset that causes robotic - audio artefacts. - - The *input_rate_hz* parameter is accepted for API consistency with - [RTPCodec.create_decoder][voip.codecs.base.RTPCodec.create_decoder] - but is not used; G.722 always decodes at 16 000 Hz internally. - - Args: - output_rate_hz: Target PCM sample rate in Hz for decoded audio. - input_rate_hz: Ignored. G.722 always decodes at `sample_rate_hz`. - - Returns: - A new [G722Decoder][voip.codecs.g722.G722Decoder] instance. - """ return G722Decoder(output_rate_hz) @@ -115,9 +95,6 @@ class G722Decoder: robotic artefacts when decoding a G.722 stream with independent codec contexts. - Use [G722.create_decoder][voip.codecs.g722.G722.create_decoder] rather - than instantiating this class directly. - Attributes: output_rate_hz: Target PCM sample rate in Hz for decoded audio. codec_context: Persistent G.722 decoder context diff --git a/voip/codecs/opus.py b/voip/codecs/opus.py index 1f49657..7be3433 100644 --- a/voip/codecs/opus.py +++ b/voip/codecs/opus.py @@ -4,20 +4,27 @@ minimal [Ogg][] container before passing them to PyAV for decoding, and encodes float32 PCM via `libopus`. -Requires the ``pyav`` extra: ``pip install voip[pyav]``. +Requires the `pyav` extra: `pip install voip[pyav]`. [Ogg]: https://wiki.xiph.org/Ogg """ +import dataclasses +import logging import os import struct -from typing import ClassVar +from collections.abc import Iterator +from typing import ClassVar, cast +import av +import av.audio.resampler import numpy as np from voip.codecs.av import PyAVCodec -__all__ = ["Opus"] +logger = logging.getLogger(__name__) + +__all__ = ["Opus", "OpusDecoder"] class Opus(PyAVCodec): @@ -121,8 +128,8 @@ def _ogg_container(cls, packet: bytes) -> bytes: "<8sBBHIhB", b"OpusHead", 1, # version - 1, # channel count (mono) - 3840, # pre-skip: 80 ms at 48 kHz (RFC 7587) + cls.channels, # channel count + 0, # pre-skip: each RTP payload is decoded as a standalone stream cls.sample_rate_hz, 0, # output gain 0, # channel mapping family (mono/stereo) @@ -136,7 +143,7 @@ def _ogg_container(cls, packet: bytes) -> bytes: [ cls._ogg_page(0x02, 0, serial_number, 0, [opus_head]), # BOS cls._ogg_page(0x00, 0, serial_number, 1, [opus_tags]), - cls._ogg_page(0x04, 0, serial_number, 2, [packet]), + cls._ogg_page(0x04, cls.frame_size, serial_number, 2, [packet]), ] ) @@ -152,4 +159,84 @@ def decode( @classmethod def encode(cls, samples: np.ndarray) -> bytes: - return cls.encode_pcm(samples, "libopus", cls.sample_rate_hz) + return next(cls.packetize(samples), b"") + + @classmethod + def packetize(cls, audio: np.ndarray) -> Iterator[bytes]: + codec = cast(av.AudioCodecContext, av.CodecContext.create("libopus", "w")) + codec.sample_rate = cls.sample_rate_hz + codec.format = av.AudioFormat("fltp") + codec.layout = av.AudioLayout("mono") + codec.open() + for i in range(0, len(audio), cls.frame_size): + chunk = audio[i : i + cls.frame_size].astype(np.float32) + if len(chunk) < cls.frame_size: + chunk = np.pad(chunk, (0, cls.frame_size - len(chunk))) + frame = av.AudioFrame.from_ndarray( + chunk[np.newaxis, :], format="fltp", layout="mono" + ) + frame.sample_rate = cls.sample_rate_hz + frame.pts = i + yield from (bytes(pkt) for pkt in codec.encode(frame)) + + @classmethod + def create_decoder( + cls, output_rate_hz: int, *, input_rate_hz: int | None = None + ) -> OpusDecoder: + return OpusDecoder(output_rate_hz) + + +@dataclasses.dataclass(slots=True) +class OpusDecoder: + """Stateful Opus decoder that preserves `libopus` CELT state across packets. + + Creates a single persistent + [av.CodecContext](https://pyav.basswood-io.com/docs/stable/api/codec.html#av.codec.context.CodecContext) + for the life of the decoder and feeds each incoming RTP payload directly to + the same `libopus` context. This eliminates the per-packet MDCT overlap + reset that creates 50 Hz window-boundary discontinuities — heard as + choppiness — when each packet is decoded in a fresh context. + + Use [Opus.create_decoder][voip.codecs.opus.Opus.create_decoder] rather + than instantiating this class directly. + + Attributes: + output_rate_hz: Target PCM sample rate in Hz for decoded audio. + codec_context: Persistent `libopus` decoder context shared across all + [decode][voip.codecs.opus.OpusDecoder.decode] calls. + resampler: Persistent resampler targeting `output_rate_hz` Hz. + """ + + output_rate_hz: int + codec_context: av.AudioCodecContext = dataclasses.field(init=False, repr=False) + resampler: av.audio.resampler.AudioResampler = dataclasses.field( + init=False, repr=False + ) + + def __post_init__(self) -> None: + self.codec_context = cast( + av.AudioCodecContext, av.CodecContext.create("libopus", "r") + ) + self.codec_context.sample_rate = Opus.sample_rate_hz + self.codec_context.open() + self.resampler = av.audio.resampler.AudioResampler( + format="fltp", layout="mono", rate=self.output_rate_hz + ) + + def decode(self, payload: bytes) -> np.ndarray: + """Decode one Opus RTP payload, preserving CELT state from prior packets. + + Args: + payload: Raw Opus RTP payload bytes. + + Returns: + Float32 mono PCM array at `output_rate_hz` Hz. + """ + return np.concatenate( + [ + resampled.to_ndarray().flatten() + for frame in self.codec_context.decode(av.Packet(payload)) + for resampled in self.resampler.resample(frame) + ] + or [np.array([], dtype=np.float32)] + ) diff --git a/voip/mcp.py b/voip/mcp.py index 17abbeb..cd6e677 100644 --- a/voip/mcp.py +++ b/voip/mcp.py @@ -1,6 +1,6 @@ """MCP server for VoIP actions. -Requires the ``mcp`` extra: ``pip install voip[mcp]``. +Requires the `mcp` extra: `pip install voip[mcp]`. """ import asyncio @@ -137,7 +137,6 @@ async def call( async def run( - fn: typing.Callable[[], None], aor: SipURI, *, no_verify_tls: bool = False, @@ -145,7 +144,6 @@ async def run( transport: str | None = None, ) -> None: connection_pool.sip = await SessionInitiationProtocol.run( - fn, aor, Dialog, no_verify_tls=no_verify_tls, diff --git a/voip/rtp.py b/voip/rtp.py index fdb99ad..ccd837b 100644 --- a/voip/rtp.py +++ b/voip/rtp.py @@ -96,7 +96,7 @@ class Session: The `rtp` back-reference allows sending media; the `dialog` back-reference carries the SIP dialog state and a reference to the SIP session - (``dialog.sip``) so that the transport can be closed when the call ends. + (`dialog.sip`) so that the transport can be closed when the call ends. Subclass `voip.audio.AudioCall` for audio calls with codec negotiation, buffering, and decoding. @@ -120,7 +120,7 @@ def packet_received(self, packet: RTPPacket, addr: NetworkAddress) -> None: Args: packet: Parsed RTP packet. - addr: Remote ``(host, port)`` the packet arrived from. + addr: Remote `(host, port)` the packet arrived from. """ def send_packet(self, packet: RTPPacket, addr: NetworkAddress) -> None: @@ -130,7 +130,7 @@ def send_packet(self, packet: RTPPacket, addr: NetworkAddress) -> None: Args: packet: RTP packet to send. - addr: Destination ``(host, port)``. + addr: Destination `(host, port)`. """ data = bytes(packet) if self.srtp is not None: @@ -172,7 +172,7 @@ def negotiate_codec(cls, remote_media: MediaDescription) -> MediaDescription: propagates and the call is not answered. Args: - remote_media: The SDP ``m=audio`` section from the remote INVITE. + remote_media: The SDP `m=audio` section from the remote INVITE. Returns: A `MediaDescription` with the chosen codec. @@ -213,7 +213,7 @@ class RealtimeTransportProtocol(STUNProtocol): matching handler's `datagram_received` method by remote source address. - Use ``addr=None`` in `register_call` as a wildcard catch-all for + Use `addr=None` in `register_call` as a wildcard catch-all for calls whose remote RTP address is not known in advance (no SDP in INVITE). """ @@ -221,14 +221,53 @@ class RealtimeTransportProtocol(STUNProtocol): calls: dict[tuple[str, int] | None, Session] = dataclasses.field( init=False, default_factory=dict ) - public_address: NetworkAddress | None = dataclasses.field(init=False, default=None) + public_address: asyncio.Future[NetworkAddress] = dataclasses.field( + init=False, default_factory=asyncio.Future + ) + + def connection_made(self, transport: asyncio.DatagramTransport) -> None: + """Create the public-address future, then start STUN negotiation.""" + self.public_address = asyncio.get_running_loop().create_future() + super().connection_made(transport) def stun_connection_made( self, transport: asyncio.DatagramTransport, addr: NetworkAddress, ) -> None: - self.public_address = addr + logger.debug("RTP socket ready, public address is %s:%s", addr[0], addr[1]) + if self.public_address is not None and not self.public_address.done(): + self.public_address.set_result(addr) + + @classmethod + async def serve( + cls, + bind_address: str | None = None, + stun_server: NetworkAddress | None = None, + ) -> RealtimeTransportProtocol: + """Create a bound RTP endpoint and wait for the public address. + + Creates the UDP socket, sends a STUN binding request when configured, + and suspends until the public address is confirmed before returning. + + Args: + bind_address: Local bind address — `"0.0.0.0"` for IPv4 or + `"::"` for IPv6. + stun_server: STUN server for NAT traversal. `None` skips STUN + and uses the local socket address instead. + + Returns: + A ready [`RealtimeTransportProtocol`][voip.rtp.RealtimeTransportProtocol] + with [`public_address`][voip.rtp.RealtimeTransportProtocol.public_address] + already resolved. + """ + loop = asyncio.get_running_loop() + _, rtp = await loop.create_datagram_endpoint( + lambda: cls(stun_server_address=stun_server), + local_addr=(bind_address, 0), + ) + await rtp.public_address + return rtp def register_call( self, @@ -237,13 +276,13 @@ def register_call( ) -> None: """Register *handler* for RTP traffic arriving from *addr*. - Use ``addr=None`` as a wildcard to handle traffic from any source that + Use `addr=None` as a wildcard to handle traffic from any source that has no dedicated routing entry (useful when the caller's RTP address is not known in advance from the INVITE SDP). Args: - addr: Remote ``(ip, port)`` as it will appear in incoming datagrams, - or ``None`` to register a wildcard catch-all handler. + addr: Remote `(ip, port)` as it will appear in incoming datagrams, + or `None` to register a wildcard catch-all handler. handler: A `Call` instance whose `datagram_received` will be called for matching packets. @@ -283,10 +322,10 @@ def packet_received(self, data: bytes, addr: NetworkAddress) -> None: """Route an incoming SRTP datagram to the matching per-call handler. Looks up *addr* in the call registry. Falls back to the wildcard - ``None`` handler when no exact match exists. Drops the packet with a + `None` handler when no exact match exists. Drops the packet with a debug log when no handler is registered at all. - When the matched handler carries an SRTP session the packet is + When the matched handler carries an SRTP session, the packet is authenticated and decrypted before being forwarded; packets that fail authentication are logged at WARNING level and discarded. """ diff --git a/voip/sdp/messages.py b/voip/sdp/messages.py index 57265c9..20acf84 100644 --- a/voip/sdp/messages.py +++ b/voip/sdp/messages.py @@ -110,8 +110,8 @@ def _apply_line( def _apply_media_attribute(attr: Attribute, media: MediaDescription) -> bool: """Fold a media-level a= attribute into *media* if it is a format-specific attribute. - Returns ``True`` when the attribute was consumed (``a=rtpmap`` or - ``a=fmtp``), ``False`` otherwise so the caller can fall through to the + Returns `True` when the attribute was consumed (`a=rtpmap` or + `a=fmtp`), `False` otherwise so the caller can fall through to the generic attribute list. """ return media.apply_attribute(attr) diff --git a/voip/sdp/types.py b/voip/sdp/types.py index bfba645..51c2a0f 100644 --- a/voip/sdp/types.py +++ b/voip/sdp/types.py @@ -265,11 +265,11 @@ def from_pt(cls, pt: int) -> StaticPayloadType: class RTPPayloadFormat(ByteSerializableObject): """RTP payload format descriptor (RFC 3551 §6 / RFC 4566 §6). - Codec parameters from ``a=rtpmap`` are merged in by the SDP parser. + Codec parameters from `a=rtpmap` are merged in by the SDP parser. Static payload types fall back to the `StaticPayloadType` table. - Dynamic payload types (PT ≥ 96) require an explicit ``a=rtpmap``. + Dynamic payload types (PT ≥ 96) require an explicit `a=rtpmap`. - Serialises to the ``a=rtpmap`` value when codec fields are present. + Serialises to the `a=rtpmap` value when codec fields are present. """ payload_type: int @@ -351,14 +351,14 @@ class MediaDescription(ByteSerializableObject): attributes: list[Attribute] = dataclasses.field(default_factory=list) def get_format(self, pt: int | str) -> RTPPayloadFormat | None: - """Return the `RTPPayloadFormat` for payload type *pt*, or ``None``.""" + """Return the `RTPPayloadFormat` for payload type *pt*, or `None`.""" target = int(pt) return next((f for f in self.fmt if f.payload_type == target), None) def apply_attribute(self, attr: Attribute) -> bool: - """Apply a media-level ``a=`` attribute, returning ``True`` if consumed. + """Apply a media-level `a=` attribute, returning `True` if consumed. - Handles ``a=rtpmap`` and ``a=fmtp`` by updating the matching + Handles `a=rtpmap` and `a=fmtp` by updating the matching `RTPPayloadFormat` entry. Other attributes go to `attributes`. """ if attr.name == "rtpmap" and attr.value is not None: diff --git a/voip/sip/dialog.py b/voip/sip/dialog.py index 5b2ba7c..67d9456 100644 --- a/voip/sip/dialog.py +++ b/voip/sip/dialog.py @@ -187,7 +187,7 @@ async def dial( Initiate an outbound call to *target*. Args: - target: SIP or tel URI of the remote party (e.g. ``"sip:+15551234567@carrier.com"`` or ``"tel:+15551234567"``). + target: SIP or tel URI of the remote party (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`). session_class: Session subclass to create for this call. **session_kwargs: Extra keyword arguments forwarded to `session_class`. diff --git a/voip/sip/exceptions.py b/voip/sip/exceptions.py index a78f9b5..605d309 100644 --- a/voip/sip/exceptions.py +++ b/voip/sip/exceptions.py @@ -2,5 +2,5 @@ class RegistrationError(Exception): """Raised when a SIP REGISTER request fails with an unexpected response. The exception message includes the response status code and reason phrase - from the server, e.g. ``"403 Forbidden"`` or ``"500 Server Error"``. + from the server, e.g. `"403 Forbidden"` or `"500 Server Error"`. """ diff --git a/voip/sip/messages.py b/voip/sip/messages.py index 3fd5cc9..4eacbab 100644 --- a/voip/sip/messages.py +++ b/voip/sip/messages.py @@ -113,7 +113,7 @@ def __bytes__(self) -> bytes: @property def branch(self) -> str: """Branch parameter from the top Via header (RFC 3261 §20.42).""" - _, uri = self.headers["Via"].split() + _, uri = self.headers.getlist("Via")[0].split() return SipURI.parse(f"sip:{uri}").parameters["branch"] @property @@ -177,13 +177,19 @@ def from_request( cls, request: Request, *, headers=None, dialog: Dialog = None, **kwargs ) -> Response: """Create a response from a request, copying relevant headers.""" - headers = { - "Via": request.headers["Via"], - "From": request.headers["From"], - "To": f"{request.headers['To']};tag={dialog.remote_tag}" - if dialog and dialog.remote_tag - else request.headers["To"], - "Call-ID": request.headers["Call-ID"], - "CSeq": request.headers["CSeq"], - } | (headers or {}) - return cls(headers=headers, **kwargs) + response_headers = SIPHeaderDict( + { + "From": request.headers["From"], + "To": f"{request.headers['To']};tag={dialog.local_tag}" + if dialog and dialog.local_tag + else request.headers["To"], + "Call-ID": request.headers["Call-ID"], + "CSeq": request.headers["CSeq"], + } + ) + for via in request.headers.getlist("Via"): + response_headers.add("Via", via) + for record in request.headers.getlist("Record-Route"): + response_headers.add("Record-Route", record) + response_headers |= headers or {} + return cls(headers=response_headers, **kwargs) diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index 0cd9cc5..c092547 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -7,8 +7,8 @@ import asyncio import dataclasses import datetime -import ipaddress import logging +import socket import ssl import typing @@ -46,39 +46,36 @@ @dataclasses.dataclass(kw_only=True, slots=True) -class SessionInitiationProtocol(asyncio.Protocol): +class SessionInitiationProtocol(asyncio.Protocol, asyncio.DatagramProtocol): """ - SIP User Agent Client (UAC) over TLS/TCP [RFC 3261]. + SIP User Agent Client (UAC) over TLS/TCP or UDP [RFC 3261]. Handles SIP message parsing, carrier registration, and transaction management. + The transport is selected automatically from the AOR's `transport` parameter: - Example: - You can use the handler like any [asyncio.Protocol][asyncio.Protocol] in Python. + | `aor.transport` | Underlying transport | + |-----------------|----------------------| + | `UDP` (default) | UDP datagram socket | + | `TCP` | plain TCP | + | `TLS` | TCP with TLS | + + Use [`run`][voip.sip.protocol.SessionInitiationProtocol.run] for a single + outbound connection and [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve] + for a persistent inbound server with automatic reconnection. + Example: ```python import asyncio - - from voip.sip import SessionInitiationProtocol + from voip.sip import SessionInitiationProtocol, Dialog async def main(): - loop = asyncio.get_running_loop() - - transport, protocol = await loop.create_connection( - SessionInitiationProtocol, - '0.0.0.0', 5060) - - try: - await asyncio.Future() - finally: - transport.close() - - - asyncio.run(main()) + protocol = await SessionInitiationProtocol.run( + aor=SipURI.parse("sip:alice@carrier.example;transport=UDP"), + dialog_class=Dialog, + ) + # place outbound calls via protocol … ``` - However, this example is incomplete, since the protocol will require some - arguments, like a reference to the RTP protocol and an AOR. - > [!Note] > The support is limited to UAC (client mode). > This library currently does not implement server (UAS) functionality. @@ -91,8 +88,9 @@ async def main(): dialog_class: [Dialog][voip.sip.Dialog] subclass used to create dialogs for incoming calls. Defaults to the base [Dialog][voip.sip.Dialog] which rejects all calls with - ``486 Busy Here``. - keepalive_interval: Keep-alive ping interval. Should be between 30 and 90 seconds. + `486 Busy Here`. + keepalive_interval: Keep-alive ping interval for TCP transports. + Should be between 30 and 90 seconds (RFC 5626). """ @@ -115,85 +113,148 @@ async def main(): registered_event: asyncio.Event = dataclasses.field( init=False, default_factory=asyncio.Event ) - transport: asyncio.Transport | None = dataclasses.field(init=False, default=None) + transport: asyncio.BaseTransport | None = dataclasses.field( + init=False, default=None + ) is_secure: bool = dataclasses.field(init=False, default=False) recv_buffer: bytearray = dataclasses.field(init=False, default_factory=bytearray) - ready_callback: typing.Callable[[], None] | None = dataclasses.field( - default=None, repr=False, compare=False - ) def __post_init__(self): - self.public_address = self.public_address or self.rtp.public_address + if self.public_address is None and self.rtp.public_address is not None: + self.public_address = self.rtp.public_address.result() @classmethod async def run( cls, - fn: typing.Callable[[], None], aor: types.SipURI, dialog_class: type[Dialog], *, + rtp: RealtimeTransportProtocol | None = None, no_verify_tls: bool = False, stun_server: NetworkAddress | None = None, + **kwargs: typing.Any, ) -> SessionInitiationProtocol: - """Run a SIP session and call *fn* once registered. + """Connect to the SIP proxy and return once registered. - Establishes RTP and SIP/TLS connections derived from *aor*, then - **suspends until SIP registration is confirmed** before returning the - ready protocol. After this call returns, the MCP server (or any other - caller) may safely place outbound calls. + Establishes RTP (if not provided) and SIP/TLS connections derived from + *aor*, then **suspends until SIP registration is confirmed** before + returning the ready protocol. After this call returns, the caller may + safely place outbound calls or start an MCP server. - The transport protocol (TLS vs plain TCP) and proxy address are read + The transport protocol (TLS vs. plain TCP) and proxy address are read from *aor* directly — no extra arguments are needed. Args: - fn: Called when the SIP session is registered, before - `run` returns. Receives no arguments. May use - [`asyncio.create_task`][] for async work. - aor: SIP Address of Record, e.g. ``sip:alice@carrier.example``. - The host, port, and ``transport`` parameter are used to connect + aor: SIP Address of Record, e.g. `sip:alice@carrier.example`. + The host, port, and `transport` parameter are used to connect to the SIP proxy. dialog_class: [`Dialog`][voip.sip.Dialog] subclass used for - inbound calls. Defaults to the base + inbound calls. Defaults to the base [`Dialog`][voip.sip.Dialog], which rejects all calls. + rtp: Existing RTP endpoint to reuse. When `None` (default) a + new datagram endpoint is created from *aor* and *stun_server*. + Pass an existing instance to share one endpoint across + reconnections (see [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve]). no_verify_tls: Disable TLS certificate verification. Insecure; for - testing only. Defaults to ``False``. - stun_server: STUN server for RTP NAT traversal. Defaults to - ``stun.cloudflare.com:3478``. + testing only. Defaults to `False`. + stun_server: STUN server for RTP NAT traversal. Ignored when *rtp* + is provided. + **kwargs: Extra keyword arguments forwarded to the protocol constructor. Returns: The registered [`SessionInitiationProtocol`][voip.sip.protocol.SessionInitiationProtocol] instance, ready to place calls. """ loop = asyncio.get_running_loop() + if rtp is None: + addr_info = socket.getaddrinfo( + str(aor.maddr[0]), aor.maddr[1], type=socket.SOCK_DGRAM + ) + rtp_bind_address = ( + "::" if addr_info[0][0] == socket.AF_INET6 else "0.0.0.0" # noqa: S104 + ) + rtp = await RealtimeTransportProtocol.serve(rtp_bind_address, stun_server) + if aor.transport == "UDP": + _, protocol = await loop.create_datagram_endpoint( + lambda: cls(aor=aor, rtp=rtp, dialog_class=dialog_class, **kwargs), + remote_addr=(str(aor.maddr[0]), aor.maddr[1]), + ) + else: + ssl_context: ssl.SSLContext | None = None + if aor.transport == "TLS": + ssl_context = ssl.create_default_context() + if no_verify_tls: + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + _, protocol = await loop.create_connection( + lambda: cls(aor=aor, rtp=rtp, dialog_class=dialog_class, **kwargs), + host=str(aor.maddr[0]), + port=aor.maddr[1], + ssl=ssl_context, + ) + await protocol.registered_event.wait() + return protocol - rtp_bind_address = ( - "::" if isinstance(aor.maddr[0], ipaddress.IPv6Address) else "0.0.0.0" # noqa: S104 - ) - _, rtp_protocol = await loop.create_datagram_endpoint( - lambda: RealtimeTransportProtocol(stun_server_address=stun_server), - local_addr=(rtp_bind_address, 0), - ) + @classmethod + async def serve( + cls, + aor: types.SipURI, + dialog_class: type[Dialog], + *, + rtp: RealtimeTransportProtocol | None = None, + no_verify_tls: bool = False, + stun_server: NetworkAddress | None = None, + **kwargs: typing.Any, + ) -> None: + """Register with a carrier and handle inbound calls, reconnecting on disconnect. - ssl_context: ssl.SSLContext | None = None - if aor.transport == "TLS": - ssl_context = ssl.create_default_context() - if no_verify_tls: - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE + Creates one RTP endpoint for the lifetime of the process, then enters a + persistent loop: connect to the SIP proxy, wait for the connection to drop, + and reconnect with exponential back-off. Use this for long-running + inbound-call servers. - _, protocol = await loop.create_connection( - lambda: cls( - aor=aor, rtp=rtp_protocol, dialog_class=dialog_class, ready_callback=fn - ), - host=str(aor.maddr[0]), - port=aor.maddr[1], - ssl=ssl_context, + The transport protocol (TLS vs. plain TCP) and proxy address are read from + *aor* directly. + + Args: + aor: SIP Address of Record, e.g. `sip:alice@carrier.example`. + dialog_class: [`Dialog`][voip.sip.Dialog] subclass used for + inbound calls. + rtp: Existing RTP endpoint to reuse. When `None` (default) a + new datagram endpoint is created from *aor* and *stun_server*. + Pass an existing instance to share one endpoint across + reconnections (see [`serve`][voip.sip.protocol.SessionInitiationProtocol.serve]). + no_verify_tls: Disable TLS certificate verification. Insecure; for + testing only. Defaults to `False`. + stun_server: STUN server for RTP NAT traversal. + **kwargs: Extra keyword arguments forwarded to the protocol constructor. + """ + addr_info = socket.getaddrinfo( + str(aor.maddr[0]), aor.maddr[1], type=socket.SOCK_DGRAM ) - await protocol.registered_event.wait() - return protocol + if rtp is None: + rtp_bind_address = ( + "::" if addr_info[0][0] == socket.AF_INET6 else "0.0.0.0" # noqa: S104 + ) + rtp = await RealtimeTransportProtocol.serve(rtp_bind_address, stun_server) + backoff_secs = 1 + while True: + try: + protocol = await cls.run( + aor, dialog_class, rtp=rtp, no_verify_tls=no_verify_tls, **kwargs + ) + 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) def register_dialog(self, dialog: Dialog) -> None: - """Register *dialog* keyed by ``(dialog.local_tag, dialog.remote_tag)``.""" + """Register *dialog* keyed by `(dialog.local_tag, dialog.remote_tag)`.""" if dialog.remote_tag is None: logger.warning("Dialog without remote tag cannot be registered: %r", dialog) else: @@ -220,23 +281,32 @@ def drop_transaction(self, tx: Transaction) -> None: except KeyError: logger.warning("Transaction not found for removal: %r", tx) - def connection_made(self, transport: asyncio.Transport) -> None: # type: ignore[override] - """Store the TLS/TCP transport and start RTP mux + carrier registration.""" + def connection_made(self, transport: asyncio.BaseTransport) -> None: # type: ignore[override] + """Store the transport and start carrier registration. + + Keepalive pings (RFC 5626) are only started for TCP/TLS transports. + """ self.transport = transport - self.is_secure = transport.get_extra_info("ssl_object") is not None + self.is_secure = ( + not isinstance(transport, asyncio.DatagramTransport) + and transport.get_extra_info("ssl_object") is not None + ) try: loop = asyncio.get_running_loop() tx = RegistrationTransaction(sip=self, method=SIPMethod.REGISTER) self.register_transaction(tx) loop.create_task(self.handle_registration(tx)) - self.keepalive_task = loop.create_task(self.send_keepalive()) + if not isinstance(transport, asyncio.DatagramTransport): + self.keepalive_task = loop.create_task(self.send_keepalive()) except RuntimeError: pass # no running loop in synchronous test setups async def send_keepalive(self) -> None: while True: await asyncio.sleep(self.keepalive_interval.total_seconds()) - if self.transport is None: + if self.transport is None or isinstance( + self.transport, asyncio.DatagramTransport + ): return logger.info("PING", extra={"addr": self.public_address}) self.transport.write(PING) @@ -250,6 +320,14 @@ def data_received(self, data: bytes) -> None: for frame in self._extract_frames(): self._dispatch_frame(frame) + def datagram_received(self, data: bytes, addr: tuple) -> None: # type: ignore[override] + """Dispatch a complete UDP SIP datagram.""" + self._dispatch_frame(data) + + def error_received(self, exc: Exception) -> None: # type: ignore[override] + """Log a UDP transport error.""" + logger.warning("UDP error received", exc_info=exc) + def _extract_frames(self) -> typing.Generator[memoryview | bytes]: # noqa: C901 while self.recv_buffer: if self.recv_buffer[0:1] != b"\r": @@ -287,12 +365,14 @@ def _extract_frames(self) -> typing.Generator[memoryview | bytes]: # noqa: C901 break def _dispatch_frame(self, frame: memoryview | bytes) -> None: - peer = NetworkAddress(*self.transport.get_extra_info("peername")) + peer = NetworkAddress(*self.transport.get_extra_info("peername")[:2]) if frame == PONG: logger.info("PONG", extra={"addr": peer}) elif frame == PING: logger.info("PING", extra={"addr": peer}) - if self.transport: + if self.transport and not isinstance( + self.transport, asyncio.DatagramTransport + ): logger.info("PONG", extra={"addr": self.public_address}) self.transport.write(PONG) else: @@ -313,14 +393,18 @@ def _dispatch_frame(self, frame: memoryview | bytes) -> None: self.response_received(response) def send(self, message: Response | Request) -> None: - """Serialize and send a SIP message over the TLS/TCP connection.""" + """Serialize and send a SIP message over the active transport.""" logger.debug("Sending %r", message) message.headers.setdefault("User-Agent", USER_AGENT) - if self.transport is not None: + if self.transport is None: + return + if isinstance(self.transport, asyncio.DatagramTransport): + self.transport.sendto(bytes(message)) + else: self.transport.write(bytes(message)) def close(self) -> None: - """Close the TLS/TCP transport and the RTP mux.""" + """Close the transport.""" if self.transport is not None: self.transport.close() @@ -372,9 +456,11 @@ def request_received(self, request: Request) -> None: InviteTransaction.receive(request=request, sip=self) ) case SIPMethod.ACK: - # For non-2xx ACKs the INVITE tx is still present; route by branch. + # For non-2xx ACKs the INVITE tx is still present; route by dialog. try: - tx = self._transactions[request.branch] + tx = self._dialogs[ + request.remote_tag, request.local_tag + ].invite_transaction except KeyError: self.send( Response.from_request( @@ -436,21 +522,13 @@ def on_registered(self) -> None: post-registration activity. The base implementation is a no-op. """ self.registered_event.set() - if self.ready_callback is not None: - self.ready_callback() @property def contact(self) -> str: - """Return a ``Contact:`` header value for this UA. - - The URI scheme mirrors `aor`: a ``sips:`` AOR produces a - ``sips:`` Contact (the strongest TLS guarantee); a ``sip:`` AOR over - TLS produces ``sip:`` with ``transport=tls``; plain TCP produces plain - ``sip:``. + """Return a `Contact:` header value for this UA. - 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. + The `ob` parameter ([RFC 5626 §5]) advertises outbound keep-alive + support to the registrar for TCP/TLS transports. [RFC 5626 §5]: https://datatracker.ietf.org/doc/html/rfc5626#section-5 """ @@ -459,14 +537,15 @@ def contact(self) -> str: if self.aor.user else str(self.public_address) ) - ob_uri_param = ";ob" if self.aor.scheme == "sips": - return f"" - tls_param = ";transport=tls" if self.is_secure else ";transport=tcp" - return f"" + return f"" + if isinstance(self.transport, asyncio.DatagramTransport): + return f"" + transport_param = "tls" if self.is_secure else "tcp" + return f"" def connection_lost(self, exc: Exception | None) -> None: - """Handle a lost TLS/TCP connection.""" + """Handle a lost or closed transport connection.""" if exc is not None: logger.exception("Connection lost", exc_info=exc) if self.keepalive_task is not None: diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py index d515dfd..42ff5b6 100644 --- a/voip/sip/transactions.py +++ b/voip/sip/transactions.py @@ -52,7 +52,7 @@ class Transaction(asyncio.Future): """ Initiated by a request, completed by any number of responses. - Transactions are awaitable: ``await tx`` suspends until the transaction + Transactions are awaitable: `await tx` suspends until the transaction reaches its terminal state and resolves to the dialog. Args: @@ -214,7 +214,7 @@ def response_received(self, response: Response) -> None: realm = params.get("realm", "") nonce = params.get("nonce", "") opaque = params.get("opaque") - algorithm = params.get("algorithm", DigestAlgorithm.SHA_256) + algorithm = params.get("algorithm", DigestAlgorithm.MD5) qop_options = params.get("qop", "") qop = ( DigestQoP.AUTH.value @@ -282,7 +282,7 @@ def parse_auth_challenge(header: str) -> dict[str, str]: """Parse Digest challenge parameters from a WWW-Authenticate/Proxy-Authenticate header. Args: - header: The raw ``WWW-Authenticate`` or ``Proxy-Authenticate`` header value. + header: The raw `WWW-Authenticate` or `Proxy-Authenticate` header value. Returns: A dict mapping parameter names to their unquoted values. @@ -313,27 +313,27 @@ def digest_response( """Compute a SIP digest response per RFC 3261 §22 and RFC 8760. RFC 8760 deprecates MD5 and mandates support for SHA-256 and - SHA-512-256. The ``algorithm`` parameter selects the hash function; - it defaults to ``SHA-256``. + SHA-512-256. The `algorithm` parameter selects the hash function; + it defaults to `SHA-256`. Args: username: SIP username (AOR user part). password: SIP password. realm: Digest realm from the challenge. nonce: Digest nonce from the challenge. - method: SIP method string (e.g. ``"REGISTER"``). + method: SIP method string (e.g. `"REGISTER"`). uri: Request-URI string used in the digest. - algorithm: Digest algorithm identifier (default: ``"SHA-256"``). - qop: Quality-of-protection value, or ``None``. - nc: Nonce count hex string (default: ``"00000001"``). - cnonce: Client nonce, required for ``*-sess`` algorithms and ``qop``. + algorithm: Digest algorithm identifier (default: `"SHA-256"`). + qop: Quality-of-protection value, or `None`. + nc: Nonce count hex string (default: `"00000001"`). + cnonce: Client nonce, required for `*-sess` algorithms and `qop`. Returns: Hex-encoded digest response string. Raises: - ValueError: If ``algorithm`` is not a recognised `DigestAlgorithm`, - or if a ``*-sess`` algorithm is requested without a ``cnonce``. + ValueError: If `algorithm` is not a recognised `DigestAlgorithm`, + or if a `*-sess` algorithm is requested without a `cnonce`. """ try: hash_name = cls.DIGEST_HASH_NAME[algorithm] @@ -537,7 +537,7 @@ def call_received(self) -> None: self.dialog.route_set = list(self.request.headers.getlist("Record-Route")) self.sip.register_dialog(self.dialog) - call_handler = session_class( + session = session_class( rtp=self.sip.rtp, caller=caller, media=negotiated_media, @@ -560,14 +560,13 @@ def call_received(self) -> None: ) else: remote_rtp_address = None - self.sip.rtp.register_call(remote_rtp_address, call_handler) + self.sip.rtp.register_call(remote_rtp_address, session) if remote_rtp_address is not None: self.sip.rtp.send(b"\x00", remote_rtp_address) - record_route = self.request.headers.get("Record-Route") session_id = str(secrets.randbelow(2**32) + 1) - rtp_public = self.sip.public_address + rtp_public = self.sip.rtp.public_address.result() sdp_media_attributes = [Attribute(name="sendrecv")] if srtp_session is not None: sdp_media_attributes.append( @@ -580,7 +579,6 @@ def call_received(self) -> None: status_code=SIPStatus.OK, phrase=SIPStatus.OK.phrase, headers={ - **({"Record-Route": record_route} if record_route else {}), "Contact": self.sip.contact, "Allow": self.sip.allow_header, "Supported": "replaces", @@ -632,7 +630,7 @@ async def send( Args: sip: The SIP session to send from. - target: SIP or tel URI of the callee (e.g. ``"sip:+15551234567@carrier.com"`` or ``"tel:+15551234567"``). + target: SIP or tel URI of the callee (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`). dialog: The dialog to associate with this call. session_class: Session implementation that will be initialized for the call. **session_kwargs: Additional keyword arguments forwarded to the @@ -656,7 +654,7 @@ async def send( tx.pending_call_class = session_class tx.pending_call_kwargs = session_kwargs - rtp_public = sip.rtp.public_address + rtp_public = sip.rtp.public_address.result() session_id = str(secrets.randbelow(2**32) + 1) sdp_offer = SessionDescription( origin=Origin( @@ -874,7 +872,7 @@ async def send( { "Via": ( f"SIP/2.0/{sip.aor.transport}" - f' {sip.rtp.public_address};oc-algo="loss";oc;rport;branch={tx.branch}' + f' {sip.rtp.public_address.result()};oc-algo="loss";oc;rport;branch={tx.branch}' ), "Max-Forwards": "70", "From": dialog.local_party, diff --git a/voip/sip/types.py b/voip/sip/types.py index cd5a886..af833d0 100644 --- a/voip/sip/types.py +++ b/voip/sip/types.py @@ -43,7 +43,7 @@ class SipURI(str): stored in header dicts unchanged. The `parse` classmethod decodes a raw SIP URI string into structured fields. IPv6 addresses in the host part must be enclosed in square brackets per [RFC 2732] - (e.g. ``sip:alice@[::1]:5060``); the stored `host` is the bare address + (e.g. `sip:alice@[::1]:5060`); the stored `host` is the bare address without brackets. [RFC 3261 §19.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-19.1 @@ -194,7 +194,7 @@ def ttl(self) -> int | None: @property def transport(self): return ( - self.parameters.get("transport", "TLS").upper() + self.parameters.get("transport", "UDP").upper() if self.scheme == "sip" else "TLS" ) @@ -203,8 +203,8 @@ def transport(self): class CallerID(str): """SIP From/To header value with structured access and privacy-safe repr. - Behaves as a plain ``str`` so it is wire-format compatible and can be - stored in header dicts unchanged. ``repr()`` returns a short anonymized + Behaves as a plain `str` so it is wire-format compatible and can be + stored in header dicts unchanged. `repr()` returns a short anonymized form that shows only the last four characters of the user part and the carrier domain — useful for log messages. diff --git a/voip/stun.py b/voip/stun.py index e19311e..422ce5e 100644 --- a/voip/stun.py +++ b/voip/stun.py @@ -10,6 +10,7 @@ __all__ = ["STUNAttributeType", "STUNMessageType", "STUNProtocol"] + from voip.types import NetworkAddress logger = logging.getLogger(__name__) @@ -43,11 +44,11 @@ def _parse_address( Args: value: Raw attribute value bytes (everything after the type/length TLV header). xor_key: XOR key bytes — must be exactly 16 bytes - (``MAGIC_COOKIE (4 bytes) || transaction_id (12 bytes)``) + (`MAGIC_COOKIE (4 bytes) || transaction_id (12 bytes)`) for XOR-MAPPED-ADDRESS, or empty bytes for plain MAPPED-ADDRESS. Returns: - ``(ip_address, port)`` on success, ``None`` when *value* is + `(ip_address, port)` on success, `None` when *value* is too short or the address family is unrecognised. """ assert not xor_key or len(xor_key) == 16, "xor_key must be 16 bytes or empty" # noqa: S101 @@ -177,7 +178,7 @@ def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: ): self._parse_stun_response(data) return - self.packet_received(data, NetworkAddress(*addr)) + self.packet_received(data, NetworkAddress(*addr[:2])) def connection_lost(self, exc: Exception | None) -> None: """Clear the internal transport reference on disconnect.""" @@ -191,7 +192,7 @@ def packet_received(self, data: bytes, addr: NetworkAddress) -> None: Args: data: Raw datagram payload (first byte ≥ 4, not a STUN packet). - addr: Source ``(host, port)`` of the datagram. + addr: Source `(host, port)` of the datagram. """ def _send_stun_request(self) -> None: @@ -215,6 +216,7 @@ def _send_stun_request(self) -> None: def _parse_stun_response(self, data: bytes) -> None: """Parse a STUN Binding Success Response and invoke :meth:`stun_connection_made`.""" + logger.debug("Parsing STUN response (len=%d)", len(data)) if len(data) < 20: return message_type, _message_len, magic_cookie = struct.unpack(">HHI", data[:8])