diff --git a/README.md b/README.md index 2170101..8c86910 100644 --- a/README.md +++ b/README.md @@ -74,7 +74,7 @@ async def main(): lambda: MySession( aor="sips:alice@example.com", username="alice", - password="secret", + password="secret", # noqa: S106 ), host="sip.example.com", port=5061, diff --git a/docs/feature_roadmap.md b/docs/feature_roadmap.md index d46eb01..283dee7 100644 --- a/docs/feature_roadmap.md +++ b/docs/feature_roadmap.md @@ -18,15 +18,25 @@ with SDES key exchange carried inline in the SDP `a=crypto:` attribute ### NAT Traversal (STUN) -STUN Binding Request / Response with `XOR-MAPPED-ADDRESS` for RTP public -address discovery ([RFC 5389]). Uses Cloudflare's STUN server by default; +STUN Binding Request / Response with `XOR-MAPPED-ADDRESS` and `MAPPED-ADDRESS` +for RTP public address discovery ([RFC 5389]). Full IPv4 and IPv6 address +parsing for both attribute types. Uses Cloudflare's STUN server by default; configurable or disabled per session. ### Session Description (SDP) Offer / answer model for audio calls. Codec negotiation for Opus ([RFC 7587]), G.722, PCMU, and PCMA ([RFC 3551]). Full SDP lexer with Pygments syntax -highlighting. +highlighting. IPv6 connection addresses advertised with `IP6` address type +per [RFC 4566 §5.7]. + +### IPv6 + +Full dual-stack support across SIP signalling, RTP media, and STUN discovery. +IPv6 addresses in SIP URIs and Via/Contact headers are wrapped in square +brackets per [RFC 2732]. The RTP UDP socket is bound to `::` when the SIP +signalling connection is over IPv6. STUN XOR-MAPPED-ADDRESS and MAPPED-ADDRESS +attributes with IPv6 address family are correctly decoded per [RFC 5389 §15.2]. ### Audio Codecs @@ -59,12 +69,15 @@ ______________________________________________________________________ [ollama]: https://ollama.com/ [pocket tts]: https://github.com/pocket-ai/pocket-tts [pyav]: https://pyav.org/ +[rfc 2732]: https://datatracker.ietf.org/doc/html/rfc2732 [rfc 3261]: https://datatracker.ietf.org/doc/html/rfc3261 [rfc 3550]: https://datatracker.ietf.org/doc/html/rfc3550 [rfc 3551]: https://datatracker.ietf.org/doc/html/rfc3551 [rfc 3711]: https://datatracker.ietf.org/doc/html/rfc3711 +[rfc 4566 §5.7]: https://datatracker.ietf.org/doc/html/rfc4566#section-5.7 [rfc 4568]: https://datatracker.ietf.org/doc/html/rfc4568 [rfc 5389]: https://datatracker.ietf.org/doc/html/rfc5389 +[rfc 5389 §15.2]: https://datatracker.ietf.org/doc/html/rfc5389#section-15.2 [rfc 7587]: https://datatracker.ietf.org/doc/html/rfc7587 [rfc 7983]: https://datatracker.ietf.org/doc/html/rfc7983 [rfc 8760]: https://datatracker.ietf.org/doc/html/rfc8760 diff --git a/tests/sip/test_messages.py b/tests/sip/test_messages.py index cc8488f..b70b185 100644 --- a/tests/sip/test_messages.py +++ b/tests/sip/test_messages.py @@ -53,7 +53,7 @@ def test_parse__response(self): result = Message.parse(data) assert isinstance(result, Response) assert result.status_code == 200 - assert result.reason == "OK" + assert result.phrase == "OK" assert result.version == "SIP/2.0" assert result.headers == { "Via": "SIP/2.0/UDP pc33.atlanta.com;branch=z9hG4bK776asdhds" @@ -81,7 +81,7 @@ def test_parse__roundtrip_response(self): """Round-trip a SIP response through parse and bytes.""" response = Response( status_code=404, - reason="Not Found", + phrase="Not Found", headers={"From": "sip:bob@biloxi.com"}, ) assert Message.parse(bytes(response)) == response @@ -112,7 +112,7 @@ def test_parse__from_header__roundtrip_preserves_raw_value(self): """str(CallerID) equals the original header string, so serialization is unchanged.""" data = ( b"INVITE sip:bob@biloxi.com SIP/2.0\r\n" - b'From: "015114455910" ;tag=abc\r\n' + b'From: "08001234567" ;tag=abc\r\n' b"\r\n" ) result = Message.parse(data) @@ -165,7 +165,7 @@ def test_response__bytes(self): """Serialize a SIP response to bytes.""" response = Response( status_code=200, - reason="OK", + phrase="OK", headers={"Via": "SIP/2.0/UDP pc33.atlanta.com;branch=z9hG4bK776asdhds"}, ) assert bytes(response) == ( @@ -177,7 +177,7 @@ def test_response__bytes(self): def test_response__bytes__with_sdp_body(self): """Serialize a SIP response with an SDP body to bytes.""" sdp = SessionDescription() - response = Response(status_code=200, reason="OK", body=sdp) + response = Response(status_code=200, phrase="OK", body=sdp) serialized = bytes(response) assert b"Content-Length:" in serialized assert b"v=0" in serialized @@ -185,7 +185,7 @@ def test_response__bytes__with_sdp_body(self): def test_response__bytes__with_sdp_body__auto_content_length(self): """Auto-calculate Content-Length when SDP body is present and header is not set.""" sdp = SessionDescription() - response = Response(status_code=200, reason="OK", body=sdp) + response = Response(status_code=200, phrase="OK", body=sdp) serialized = bytes(response) assert b"Content-Length:" in serialized parsed = Message.parse(serialized) diff --git a/tests/sip/test_protocol.py b/tests/sip/test_protocol.py index 769ebdb..2cae877 100644 --- a/tests/sip/test_protocol.py +++ b/tests/sip/test_protocol.py @@ -3,7 +3,9 @@ import asyncio import dataclasses import hashlib -from unittest.mock import MagicMock +import ipaddress +import re +from unittest.mock import MagicMock, patch import pytest from voip.rtp import RealtimeTransportProtocol, RTPCall @@ -14,9 +16,10 @@ SIP, RegistrationError, SessionInitiationProtocol, + _format_host, _mask_caller, ) -from voip.sip.types import CallerID, DigestAlgorithm +from voip.sip.types import CallerID, DigestAlgorithm, SIPStatus INVITE_WITH_PCMA = ( b"INVITE sip:bob@biloxi.com SIP/2.0\r\n" @@ -141,8 +144,8 @@ def response_received(self, response, addr): class TestMaskCaller: def test_full_from_header_with_display_name(self): """Mask all but the last 4 chars of a 12-digit display name (8 asterisks).""" - header = '"015114455910" ;tag=abc123' - assert _mask_caller(header) == "********5910" + header = '"08001234567" ;tag=abc123' + assert _mask_caller(header) == "*******4567" def test_bare_sip_uri(self): """Extract user part from a bare SIP URI and mask all but the last 4 chars.""" @@ -154,10 +157,10 @@ def test_short_caller__no_masking(self): def test_strips_tag_parameter(self): """The tag= and any subsequent parameters are stripped before masking.""" - header = '"015114455910" ;tag=xyz;other=1' + header = '"08001234567" ;tag=xyz;other=1' result = _mask_caller(header) assert "tag" not in result - assert result.endswith("5910") + assert result.endswith("4567") def test_angle_bracket_uri_without_display_name(self): """Parse style without a display name.""" @@ -167,13 +170,13 @@ def test_angle_bracket_uri_without_display_name(self): class TestCallerID: def test_str__returns_raw_header(self): """str() returns the original SIP header value unchanged.""" - raw = '"015114455910" ;tag=abc' + raw = '"08001234567" ;tag=abc' assert str(CallerID(raw)) == raw def test_repr__masks_display_name_and_includes_domain(self): """repr() shows last 4 chars of display name and the carrier domain.""" - caller = CallerID('"015114455910" ;tag=abc') - assert repr(caller) == "********5910@telefonica.de" + caller = CallerID('"08001234567" ;tag=abc') + assert repr(caller) == "*******4567@telefonica.de" def test_repr__bare_sip_uri(self): """repr() masks the user part of a bare SIP URI and includes the domain.""" @@ -185,8 +188,8 @@ def test_repr__angle_bracket_uri(self): def test_user__phone_number(self): """User property extracts the SIP user part from a phone number URI.""" - caller = CallerID('"015114455910" ') - assert caller.user == "015114455910" + caller = CallerID('"08001234567" ') + assert caller.user == "08001234567" def test_user__bare_uri(self): """User property extracts the username from a bare SIP URI.""" @@ -490,11 +493,11 @@ async def _run_answer(self, protocol, invite, fake_rtp_transport): # Pre-populate the shared RTP mux so _answer() skips socket creation. mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) protocol._rtp_protocol = mux protocol._rtp_transport = fake_rtp_transport # Resolve the SIP protocol's own local address (for Contact header). - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) await protocol.answer(invite, call_class=_CodecAwareCall) @pytest.mark.asyncio @@ -728,6 +731,37 @@ async def test_answer__omits_record_route_when_absent(self, fake_rtp_transport): response, _ = protocol._sent_responses[-1] assert "Record-Route" not in response.headers + @pytest.mark.asyncio + async def test_answer__ipv6_public_address_uses_ip6_addrtype(self): + """When the RTP public address is IPv6, the SDP uses addrtype IP6.""" + protocol = FakeProtocol() + addr = ("2001:db8::1", 5060) + sdp_body = SessionDescription.parse( + "v=0\r\n" + "o=- 0 0 IN IP6 2001:db8::1\r\n" + "s=-\r\n" + "c=IN IP6 2001:db8::1\r\n" + "t=0 0\r\n" + "m=audio 49170 RTP/AVP 0\r\n" + ) + invite = self._make_invite("answer-ipv6-1", sdp_body) + protocol.request_received(invite, addr) + + loop = asyncio.get_running_loop() + mux = RealtimeTransportProtocol() + mux.public_address = loop.create_future() + mux.public_address.set_result((ipaddress.IPv6Address("2001:db8::2"), 12000)) + protocol._rtp_protocol = mux + protocol._rtp_transport = FakeTransport(("2001:db8::2", 12000)) + protocol.local_address = (ipaddress.IPv6Address("2001:db8::2"), 5061) + + await protocol.answer(invite, call_class=_CodecAwareCall) + response, _ = protocol._sent_responses[-1] + assert response.body.origin.addrtype == "IP6" + assert response.body.connection.addrtype == "IP6" + assert response.body.origin.unicast_address == "2001:db8::2" + assert response.body.connection.connection_address == "2001:db8::2" + class TestCANCELHandler: def test_cancel__sends_200_ok_for_cancel(self): @@ -795,7 +829,7 @@ def test_cancel__sends_487_request_terminated_for_invite(self): (r for r, _ in protocol._sent_responses if r.status_code == 487), None ) assert terminated is not None - assert terminated.reason == "Request Terminated" + assert terminated.phrase == "Request Terminated" def test_cancel__487_includes_to_tag(self): """Include the stored To tag in the 487 Request Terminated response.""" @@ -1074,7 +1108,7 @@ async def test_send__serializes_and_forwards_to_transport(self): transport = make_mock_transport() protocol.connection_made(transport) transport.write.reset_mock() # clear any calls made during connection_made - response = Response(status_code=200, reason="OK") + response = Response(status_code=200, phrase="OK") protocol.send(response) protocol.transport.write.assert_called_once_with(bytes(response)) @@ -1087,7 +1121,7 @@ def call_received(self, request): received.append(request) protocol = MySIP(outbound_proxy=("127.0.0.1", 5060), aor="sip:test@example.com") - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) request = make_invite() addr = ("192.0.2.1", 5060) protocol.request_received(request, addr) @@ -1098,7 +1132,7 @@ def call_received(self, request): async def test_call_received__noop_by_default(self): """call_received is a no-op in the base class.""" protocol = SIP(outbound_proxy=("127.0.0.1", 5060), aor="sip:test@example.com") - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) protocol.call_received(make_invite()) # must not raise async def test_answer__sends_200_ok(self): @@ -1106,10 +1140,10 @@ async def test_answer__sends_200_ok(self): loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) mock_rtp_transport = MagicMock() mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) protocol._rtp_protocol = mux @@ -1120,17 +1154,17 @@ async def test_answer__sends_200_ok(self): assert len(protocol._sent) == 1 response, _ = protocol._sent[0] assert response.status_code == 200 - assert response.reason == "OK" + assert response.phrase == "OK" async def test_answer__sdp_contains_opus_audio_line(self): """Include an audio media line in the SDP body of the 200 OK.""" loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) mock_rtp_transport = MagicMock() mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) protocol._rtp_protocol = mux @@ -1147,10 +1181,10 @@ async def _setup_answer_protocol(self): loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) mock_rtp_transport = MagicMock() mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) protocol._rtp_protocol = mux @@ -1237,10 +1271,10 @@ async def test_answer__copies_dialog_headers(self): loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) mock_rtp_transport = MagicMock() mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) protocol._rtp_protocol = mux @@ -1267,10 +1301,10 @@ def __post_init__(self) -> None: loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) mock_rtp_transport = MagicMock() mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) protocol._rtp_protocol = mux @@ -1294,7 +1328,7 @@ def packet_received(self, packet: RTPPacket, addr) -> None: loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol(stun_server_address=None) rtp_transport, _ = await loop.create_datagram_endpoint( lambda: mux, local_addr=("127.0.0.1", 0) @@ -1342,7 +1376,7 @@ def packet_received(self, packet: RTPPacket, addr) -> None: loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol(stun_server_address=None) rtp_transport, _ = await loop.create_datagram_endpoint( lambda: mux, local_addr=("127.0.0.1", 0) @@ -1381,10 +1415,10 @@ async def test_answer__content_length_serialized(self): loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) mock_rtp_transport = MagicMock() mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) protocol._rtp_protocol = mux @@ -1405,7 +1439,7 @@ async def test_answer__reuses_shared_rtp_socket_for_second_call(self): loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol(stun_server_address=None) rtp_transport, _ = await loop.create_datagram_endpoint( lambda: mux, local_addr=("127.0.0.1", 0) @@ -1463,7 +1497,7 @@ async def test_answer__bye_unregisters_call_from_rtp_mux(self): loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol(stun_server_address=None) rtp_transport, _ = await loop.create_datagram_endpoint( lambda: mux, local_addr=("127.0.0.1", 0) @@ -1499,10 +1533,10 @@ async def test_answer__logs_info(self, caplog): loop = asyncio.get_running_loop() protocol = self._CapturingSIP() protocol.transport = make_mock_transport() - protocol.local_address = ("127.0.0.1", 5061) + protocol.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) mux = RealtimeTransportProtocol() mux.public_address = loop.create_future() - mux.public_address.set_result(("127.0.0.1", 12000)) + mux.public_address.set_result((ipaddress.IPv4Address("127.0.0.1"), 12000)) mock_rtp_transport = MagicMock() mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) protocol._rtp_protocol = mux @@ -1523,17 +1557,17 @@ def test_reject__sends_busy_here_by_default(self): response, _ = protocol._sent[0] assert isinstance(response, Response) assert response.status_code == 486 - assert response.reason == "Busy Here" + assert response.phrase == "Busy Here" def test_reject__custom_status(self): """Send the specified status code and reason.""" protocol = self._CapturingSIP() request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) - protocol.reject(request, status_code=603, reason="Decline") + protocol.reject(request, status_code=SIPStatus.DECLINE) response, _ = protocol._sent[0] assert response.status_code == 603 - assert response.reason == "Decline" + assert response.phrase == "Decline" def test_reject__copies_dialog_headers(self): """Copy Via, To, From, Call-ID, and CSeq headers into the response.""" @@ -1585,7 +1619,7 @@ async def test_data_received__keepalive__sends_pong(self): async def test_request_received__unsupported_method__raises(self): """Raise NotImplementedError for any non-INVITE SIP request method.""" protocol = SIP(outbound_proxy=("127.0.0.1", 5060), aor="sip:test@example.com") - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) request = Request(method="OPTIONS", uri="sip:alice@atlanta.com") with pytest.raises(NotImplementedError, match="OPTIONS"): protocol.request_received(request, ("192.0.2.1", 5060)) @@ -1623,6 +1657,20 @@ def call_received(self, request): # --------------------------------------------------------------------------- +class TestFormatHost: + def test_format_host__ipv4__unchanged(self): + """IPv4 addresses are returned unchanged.""" + assert _format_host("192.0.2.1") == "192.0.2.1" + + def test_format_host__ipv6__bracketed(self): + """IPv6 addresses are wrapped in square brackets.""" + assert _format_host("2001:db8::1") == "[2001:db8::1]" + + def test_format_host__hostname__unchanged(self): + """Hostnames (non-IP strings) are returned unchanged.""" + assert _format_host("example.com") == "example.com" + + class TestRegistration: def test_registrar_uri__strips_user_from_aor(self): """Derive registrar URI from AOR by stripping the user part.""" @@ -1667,12 +1715,45 @@ async def _initialize(self): # sip: AOR → sip: registrar URI even over TLS. assert b"REGISTER sip:example.com SIP/2.0" in data + async def test_initialize__ipv6_local_address_binds_rtp_to_double_colon(self): + """When the SIP connection is IPv6, RTP is bound to '::' instead of '0.0.0.0'.""" + bound_addresses: list[tuple] = [] + + class _TrackingSession(SessionInitiationProtocol): + pass + + p = _TrackingSession( + outbound_proxy=("2001:db8::1", 5061), + aor="sips:alice@example.com", + rtp_stun_server_address=None, + ) + p.local_address = (ipaddress.IPv6Address("2001:db8::2"), 5061) + p._is_tls = True + + loop = asyncio.get_running_loop() + + async def fake_create_datagram(factory, *, local_addr=None, **kwargs): + if local_addr is not None: + bound_addresses.append(local_addr) + transport = MagicMock() + transport.get_extra_info.return_value = local_addr or ("::1", 0) + proto = factory() + proto.connection_made(transport) + return transport, proto + + with patch.object(loop, "create_datagram_endpoint", fake_create_datagram): + p.transport = make_mock_transport("2001:db8::2", 5061) + await p._initialize() + + assert bound_addresses, "create_datagram_endpoint was not called" + assert bound_addresses[0][0] == "::" + async def test_register__includes_required_headers(self): """REGISTER request includes From, To, Call-ID, CSeq, Contact and Expires.""" p = make_register_session() transport = make_mock_transport() p.transport = transport - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) p._is_tls = True await p.register() (data,) = transport.write.call_args[0] @@ -1685,7 +1766,7 @@ async def test_register__includes_required_headers(self): async def test_register__increments_cseq(self): """CSeq increments with each REGISTER sent.""" p = make_register_session() - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) p.transport = make_mock_transport() await p.register() assert p.cseq == 1 @@ -1698,7 +1779,7 @@ async def test_register__with_authorization(self): p = make_register_session() transport = make_mock_transport() p.transport = transport - p.local_address = "127.0.0.1", 5061 + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) await p.register(authorization='Digest username="alice"') (data,) = transport.write.call_args[0] assert b'Authorization: Digest username="alice"' in data @@ -1708,7 +1789,7 @@ async def test_register__with_proxy_authorization(self): p = make_register_session() transport = make_mock_transport() p.transport = transport - p.local_address = "127.0.0.1", 5061 + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) await p.register(proxy_authorization='Digest username="alice"') (data,) = transport.write.call_args[0] assert b'Proxy-Authorization: Digest username="alice"' in data @@ -1729,7 +1810,7 @@ def registered(self): ) p.connection_made(make_mock_transport()) p.response_received( - Response(status_code=200, reason="OK", headers={"CSeq": "1 REGISTER"}), + Response(status_code=200, phrase="OK", headers={"CSeq": "1 REGISTER"}), ("192.0.2.2", 5060), ) assert calls == [True] @@ -1740,7 +1821,7 @@ async def test_response_received__200_non_register_raises(self): p.connection_made(make_mock_transport()) with pytest.raises(RegistrationError): p.response_received( - Response(status_code=200, reason="OK", headers={"CSeq": "1 INVITE"}), + Response(status_code=200, phrase="OK", headers={"CSeq": "1 INVITE"}), ("192.0.2.2", 5060), ) @@ -1749,12 +1830,12 @@ async def test_response_received__401_retries_with_authorization(self): p = make_register_session(username="alice", password="secret") # noqa: S106 transport = make_mock_transport() p.transport = transport - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) challenge = 'Digest realm="example.com", nonce="abc123"' p.response_received( Response( status_code=401, - reason="Unauthorized", + phrase="Unauthorized", headers={"WWW-Authenticate": challenge, "CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5061), @@ -1772,12 +1853,12 @@ async def test_response_received__407_retries_with_proxy_authorization(self): p = make_register_session(username="alice", password="secret") # noqa: S106 transport = make_mock_transport() p.transport = transport - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) challenge = 'Digest realm="example.com", nonce="xyz"' p.response_received( Response( status_code=407, - reason="Proxy Auth Required", + phrase="Proxy Auth Required", headers={"Proxy-Authenticate": challenge, "CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5061), @@ -1792,12 +1873,12 @@ async def test_response_received__401_with_qop_auth_includes_nc_cnonce(self): p = make_register_session() transport = make_mock_transport() p.transport = transport - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) challenge = 'Digest realm="example.com", nonce="n", qop="auth"' p.response_received( Response( status_code=401, - reason="Unauthorized", + phrase="Unauthorized", headers={"WWW-Authenticate": challenge, "CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5061), @@ -1813,12 +1894,12 @@ async def test_response_received__401_with_opaque_echoes_opaque(self): p = make_register_session() transport = make_mock_transport() p.transport = transport - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) challenge = 'Digest realm="example.com", nonce="n", opaque="secret-opaque"' p.response_received( Response( status_code=401, - reason="Unauthorized", + phrase="Unauthorized", headers={"WWW-Authenticate": challenge, "CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5061), @@ -1829,10 +1910,8 @@ async def test_response_received__401_with_opaque_echoes_opaque(self): async def test_register__via_header_has_rport(self): """REGISTER request includes a Via header with the rport parameter.""" - import re - p = make_register_session() - p.local_address = "192.0.2.10", 5061 + p.local_address = (ipaddress.IPv4Address("192.0.2.10"), 5061) transport = make_mock_transport("192.0.2.10", 5061) p.transport = transport p._is_tls = True @@ -1843,10 +1922,8 @@ async def test_register__via_header_has_rport(self): async def test_register__via_branch_is_unique_per_request(self): """Each REGISTER generates a unique Via branch.""" - import re - p = make_register_session() - p.local_address = "127.0.0.1", 5061 + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) transport = make_mock_transport() p.transport = transport await p.register() @@ -1861,7 +1938,7 @@ async def test_register__via_branch_is_unique_per_request(self): async def test_register__contact_uses_local_addr(self): """Contact header uses sip:;transport=tls when AOR is sip: over TLS.""" p = make_register_session() - p.local_address = "10.0.0.5", 5061 + p.local_address = (ipaddress.IPv4Address("10.0.0.5"), 5061) transport = make_mock_transport("10.0.0.5", 5061) p.transport = transport p._is_tls = True @@ -1872,7 +1949,7 @@ async def test_register__contact_uses_local_addr(self): async def test_register__contact_uses_sips_when_aor_is_sips(self): """Contact header uses sips: when AOR scheme is sips:.""" p = make_register_session(aor="sips:alice@example.com") - p.local_address = "10.0.0.5", 5061 + p.local_address = (ipaddress.IPv4Address("10.0.0.5"), 5061) transport = make_mock_transport("10.0.0.5", 5061) p.transport = transport p._is_tls = True @@ -1880,16 +1957,38 @@ async def test_register__contact_uses_sips_when_aor_is_sips(self): (data,) = transport.write.call_args[0] 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.""" + p = make_register_session(aor="sips:alice@example.com") + p.local_address = (ipaddress.IPv6Address("2001:db8::1"), 5061) + transport = make_mock_transport("2001:db8::1", 5061) + p.transport = transport + p._is_tls = True + await p.register() + (data,) = transport.write.call_args[0] + 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.""" + p = make_register_session() + p.local_address = (ipaddress.IPv6Address("::1"), 5061) + transport = make_mock_transport("::1", 5061) + p.transport = transport + p._is_tls = True + await p.register() + (data,) = transport.write.call_args[0] + assert b"Via: SIP/2.0/TLS [::1]:5061" in data + async def test_response_received__403_raises_registration_error(self): """403 Forbidden for REGISTER raises RegistrationError with the response message.""" p = make_register_session() - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) p.transport = make_mock_transport() with pytest.raises(RegistrationError, match="403 Forbidden"): p.response_received( Response( status_code=403, - reason="Forbidden", + phrase="Forbidden", headers={"CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5061), @@ -1898,13 +1997,13 @@ async def test_response_received__403_raises_registration_error(self): async def test_response_received__unexpected_raises_registration_error(self): """Any unexpected REGISTER response raises RegistrationError.""" p = make_register_session() - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) p.transport = make_mock_transport() with pytest.raises(RegistrationError, match="500 Server Error"): p.response_received( Response( status_code=500, - reason="Server Error", + phrase="Server Error", headers={"CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5061), @@ -1962,7 +2061,7 @@ async def test_response_received__200_ok__logs_info(self, caplog): p.connection_made(make_mock_transport()) with caplog.at_level(logging.INFO, logger="voip.sip"): p.response_received( - Response(status_code=200, reason="OK", headers={"CSeq": "1 REGISTER"}), + Response(status_code=200, phrase="OK", headers={"CSeq": "1 REGISTER"}), ("192.0.2.2", 5060), ) assert any("Registration successful" in r.message for r in caplog.records) @@ -1980,7 +2079,7 @@ async def test_response_received__unexpected_status__raises_registration_error( p.response_received( Response( status_code=500, - reason="Server Error", + phrase="Server Error", headers={"CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5060), @@ -2086,12 +2185,12 @@ async def test_response_received__401_uses_server_algorithm(self): p = make_register_session(username="alice", password="secret") # noqa: S106 transport = make_mock_transport() p.transport = transport - p.local_address = ("127.0.0.1", 5061) + p.local_address = (ipaddress.IPv4Address("127.0.0.1"), 5061) challenge = 'Digest realm="example.com", nonce="abc123", algorithm="SHA-256"' p.response_received( Response( status_code=401, - reason="Unauthorized", + phrase="Unauthorized", headers={"WWW-Authenticate": challenge, "CSeq": "1 REGISTER"}, ), ("192.0.2.2", 5061), diff --git a/tests/sip/test_types.py b/tests/sip/test_types.py new file mode 100644 index 0000000..9c22eb5 --- /dev/null +++ b/tests/sip/test_types.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +import ipaddress + +import pytest +from voip.sip import SipUri + + +class TestSipUri: + @pytest.mark.parametrize( + "uri_str, expected_uri_obj", + [ + # domain + ( + "sip:alice@example.com", + SipUri(scheme="sip", user="alice", host="example.com", port=5060), + ), + ( + "sips:alice@example.com", + SipUri(scheme="sips", user="alice", host="example.com", port=5061), + ), + ( + "sip:alice@example.com:4050", + SipUri(scheme="sip", user="alice", host="example.com", port=4050), + ), + ( + "sips:alice@example.com:4051", + SipUri(scheme="sips", user="alice", host="example.com", port=4051), + ), + # ipv4 + ( + "sip:alice@192.168.1.1", + SipUri(scheme="sip", user="alice", host="192.168.1.1", port=5060), + ), + ( + "sips:alice@192.168.1.1", + SipUri(scheme="sips", user="alice", host="192.168.1.1", port=5061), + ), + ( + "sip:alice@192.168.1.1:4050", + SipUri(scheme="sip", user="alice", host="192.168.1.1", port=4050), + ), + ( + "sips:alice@192.168.1.1:4051", + SipUri(scheme="sips", user="alice", host="192.168.1.1", port=4051), + ), + # ipv6 + ( + "sip:alice@[::1]", + SipUri(scheme="sip", user="alice", host="::1", port=5060), + ), + ( + "sips:alice@[::1]", + SipUri(scheme="sips", user="alice", host="::1", port=5061), + ), + ( + "sip:alice@[::1]:4050", + SipUri(scheme="sip", user="alice", host="::1", port=4050), + ), + ( + "sips:alice@[::1]:4051", + SipUri(scheme="sips", user="alice", host="::1", port=4051), + ), + # uri-parameters + ( + "sip:alice@example.com;transport=tcp", + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + parameters={"transport": "tcp"}, + ), + ), + ( + "sip:alice@example.com;transport=udp;ttl=15", + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + parameters={"transport": "udp", "ttl": "15"}, + ), + ), + # headers + ( + "sip:alice@example.com?foo=bar", + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + headers={"foo": "bar"}, + ), + ), + ( + "sip:alice@example.com?tag=12345&foo=bar", + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + headers={"tag": "12345", "foo": "bar"}, + ), + ), + ( + r"sip:%61lice@atlanta.com;transport=TCP", + SipUri( + scheme="sip", + user="alice", + host="atlanta.com", + port=5060, + parameters={"transport": "TCP"}, + ), + ), + ( + r"sip:atlanta.com;method=REGISTER?to=alice%40atlanta.com", + SipUri( + scheme="sip", + user=None, + host="atlanta.com", + port=5060, + parameters={"method": "REGISTER"}, + headers={"to": "alice@atlanta.com"}, + ), + ), + ], + ) + def test_parse_valid(self, uri_str, expected_uri_obj): + """Parse scheme, user, host and optional port from a valid SIP URI.""" + assert SipUri.parse(uri_str) == expected_uri_obj + + @pytest.mark.parametrize( + "uri_str", + [ + "http://example.com", # wrong scheme + "sip:@example.com", # missing user + "sip:@example.com:invalid-port", # non-integer port + ], + ) + def test_parse_invalid(self, uri_str): + """Raise ValueError when parsing an invalid SIP URI.""" + with pytest.raises(ValueError): + SipUri.parse(uri_str) + + @pytest.mark.parametrize( + "uri_obj, expected_uri_str", + [ + ( + SipUri(scheme="sip", user="alice", host="example.com", port=5061), + "sip:alice@example.com:5061", + ), + ( + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + parameters={"transport": "TCP"}, + ), + "sip:alice@example.com:5060;transport=TCP", + ), + ( + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + headers={"foo": "bar"}, + ), + "sip:alice@example.com:5060?foo=bar", + ), + ( + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + parameters={"transport": "TCP"}, + headers={"foo": "bar"}, + ), + "sip:alice@example.com:5060;transport=TCP?foo=bar", + ), + # IPv6 + ( + SipUri(scheme="sip", user="alice", host="::1", port=5060), + "sip:alice@[::1]:5060", + ), + ( + SipUri( + scheme="sip", + user="alice", + host=ipaddress.IPv6Address("::1"), + port=5060, + ), + "sip:alice@[::1]:5060", + ), + # IPv4 + ( + SipUri(scheme="sip", user="alice", host="127.0.0.1", port=5060), + "sip:alice@127.0.0.1:5060", + ), + ( + SipUri( + scheme="sip", + user="alice", + host=ipaddress.IPv4Address("127.0.0.1"), + port=5060, + ), + "sip:alice@127.0.0.1:5060", + ), + # password in user-info + ( + SipUri( + scheme="sip", + user="alice", + password="secret", # noqa: S106 + host="example.com", + port=5060, + ), + "sip:alice:secret@example.com:5060", + ), + # flag URI parameter (value=None) in __str__ + ( + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + parameters={"lr": None}, + ), + "sip:alice@example.com:5060;lr", + ), + ], + ) + def test_str(self, uri_obj, expected_uri_str): + """Test string representation of SipUri objects.""" + assert str(uri_obj) == expected_uri_str + + @pytest.mark.parametrize( + "uri_str, expected_uri_obj", + [ + # flag URI parameter (;lr with no value) + ( + "sip:alice@example.com;lr", + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + parameters={"lr": None}, + ), + ), + # header without '=' value + ( + "sip:alice@example.com?Subject", + SipUri( + scheme="sip", + user="alice", + host="example.com", + port=5060, + headers={"Subject": ""}, + ), + ), + ], + ) + def test_parse__flag_parameter_and_valueless_header( + self, uri_str, expected_uri_obj + ): + """Parse flag URI parameters and valueless headers.""" + assert SipUri.parse(uri_str) == expected_uri_obj diff --git a/tests/test_main.py b/tests/test_main.py index d4db516..ee0d55e 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import ipaddress import sys from unittest.mock import MagicMock, patch @@ -34,75 +35,11 @@ def make_runner(): return CliRunner() -class TestParseAOR: - def test_parse_aor__sips_no_port(self): - """Parse scheme, user and host from a sips URI without port.""" - from voip.__main__ import _parse_aor - - assert _parse_aor("sips:alice@example.com") == ( - "sips", - "alice", - "example.com", - None, - ) - - def test_parse_aor__sip_with_port(self): - """Parse scheme, user, host and port from a sip URI with port.""" - from voip.__main__ import _parse_aor - - assert _parse_aor("sip:alice@example.com:5060") == ( - "sip", - "alice", - "example.com", - 5060, - ) - - def test_parse_aor__sips_with_port(self): - """Parse all components including an explicit port.""" - from voip.__main__ import _parse_aor - - assert _parse_aor("sips:+15551234567@carrier.com:5061") == ( - "sips", - "+15551234567", - "carrier.com", - 5061, - ) - - def test_parse_aor__invalid_no_at(self): - """Raise BadParameter when user@host part is missing.""" - import click - from voip.__main__ import _parse_aor - - with pytest.raises(click.BadParameter): - _parse_aor("sip:example.com") - - def test_parse_aor__invalid_no_scheme(self): - """Raise BadParameter when scheme is missing.""" - import click - from voip.__main__ import _parse_aor - - with pytest.raises(click.BadParameter): - _parse_aor("alice@example.com") - - -class TestParseHostport: - def test_parse_hostport__without_port(self): - """Return default port 5061 when no port is specified.""" - from voip.__main__ import _parse_hostport - - assert _parse_hostport(None, None, "sip.example.com") == ( - "sip.example.com", - 5061, - ) - - def test_parse_hostport__with_port(self): - """Parse host and port from HOST:PORT format.""" - from voip.__main__ import _parse_hostport - - assert _parse_hostport(None, None, "sip.example.com:5080") == ( - "sip.example.com", - 5080, - ) +def make_mock_transport(host: str = "127.0.0.1", port: int = 5060) -> MagicMock: + """Return a MagicMock transport with a pre-configured sockname.""" + transport = MagicMock() + transport.get_extra_info.return_value = (host, port) + return transport class TestParseStunServer: @@ -128,12 +65,51 @@ def test_parse_stun_server__without_port_uses_stun_default(self): ) +class TestParseHostport: + def test_parse_hostport__bracketed_ipv6_without_port_uses_default(self): + """Return default port and IPv6Address when bracketed IPv6 address has no port.""" + from voip.__main__ import _parse_hostport + + assert _parse_hostport(None, None, "[::1]", default_port=5061) == ( + ipaddress.IPv6Address("::1"), + 5061, + ) + + def test_parse_hostport__bracketed_ipv6_with_port(self): + """Return explicit port and IPv6Address when bracketed IPv6 address includes a port.""" + from voip.__main__ import _parse_hostport + + assert _parse_hostport(None, None, "[::1]:5061") == ( + ipaddress.IPv6Address("::1"), + 5061, + ) + + def test_parse_hostport__unbracketed_ipv6_raises_bad_parameter(self): + """Raise BadParameter when an unbracketed IPv6 literal is given.""" + import click + from voip.__main__ import _parse_hostport + + with pytest.raises(click.BadParameter, match="enclosed in brackets"): + _parse_hostport(None, None, "::1") + + class TestVoIPCommand: def test_voip__verbose_flag(self): """Accept -v flag without error.""" result = make_runner().invoke(voip, ["-v", "--help"]) assert result.exit_code == 0 + def test_sip__aor_without_user_raises_error(self): + """Raise BadParameter when AOR has no user part.""" + from voip.__main__ import voip + + result = make_runner().invoke( + voip, + ["sip", "--password=p", "--stun-server=none", "sip:example.com", "echo"], + ) + assert result.exit_code != 0 + assert "AOR must contain a user part" in (result.output or "") + class TestTranscribeCLI: def test_transcribe__sips_aor_uses_tls(self): @@ -460,7 +436,7 @@ async def fake_connection(factory, *, host, port, ssl): async def run(): with patch.object(protocol, "answer") as mock_answer: - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) mock_answer.assert_called_once() @@ -510,7 +486,7 @@ async def fake_connection(factory, *, host, port, ssl): async def _run_whisper(): with patch.object(protocol, "answer") as mock_answer: - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) mock_answer.assert_called_once() @@ -617,7 +593,7 @@ async def fake_connection(factory, *, host, port, ssl): async def _run_agent(): with patch.object(protocol, "answer") as mock_answer: - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) mock_answer.assert_called_once() @@ -667,7 +643,7 @@ async def fake_connection(factory, *, host, port, ssl): async def _run_ollama(): with patch.object(protocol, "answer") as mock_answer: - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) _, kwargs = mock_answer.call_args @@ -715,7 +691,7 @@ async def fake_connection(factory, *, host, port, ssl): async def _run_voice(): with patch.object(protocol, "answer") as mock_answer: - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) _, kwargs = mock_answer.call_args @@ -820,7 +796,7 @@ async def fake_connection(factory, *, host, port, ssl): async def run(): with patch.object(protocol, "answer") as mock_answer: - protocol.connection_made(MagicMock()) + protocol.connection_made(make_mock_transport()) protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) mock_answer.assert_called_once() diff --git a/tests/test_rtp.py b/tests/test_rtp.py index fb617e2..4b7b8b1 100644 --- a/tests/test_rtp.py +++ b/tests/test_rtp.py @@ -4,6 +4,7 @@ import asyncio import dataclasses +import ipaddress import struct from unittest.mock import MagicMock @@ -261,7 +262,7 @@ def datagram_received(self, data, addr): ) try: result = await proto.public_address - assert result == ("203.0.113.5", 54321) + assert result == (ipaddress.IPv4Address("203.0.113.5"), 54321) assert len(received_requests) == 1 finally: rtp_t.close() diff --git a/tests/test_stun.py b/tests/test_stun.py index 91b5d2c..f520447 100644 --- a/tests/test_stun.py +++ b/tests/test_stun.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import ipaddress import socket import struct import unittest.mock @@ -12,6 +13,7 @@ STUNAttributeType, STUNMessageType, STUNProtocol, + _parse_address, ) @@ -29,6 +31,30 @@ def make_mapped_address_attribute(ip: str, port: int) -> bytes: return struct.pack(">HH", STUNAttributeType.MAPPED_ADDRESS, len(value)) + value +def make_xor_mapped_address_attribute_ipv6( + ip: str, port: int, transaction_id: bytes +) -> bytes: + """Build a XOR-MAPPED-ADDRESS attribute for the given IPv6 address and port. + + Per RFC 5389 §15.2 the port is XORed with the top 16 bits of the magic + cookie and the 128-bit address is XORed with the concatenation of the + magic cookie and the transaction ID. + """ + xor_port = port ^ (MAGIC_COOKIE >> 16) + xor_key = struct.pack(">I", MAGIC_COOKIE) + transaction_id + raw_addr = socket.inet_pton(socket.AF_INET6, ip) + xor_addr = bytes(a ^ b for a, b in zip(raw_addr, xor_key, strict=False)) + value = struct.pack(">BBH", 0x00, 0x02, xor_port) + xor_addr + return struct.pack(">HH", STUNAttributeType.XOR_MAPPED_ADDRESS, len(value)) + value + + +def make_mapped_address_attribute_ipv6(ip: str, port: int) -> bytes: + """Build a MAPPED-ADDRESS attribute for the given IPv6 address and port.""" + raw_addr = socket.inet_pton(socket.AF_INET6, ip) + value = struct.pack(">BBH", 0x00, 0x02, port) + raw_addr + return struct.pack(">HH", STUNAttributeType.MAPPED_ADDRESS, len(value)) + value + + def make_success_response(transaction_id: bytes, *attributes: bytes) -> bytes: """Build a STUN Binding Success Response with the given attributes.""" body = b"".join(attributes) @@ -64,6 +90,17 @@ def test_xor_mapped_address__value(self): assert STUNAttributeType.XOR_MAPPED_ADDRESS == 0x0020 +class TestParseAddress: + def test_too_short__returns_none(self): + """Return None when the attribute value is shorter than 4 bytes.""" + assert _parse_address(b"\x00\x01", b"") is None + + def test_unknown_family__returns_none(self): + """Return None for an unrecognised address family byte.""" + value = struct.pack(">BBH4s", 0x00, 0x03, 1234, b"\x00" * 4) + assert _parse_address(value, b"") is None + + class TestSTUNProtocol: def test_is_datagram_protocol(self): """STUNProtocol is an asyncio.DatagramProtocol subclass.""" @@ -86,7 +123,7 @@ def stun_connection_made(self, transport, addr): # stun_connection_made is called once with the local socket address. assert len(received) == 1 assert received[0][0] is transport - assert received[0][1] == ("127.0.0.1", 5060) + assert received[0][1] == (ipaddress.IPv4Address("127.0.0.1"), 5060) async def test_connection_made__stun_enabled__sends_binding_request(self): """When stun_server_address is set, a STUN Binding Request is sent.""" @@ -164,7 +201,9 @@ def datagram_received(self, data, addr): ) server_addr = server_t.get_extra_info("sockname") - done: asyncio.Future[tuple[str, int]] = loop.create_future() + done: asyncio.Future[ + tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int] + ] = loop.create_future() class TrackingProto(STUNProtocol): def stun_connection_made(self, transport, addr): @@ -177,8 +216,155 @@ def stun_connection_made(self, transport, addr): ) try: result = await asyncio.wait_for(done, 2.0) - assert result == ("203.0.113.5", 12345) + assert result == (ipaddress.IPv4Address("203.0.113.5"), 12345) assert len(received_requests) == 1 finally: client_t.close() server_t.close() + + async def test_parse_stun_response__xor_mapped_address_ipv6(self): + """XOR-MAPPED-ADDRESS with IPv6 family resolves to the correct address.""" + transaction_id = b"\x01" * 12 + attr = make_xor_mapped_address_attribute_ipv6( + "2001:db8::1", 54321, transaction_id + ) + response = make_success_response(transaction_id, attr) + + received: list[tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int]] = [] + + class TrackingProto(STUNProtocol): + def stun_connection_made(self, transport, addr): + received.append(addr) + + proto = TrackingProto(stun_server_address=("::1", 3478)) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + proto.connection_made(transport) + proto._stun_transaction_id = transaction_id + proto.datagram_received(response, ("::1", 3478)) + assert len(received) == 1 + assert received[0] == (ipaddress.IPv6Address("2001:db8::1"), 54321) + + async def test_parse_stun_response__mapped_address_ipv6(self): + """MAPPED-ADDRESS with IPv6 family is used when XOR-MAPPED-ADDRESS is absent.""" + transaction_id = b"\x02" * 12 + attr = make_mapped_address_attribute_ipv6("::1", 12345) + response = make_success_response(transaction_id, attr) + + received: list[tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int]] = [] + + class TrackingProto(STUNProtocol): + def stun_connection_made(self, transport, addr): + received.append(addr) + + proto = TrackingProto(stun_server_address=("::1", 3478)) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + proto.connection_made(transport) + proto._stun_transaction_id = transaction_id + proto.datagram_received(response, ("::1", 3478)) + assert len(received) == 1 + assert received[0] == (ipaddress.IPv6Address("::1"), 12345) + + async def test_parse_stun_response__mapped_address_ipv4_fallback(self): + """MAPPED-ADDRESS with IPv4 family is used when XOR-MAPPED-ADDRESS is absent.""" + transaction_id = b"\x03" * 12 + attr = make_mapped_address_attribute("203.0.113.1", 9999) + response = make_success_response(transaction_id, attr) + + received: list[tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int]] = [] + + class TrackingProto(STUNProtocol): + def stun_connection_made(self, transport, addr): + received.append(addr) + + proto = TrackingProto(stun_server_address=("127.0.0.1", 3478)) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + proto.connection_made(transport) + proto._stun_transaction_id = transaction_id + proto.datagram_received(response, ("127.0.0.1", 3478)) + assert len(received) == 1 + assert received[0] == (ipaddress.IPv4Address("203.0.113.1"), 9999) + + async def test_parse_stun_response__no_address_attribute_logs_error(self, caplog): + """Log an error when the STUN response contains no address attribute.""" + import logging # noqa: PLC0415 + + transaction_id = b"\x04" * 12 + response = make_success_response(transaction_id) + + class TrackingProto(STUNProtocol): + def stun_connection_made(self, transport, addr): + pass # pragma: no cover + + proto = TrackingProto(stun_server_address=("127.0.0.1", 3478)) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + proto.connection_made(transport) + proto._stun_transaction_id = transaction_id + with caplog.at_level(logging.ERROR): + proto.datagram_received(response, ("127.0.0.1", 3478)) + assert any("No address attribute" in r.message for r in caplog.records) + + async def test_close__closes_transport(self): + """close() calls close() on the underlying transport.""" + proto = STUNProtocol(stun_server_address=None) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + transport.get_extra_info.return_value = ("127.0.0.1", 0) + proto.connection_made(transport) + proto.close() + transport.close.assert_called_once() + + def test_send__delivers_datagram(self): + """send() passes data and address to the underlying transport.""" + proto = STUNProtocol(stun_server_address=None) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + transport.get_extra_info.return_value = ("127.0.0.1", 0) + proto.connection_made(transport) + proto.send(b"hello", ("127.0.0.1", 5004)) + transport.sendto.assert_any_call(b"hello", ("127.0.0.1", 5004)) + + def test_error_received__logs_warning(self, caplog): + """error_received() logs a warning and does not raise.""" + import logging # noqa: PLC0415 + + proto = STUNProtocol(stun_server_address=None) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + transport.get_extra_info.return_value = ("127.0.0.1", 0) + proto.connection_made(transport) + with caplog.at_level(logging.WARNING): + proto.error_received(OSError("network error")) + assert any("network error" in r.message for r in caplog.records) + + def test_send_stun_request__no_op_when_transport_is_none(self): + """_send_stun_request() is a no-op when the transport is not set.""" + proto = STUNProtocol(stun_server_address=("127.0.0.1", 3478)) + # transport is None (never connected) + proto._send_stun_request() # must not raise + + def test_parse_stun_response__too_short__ignored(self): + """_parse_stun_response() silently ignores responses shorter than 20 bytes.""" + proto = STUNProtocol(stun_server_address=("127.0.0.1", 3478)) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + transport.get_extra_info.return_value = ("127.0.0.1", 0) + proto.connection_made(transport) + proto._stun_transaction_id = b"\x05" * 12 + proto._parse_stun_response(b"\x01\x01" + b"\x00" * 10) # only 12 bytes + + def test_parse_stun_response__wrong_transaction_id__ignored(self): + """_parse_stun_response() ignores responses with a mismatched transaction ID.""" + transaction_id = b"\x06" * 12 + received: list = [] + + class TrackingProto(STUNProtocol): + def stun_connection_made(self, transport, addr): + received.append(addr) # pragma: no cover + + proto = TrackingProto(stun_server_address=("127.0.0.1", 3478)) + transport = unittest.mock.MagicMock(spec=asyncio.DatagramTransport) + transport.get_extra_info.return_value = ("127.0.0.1", 0) + proto.connection_made(transport) + proto._stun_transaction_id = transaction_id + wrong_tid = b"\xff" * 12 + response = make_success_response( + wrong_tid, make_xor_mapped_address_attribute("203.0.113.5", 1234) + ) + proto._parse_stun_response(response) + assert received == [] diff --git a/voip/__main__.py b/voip/__main__.py index 4227885..b0f4510 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -1,12 +1,15 @@ #!/usr/bin/env python3 import asyncio import dataclasses +import ipaddress import logging +import re import ssl import time from voip.sip import messages from voip.sip.protocol import SessionInitiationProtocol +from voip.sip.types import SipUri try: import click @@ -27,47 +30,21 @@ SIP_TLS_PORT = 5061 -def _parse_aor(value: str) -> tuple[str, str, str, int | None]: - """Parse a SIP URI into `(scheme, user, host, port)`. - - The port is `None` when not present in the URI. - - Examples: - ``` - >>> _parse_aor("sip:alice@example.com") - ('sip', 'alice', 'example.com', None) - >>> _parse_aor("sips:+15551234567@carrier.com:5061") - ('sips', '+15551234567', 'carrier.com', 5061) - ``` - - Args: - value: SIP URI string. - - Returns: - Tuple of (scheme, user, host, port). - - Raises: - click.BadParameter: When the URI is malformed. - """ - scheme, _, rest = value.partition(":") - if not scheme or not rest: - raise click.BadParameter( - f"Invalid SIP URI: {value!r}. Expected sip[s]:user@host[:port]." - ) - user_part, _, hostport = rest.partition("@") - if not hostport: - raise click.BadParameter(f"Invalid SIP URI: {value!r}. Missing user@host part.") - host, _, port_str = hostport.partition(":") - if not host: - raise click.BadParameter(f"Invalid SIP URI: {value!r}. Missing host.") - port: int | None = int(port_str) if port_str else None - return scheme, user_part, host, port +HOSTPORT_PATTERN: re.Pattern[str] = re.compile( + r"^(?:\[(?P[0-9a-fA-F:]+)\]|(?P[^:\[\]]+))" + r"(?::(?P\d+))?$" +) def _parse_hostport( ctx, param, value: str, default_port: int = 5061 -) -> tuple[str, int]: - """Parse `HOST[:PORT]` into a `(host, port)` tuple. +) -> tuple[ipaddress.IPv4Address | ipaddress.IPv6Address | str, int]: + """Parse `HOST[:PORT]` or `[IPv6HOST][:PORT]` into a typed `(host, port)` tuple. + + IPv6 addresses must be enclosed in square brackets per RFC 2732, e.g. + ``[::1]:5061``. The returned host is an + [`IPv4Address`][ipaddress.IPv4Address] or [`IPv6Address`][ipaddress.IPv6Address] + when the value is a numeric IP address, otherwise a plain hostname string. Args: ctx: Click context. @@ -76,18 +53,24 @@ def _parse_hostport( default_port: Port to use when not specified. Returns: - Tuple of (host, port). + Tuple of (host, port) where host is an IP address object or hostname string. Raises: - click.BadParameter: When port is invalid. + click.BadParameter: When value is malformed (unbracketed IPv6 or invalid port). """ - host, _, port_str = value.rpartition(":") - if not host: - return value, default_port + if not (match := HOSTPORT_PATTERN.fullmatch(value)): + if value.count(":") > 1: + raise click.BadParameter( + f"IPv6 address must be enclosed in brackets, e.g. [{value}].", + param=param, + ) + raise click.BadParameter(f"Invalid host:port value: {value!r}.", param=param) + raw_host = match.group("ipv6") or match.group("host") + port = int(match.group("port")) if match.group("port") else default_port try: - return host, int(port_str) + return ipaddress.ip_address(raw_host), port except ValueError: - raise click.BadParameter(f"Invalid port in {value!r}.", param=param) from None + return raw_host, port def _parse_stun_server(ctx, param, value: str | None) -> tuple[str, int] | None: @@ -103,11 +86,8 @@ def _parse_stun_server(ctx, param, value: str | None) -> tuple[str, int] | None: """ if value is None or value.lower() == "none": return None - return _parse_hostport(ctx, param, value, default_port=3478) - - -# Keep the old name as an alias so existing internal callers still work. -_parse_server = _parse_hostport + host, port = _parse_hostport(ctx, param, value, default_port=3478) + return str(host), port class ConsoleMessageProtocol(SessionInitiationProtocol): @@ -220,21 +200,32 @@ def sip(ctx, aor, password, username, proxy, stun_server, no_tls, no_verify_tls) """Session Initiation Protocol (SIP).""" ctx.ensure_object(dict) try: - scheme, aor_user, aor_host, aor_port = _parse_aor(aor) - except click.BadParameter as exc: + parsed_aor = SipUri.parse(aor) + except ValueError as exc: raise click.BadParameter(str(exc), param_hint="AOR") from exc - effective_username = username or aor_user + effective_username = username or parsed_aor.user + if not effective_username: + raise click.BadParameter( + "AOR must contain a user part (e.g. sip:alice@example.com).", + param_hint="AOR", + ) if proxy is not None: proxy_addr = _parse_hostport(ctx, None, proxy) else: - default_port = SIP_TCP_PORT if scheme == "sip" else SIP_TLS_PORT - port = aor_port if aor_port is not None else default_port - proxy_addr = (aor_host, port) + default_port = SIP_TCP_PORT if parsed_aor.scheme == "sip" else SIP_TLS_PORT + port = parsed_aor.port if parsed_aor.port is not None else default_port + proxy_addr = (parsed_aor.host, port) use_tls = not no_tls and proxy_addr[1] != SIP_TCP_PORT - normalized_aor = f"{scheme}:{effective_username}@{aor_host}" + # Build the canonical AOR; IPv6 hosts must be enclosed in brackets per RFC 2732. + host_in_aor = ( + f"[{parsed_aor.host}]" + if isinstance(parsed_aor.host, ipaddress.IPv6Address) + else str(parsed_aor.host) + ) + normalized_aor = f"{parsed_aor.scheme}:{effective_username}@{host_in_aor}" ctx.obj.update( aor=normalized_aor, @@ -249,7 +240,7 @@ def sip(ctx, aor, password, username, proxy, stun_server, no_tls, no_verify_tls) async def _connect_sip( session_factory, - proxy_addr: tuple[str, int], + proxy_addr: tuple[ipaddress.IPv4Address | ipaddress.IPv6Address | str, int], use_tls: bool, no_verify_tls: bool, ) -> None: @@ -263,7 +254,7 @@ async def _connect_sip( ssl_context.verify_mode = ssl.CERT_NONE await loop.create_connection( session_factory, - host=proxy_addr[0], + host=str(proxy_addr[0]), port=proxy_addr[1], ssl=ssl_context, ) diff --git a/voip/ai.py b/voip/ai.py index 5345bf2..530a41f 100644 --- a/voip/ai.py +++ b/voip/ai.py @@ -172,10 +172,10 @@ async def respond(self) -> None: messages=self._messages, ) # clean non-ascii characters from the response for TTS processing - reply = self.emoji_pattern.sub("", response.message.content or "") - self._messages.append({"role": "assistant", "content": reply}) - logger.debug("Agent reply: %r", reply) - await self.send_speech(reply) + if reply := self.emoji_pattern.sub("", response.message.content or ""): + self._messages.append({"role": "assistant", "content": reply}) + logger.debug("Agent reply: %r", reply) + await self.send_speech(reply) async def send_speech(self, text: str) -> None: audio = self.tts_model.generate_audio( diff --git a/voip/rtp.py b/voip/rtp.py index beb3115..de0be4a 100644 --- a/voip/rtp.py +++ b/voip/rtp.py @@ -9,6 +9,7 @@ import asyncio import dataclasses import enum +import ipaddress import json import logging import struct @@ -189,11 +190,15 @@ class RealtimeTransportProtocol(STUNProtocol): calls: dict[tuple[str, int] | None, RTPCall] = dataclasses.field( init=False, default_factory=dict ) - public_address: asyncio.Future[tuple[str, int]] = dataclasses.field( - init=False, default_factory=asyncio.Future - ) + public_address: asyncio.Future[ + tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int] + ] = dataclasses.field(init=False, default_factory=asyncio.Future) - def stun_connection_made(self, transport, addr): + def stun_connection_made( + self, + transport: asyncio.DatagramTransport, + addr: tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int], + ) -> None: self.public_address.set_result(addr) def register_call( diff --git a/voip/sip/__init__.py b/voip/sip/__init__.py index 2ffb3e7..4fd9cac 100644 --- a/voip/sip/__init__.py +++ b/voip/sip/__init__.py @@ -6,6 +6,15 @@ from .messages import Message, Request, Response from .protocol import SessionInitiationProtocol -from .types import CallerID +from .types import CallerID, SIPMethod, SIPStatus, SipUri -__all__ = ["Message", "Request", "Response", "SessionInitiationProtocol", "CallerID"] +__all__ = [ + "Message", + "Request", + "Response", + "SessionInitiationProtocol", + "CallerID", + "SipUri", + "SIPStatus", + "SIPMethod", +] diff --git a/voip/sip/messages.py b/voip/sip/messages.py index c366d51..73cee3e 100644 --- a/voip/sip/messages.py +++ b/voip/sip/messages.py @@ -4,11 +4,15 @@ import abc import dataclasses +import typing from voip.sdp.messages import SessionDescription from ..types import ByteSerializableObject -from .types import CallerID +from .types import CallerID, SIPMethod, SIPStatus + +if typing.TYPE_CHECKING: + from . import SipUri __all__ = ["Request", "Response", "Message"] @@ -46,7 +50,7 @@ def parse(cls, data: bytes) -> Request | Response: version, status_code_str, reason = parts return Response( status_code=int(status_code_str), - reason=reason, + phrase=reason, headers=headers, body=cls._parse_body(headers, body), version=version, @@ -92,8 +96,8 @@ class Request(Message): [RFC 3261 §7.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-7.1 """ - method: str - uri: str + method: SIPMethod | str + uri: SipUri | str def _first_line(self) -> str: return f"{self.method} {self.uri} {self.version}" @@ -107,8 +111,8 @@ class Response(Message): [RFC 3261 §7.2]: https://datatracker.ietf.org/doc/html/rfc3261#section-7.2 """ - status_code: int - reason: str + status_code: SIPStatus | int + phrase: str def _first_line(self) -> str: - return f"{self.version} {self.status_code} {self.reason}" + return f"{self.version} {self.status_code} {self.phrase}" diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index 6c6660d..e9fdb1b 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -10,6 +10,7 @@ import collections import dataclasses import hashlib +import ipaddress import json import logging import re @@ -31,13 +32,36 @@ from voip.srtp import SRTPSession from .messages import Message, Request, Response -from .types import CallerID, DigestAlgorithm, DigestQoP, Status +from .types import CallerID, DigestAlgorithm, DigestQoP, SIPStatus logger = logging.getLogger("voip.sip") __all__ = ["RegistrationError", "SIP", "SessionInitiationProtocol"] +def _format_host(host: str | ipaddress.IPv4Address | ipaddress.IPv6Address) -> str: + """Return *host* wrapped in brackets when it is an IPv6 address. + + RFC 3261 §19.1.1 and RFC 2732 require IPv6 addresses in SIP URIs and + Via/Contact headers to be enclosed in square brackets. + + Args: + host: Host as a typed IP address object or bare host string. + + Returns: + ``[host]`` for IPv6 addresses, *host* unchanged otherwise. + """ + if isinstance(host, ipaddress.IPv6Address): + return f"[{host}]" + if isinstance(host, ipaddress.IPv4Address): + return str(host) + try: + addr = ipaddress.ip_address(host) + return f"[{addr}]" if isinstance(addr, ipaddress.IPv6Address) else host + except ValueError: + return host + + class RegistrationError(Exception): """Raised when a SIP REGISTER request fails with an unexpected response. @@ -54,8 +78,8 @@ def _mask_caller(header: str) -> str: Examples: ``` - >>> _mask_caller('"015114455910" ;tag=abc') - '********5910' + >>> _mask_caller('"08001234567" ;tag=abc') + '*******4567' >>> _mask_caller('sip:alice@example.com') '*lice' ``` @@ -139,7 +163,9 @@ def call_received(self, request: Request) -> None: #: When ``None`` the caller connects directly to the registrar server. #: The address may differ from the registrar domain derived from #: `aor` (e.g. ``proxy.carrier.com`` vs ``carrier.com``). - outbound_proxy: tuple[str, int] | None = None + outbound_proxy: ( + tuple[ipaddress.IPv4Address | ipaddress.IPv6Address | str, int] | None + ) = None aor: str username: str | None = None password: str | None = None @@ -148,7 +174,9 @@ def call_received(self, request: Request) -> None: 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. - local_address: tuple[str, int] = dataclasses.field(init=False) + local_address: tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int] = ( + dataclasses.field(init=False) + ) transport: asyncio.Transport | None = dataclasses.field(init=False, default=None) #: True when the underlying transport is TLS-wrapped; False for plain TCP. _is_tls: bool = dataclasses.field(init=False, default=False) @@ -159,7 +187,11 @@ def __post_init__(self): def connection_made(self, transport: asyncio.Transport) -> None: # type: ignore[override] """Store the TLS/TCP transport and start RTP mux + carrier registration.""" self.transport = transport - self.local_address = transport.get_extra_info("sockname") + # IPv6 sockets return a 4-tuple (host, port, flowinfo, scope_id); + # we only need the first two elements. + sockname = transport.get_extra_info("sockname") + host, port = sockname[0], sockname[1] + 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( @@ -170,11 +202,16 @@ def connection_made(self, transport: asyncio.Transport) -> None: # type: ignore async def _initialize(self) -> None: loop = asyncio.get_running_loop() + rtp_bind = ( + "::" + if isinstance(self.local_address[0], ipaddress.IPv6Address) + else "0.0.0.0" # noqa: S104 + ) self._rtp_transport, self._rtp_protocol = await loop.create_datagram_endpoint( lambda: RealtimeTransportProtocol( stun_server_address=self.rtp_stun_server_address ), - local_addr=("0.0.0.0", 0), # noqa: S104 + local_addr=(rtp_bind, 0), ) await self.register() @@ -290,8 +327,8 @@ def request_received(self, request: Request, addr: tuple[str, int]) -> None: ) self.send( Response( - status_code=Status["OK"], - reason=Status["OK"].name, + status_code=SIPStatus.OK, + phrase=SIPStatus.OK.phrase, headers=self._with_to_tag( { key: value @@ -320,8 +357,8 @@ def request_received(self, request: Request, addr: tuple[str, int]) -> None: ) self.send( Response( - status_code=Status["OK"], - reason=Status["OK"].name, + status_code=SIPStatus.OK, + phrase=SIPStatus.OK.phrase, headers={ key: value for key, value in request.headers.items() @@ -333,8 +370,8 @@ def request_received(self, request: Request, addr: tuple[str, int]) -> None: self._pending_invites.discard(call_id) self.send( Response( - status_code=Status["Request Terminated"], - reason=Status["Request Terminated"].name, + status_code=SIPStatus.REQUEST_TERMINATED, + phrase=SIPStatus.REQUEST_TERMINATED.phrase, headers=self._with_to_tag( { key: value @@ -361,15 +398,15 @@ def response_received( Only processes responses when registration parameters are configured. """ - if response.status_code == Status["OK"] and response.headers.get( + if response.status_code == SIPStatus.OK and response.headers.get( "CSeq", "" ).split()[-1:] == ["REGISTER"]: logger.info("Registration successful") self.registered() return if response.status_code in ( - Status["Unauthorized"], - Status["Proxy Authentication Required"], + SIPStatus.UNAUTHORIZED, + SIPStatus.PROXY_AUTHENTICATION_REQUIRED, ): if not self.username or not self.password: logger.error( @@ -380,7 +417,7 @@ def response_received( "Auth challenge received (%s), retrying with credentials", response.status_code, ) - is_proxy = response.status_code == Status["Proxy Authentication Required"] + is_proxy = response.status_code == SIPStatus.PROXY_AUTHENTICATION_REQUIRED challenge_key = "Proxy-Authenticate" if is_proxy else "WWW-Authenticate" params = self.parse_auth_challenge(response.headers.get(challenge_key, "")) realm = params.get("realm", "") @@ -421,7 +458,7 @@ def response_received( else: asyncio.create_task(self.register(authorization=auth_value)) return - raise RegistrationError(f"{response.status_code} {response.reason}") + raise RegistrationError(f"{response.status_code} {response.phrase}") def call_received(self, request: Request) -> None: """Handle an incoming call. @@ -598,8 +635,8 @@ async def answer( ) self.send( Response( - status_code=Status["OK"], - reason=Status["OK"].name, + status_code=SIPStatus.OK, + phrase=SIPStatus.OK.phrase, headers={ **self._with_to_tag( { @@ -621,14 +658,18 @@ async def answer( sess_id=sess_id, sess_version=sess_id, nettype="IN", - addrtype="IP4", - unicast_address=rtp_public[0], + addrtype="IP6" + if isinstance(rtp_public[0], ipaddress.IPv6Address) + else "IP4", + unicast_address=str(rtp_public[0]), ), timings=[Timing(start_time=0, stop_time=0)], connection=ConnectionData( nettype="IN", - addrtype="IP4", - connection_address=rtp_public[0], + addrtype="IP6" + if isinstance(rtp_public[0], ipaddress.IPv6Address) + else "IP4", + connection_address=str(rtp_public[0]), ), media=[ MediaDescription( @@ -667,7 +708,7 @@ def _build_contact(self, user: str | None = None) -> str: ````. """ aor_scheme = self.aor.partition(":")[0] # "sip" or "sips" - host_port = f"{self.local_address[0]}:{self.local_address[1]}" + host_port = f"{_format_host(self.local_address[0])}:{self.local_address[1]}" addr = f"{user}@{host_port}" if user else host_port if aor_scheme == "sips": return f"" @@ -696,8 +737,8 @@ def ringing(self, request: Request) -> None: ) self.send( Response( - status_code=Status["Ringing"], - reason=Status["Ringing"].name, + status_code=SIPStatus.RINGING, + phrase=SIPStatus.RINGING.phrase, headers=self._with_to_tag( { key: value @@ -712,15 +753,13 @@ def ringing(self, request: Request) -> None: def reject( self, request: Request, - status_code: int = Status["Busy Here"], - reason: str = Status["Busy Here"].name, + status_code: SIPStatus = SIPStatus.BUSY_HERE, ) -> None: """Reject an incoming call. Args: request: The SIP INVITE request (from `call_received`). status_code: SIP response status code (default: 486 Busy Here). - reason: SIP response reason phrase. """ call_id = request.headers.get("Call-ID", "") if call_id not in self._pending_invites: @@ -737,7 +776,7 @@ def reject( "ip": peer[0] if peer else None, "call_id": call_id, "status": status_code, - "reason": reason, + "reason": status_code.phrase, } ), extra={ @@ -750,7 +789,7 @@ def reject( self.send( Response( status_code=status_code, - reason=reason, + phrase=status_code.phrase, headers=self._with_to_tag( { key: value @@ -817,7 +856,7 @@ async def register( aor_rest = self.aor.partition(":")[2] if self.aor else "" user = aor_rest.partition("@")[0] if "@" in aor_rest else aor_rest headers = { - "Via": f"SIP/2.0/{'TLS' if self._is_tls else 'TCP'} {self.local_address[0]}:{self.local_address[1]};rport;branch={branch}", + "Via": f"SIP/2.0/{'TLS' if self._is_tls else 'TCP'} {_format_host(self.local_address[0])}:{self.local_address[1]};rport;branch={branch}", "From": self.aor, "To": self.aor, "Call-ID": self.call_id, diff --git a/voip/sip/types.py b/voip/sip/types.py index 1f7f5cc..ddeb5cb 100644 --- a/voip/sip/types.py +++ b/voip/sip/types.py @@ -1,9 +1,173 @@ from __future__ import annotations +import dataclasses import enum +import ipaddress import re -__all__ = ["CallerID", "DigestAlgorithm", "DigestQoP", "Status"] +__all__ = [ + "CallerID", + "DigestAlgorithm", + "DigestQoP", + "SipUri", + "SIPStatus", + "SIPMethod", +] + +import typing +import urllib.parse +from collections.abc import Iterator + + +@dataclasses.dataclass(slots=True, eq=True) +class SipUri: + """A parsed SIP or SIPS URI per [RFC 3261 §19.1]. + + Format: ``sip:user:password@host:port;uri-parameters?headers`` + + 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 without brackets. + + [RFC 3261 §19.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-19.1 + [RFC 2732]: https://datatracker.ietf.org/doc/html/rfc2732 + + Examples: + >>> SipUri.parse("sip:alice@example.com") + SipUri(scheme='sip', user='alice', host='example.com', ...) + >>> SipUri.parse("sips:+15551234567@carrier.com:5061") + SipUri(scheme='sips', user='+15551234567', host='carrier.com', port=5061, ...) + >>> SipUri.parse("sip:alice@[::1]:5060") + SipUri(scheme='sip', user='alice', host=IPv6Address('::1'), port=5060, ...) + + Args: + scheme: URI scheme — `sip` or `sips`. + host: Host as a bare string — no brackets for IPv6 addresses. + user: SIP user part (phone number or username). + port: Port number. 5061 for `sips:` and 5060 for `sip:`. + parameters: URI parameters as a mapping of name → value (`None` for flag parameters). + headers: SIP headers as a mapping of name → value. + + """ + + scheme: str + host: str | ipaddress.IPv6Address | ipaddress.IPv4Address + user: str | None = None + password: str | None = None + port: int | None = None + parameters: dict[str, str | None] = dataclasses.field(default_factory=dict) + headers: dict[str, str] = dataclasses.field(default_factory=dict) + + def __post_init__(self): + self.port = ( + self.port + if self.port is not None + else 5061 + if self.scheme == "sips" + else 5060 + ) + try: + self.host = ipaddress.ip_address(self.host) + except ValueError: + pass + + SIP_URL_PATTERN: typing.ClassVar[re.Pattern[str]] = re.compile( + r"^(?Psips?):" + r"((?P[^@;:]+)(?P:[^@;]*)?@)?" + r"(?P(\[[0-9a-fA-F:]+\]|[^;?:@\[\]]+))" + r"(?P:[0-9]+)?" + r"(?P;[^?]+)?" + r"(?P\?[^?]+)?$", + re.IGNORECASE, + ) + + @classmethod + def parse(cls, value: str) -> SipUri: + """ + Parse a SIP or SIPS URI string into a `SipUri` instance. + + Returns: + Parsed `SipUri` instance. + + Raises: + ValueError: When the URI is malformed (missing scheme, invalid + characters, unclosed IPv6 bracket, empty host, or invalid port). + """ + if match := cls.SIP_URL_PATTERN.fullmatch(value): + host = match.group("host") + if host.startswith("[") and host.endswith("]"): + host = host[1:-1] + host = urllib.parse.unquote(host) + try: + ipaddress.ip_address(host) + except ValueError: + pass # Not an IP address, treat as a regular hostname + + return cls( + scheme=match.group("scheme").lower(), + user=urllib.parse.unquote(match.group("user")) + if match.group("user") + else None, + host=host, + password=urllib.parse.unquote(match.group("password")[1:]) + if match.group("password") + else None, + port=int(match.group("port")[1:]) if match.group("port") else None, + parameters=dict(cls._parse_parameters(match.group("parameters"))) + if match.group("parameters") + else {}, + headers=dict(cls._parse_headers(match.group("headers")[1:])) + if match.group("headers") + else {}, + ) + raise ValueError(f"Invalid SIP URI: {value!r}") + + @classmethod + def _parse_parameters(cls, params: str) -> Iterator[tuple[str, str | None]]: + for part in params[1:].split(";"): + if "=" in part: + name, val = part.split("=", 1) + yield urllib.parse.unquote(name), urllib.parse.unquote(val) + elif part: + yield urllib.parse.unquote(part), None + + @classmethod + def _parse_headers(cls, headers: str) -> Iterator[tuple[str, str]]: + for part in headers.split("&"): + if "=" in part: + name, val = part.split("=", 1) + yield urllib.parse.unquote(name), urllib.parse.unquote(val) + elif part: + yield urllib.parse.unquote(part), "" + + def __str__(self) -> str: + parts = [f"{self.scheme}:"] + if self.user: + parts.append(urllib.parse.quote(self.user)) + if self.password: + parts.append(f":{urllib.parse.quote(self.password)}") + parts.append("@") + parts.append( + f"[{str(self.host)}]" + if isinstance(self.host, ipaddress.IPv6Address) + else str(self.host) + ) + parts.append(f":{self.port}") + for name, val in self.parameters.items(): + if val is not None: + parts.append(f";{urllib.parse.quote(name)}={urllib.parse.quote(val)}") + else: + parts.append(f";{urllib.parse.quote(name)}") + if self.headers: + parts.append("?") + parts.append( + "&".join( + f"{urllib.parse.quote(name)}={urllib.parse.quote(val)}" + for name, val in self.headers.items() + ) + ) + return "".join(parts) class CallerID(str): @@ -15,10 +179,10 @@ class CallerID(str): carrier domain — useful for log messages. Examples: - >>> str(CallerID('"015114455910" ;tag=abc')) - '"015114455910" ;tag=abc' - >>> repr(CallerID('"015114455910" ;tag=abc')) - '****5910@telefonica.de' + >>> str(CallerID('"08001234567" ;tag=abc')) + '"08001234567" ;tag=abc' + >>> repr(CallerID('"08001234567" ;tag=abc')) + '***4567@telefonica.de' >>> repr(CallerID('sip:alice@example.com')) '*lice@example.com' """ @@ -50,73 +214,304 @@ def tag(self) -> str | None: return m.group(1) if m else None def __repr__(self) -> str: - """Anonymized label: last 4 chars of user + carrier domain.""" user = self.display_name or self.user or "" host = self.host or "" masked = ("*" * max(0, len(user) - 4)) + user[-4:] if user else "****" return f"{masked}@{host}" if host else masked -Status = enum.IntEnum( - value="Status", - names={ - # 1xx Provisional - "Trying": 100, - "Ringing": 180, - "Call Is Being Forwarded": 181, - "Queued": 182, - "Session Progress": 183, - # 2xx Success - "OK": 200, - # 3xx Redirection - "Multiple Choices": 300, - "Moved Permanently": 301, - "Moved Temporarily": 302, - "Use Proxy": 305, - "Alternative Service": 380, - # 4xx Client Failure - "Bad Request": 400, - "Unauthorized": 401, - "Payment Required": 402, - "Forbidden": 403, - "Not Found": 404, - "Method Not Allowed": 405, - "Not Acceptable": 406, - "Proxy Authentication Required": 407, - "Request Timeout": 408, - "Gone": 410, - "Request Entity Too Large": 413, - "Request-URI Too Long": 414, - "Unsupported Media Type": 415, - "Unsupported URI Scheme": 416, - "Bad Extension": 420, - "Extension Required": 421, - "Interval Too Brief": 423, - "Temporarily Unavailable": 480, - "Call/Transaction Does Not Exist": 481, - "Loop Detected": 482, - "Too Many Hops": 483, - "Address Incomplete": 484, - "Ambiguous": 485, - "Busy Here": 486, - "Request Terminated": 487, - "Not Acceptable Here": 488, - "Request Pending": 491, - "Undecipherable": 493, - # 5xx Server Failure - "Server Internal Error": 500, - "Not Implemented": 501, - "Bad Gateway": 502, - "Service Unavailable": 503, - "Server Time-out": 504, - "Version Not Supported": 505, - "Message Too Large": 513, - # 6xx Global Failure - "Busy Everywhere": 600, - "Decline": 603, - "Does Not Exist Anywhere": 604, - }, -) +class SIPStatus(enum.IntEnum): + """ + SIP Status Codes based on [RFC 3261]. + + [RFC 3261]: https://datatracker.ietf.org/doc/html/rfc3261#section-21 + """ + + def __new__(cls, value, phrase, description=""): + obj = int.__new__(cls, value) + obj._value_ = value + obj.phrase = phrase + obj.description = description + return obj + + TRYING = ( + 100, + "Trying", + "The request is being processed. No final response is available yet.", + ) + RINGING = 180, "Ringing", "The called party is being alerted of the call." + CALL_IS_BEING_FORWARDED = ( + 181, + "Call Is Being Forwarded", + "The called party is being alerted of the call, but the call is not yet established.", + ) + QUEUED = ( + 182, + "Queued", + "The called party is being alerted of the call, but the call is not yet established.", + ) + SESSION_PROGRESS = ( + 183, + "Session Progress", + "The called party is being alerted of the call, but the call is not yet established.", + ) + + OK = 200, "OK", "The request has succeeded." + + MULTIPLE_CHOICES = ( + 300, + "Multiple Choices", + "The requested resource has multiple representations, each with its own specific location.", + ) + MOVED_PERMANENTLY = ( + 301, + "Moved Permanently", + "The requested resource has been assigned a new permanent URI and any future references to this resource ought to use one of the returned URIs.", + ) + MOVED_TEMPORARILY = ( + 302, + "Moved Temporarily", + "The requested resource is temporarily unavailable and the server is asking the client to try again later.", + ) + USE_PROXY = ( + 305, + "Use Proxy", + "The requested resource is available only through a proxy, the address for which is provided in the response.", + ) + ALTERNATIVE_SERVICE = ( + 380, + "Alternative Service", + "The server has fulfilled a request for the service indicated by the URI.", + ) + + BAD_REQUEST = ( + 400, + "Bad Request", + "The request has bad syntax or cannot be fulfilled due to bad syntax.", + ) + UNAUTHORIZED = 401, "Unauthorized", "The request requires user authentication." + PAYMENT_REQUIRED = 402, "Payment Required", "Further action is required." + FORBIDDEN = ( + 403, + "Forbidden", + "The server understood the request but refuses to fulfill it.", + ) + NOT_FOUND = 404, "Not Found", "The requested resource could not be found." + METHOD_NOT_ALLOWED = ( + 405, + "Method Not Allowed", + "The method specified in the Request-URI is not allowed for the resource identified by the request URI.", + ) + NOT_ACCEPTABLE = ( + 406, + "Not Acceptable", + "The server cannot produce a response matching the Accept headers.", + ) + PROXY_AUTHENTICATION_REQUIRED = ( + 407, + "Proxy Authentication Required", + "The client must authenticate itself with the proxy.", + ) + REQUEST_TIMEOUT = ( + 408, + "Request Timeout", + "The server timed out waiting for the request.", + ) + GONE = ( + 410, + "Gone", + "The requested resource is no longer available at the server and no longer exists.", + ) + REQUEST_ENTITY_TOO_LARGE = ( + 413, + "Request Entity Too Large", + "The server will not accept the request, because the entity of the request is too large.", + ) + REQUEST_URI_TOO_LONG = ( + 414, + "Request-URI Too Long", + "The server will not accept the request, because the Request-URI is too long.", + ) + UNSUPPORTED_MEDIA_TYPE = ( + 415, + "Unsupported Media Type", + "The server will not accept the request, because the media type of the request is unsupported.", + ) + UNSUPPORTED_URI_SCHEME = ( + 416, + "Unsupported URI Scheme", + "The server will not accept the request, because the URI scheme of the request is unsupported.", + ) + BAD_EXTENSION = ( + 420, + "Bad Extension", + "This status code indicates that the server does not recognize the value of any of the parameters that it needs to understand in the request.", + ) + EXTENSION_REQUIRED = ( + 421, + "Extension Required", + "This status code indicates that the server requires the client to identify itself (usually, using the Contact header field) before it will proceed with the request.", + ) + INTERVAL_TOO_BRIEF = ( + 423, + "Interval Too Brief", + "This status code indicates that the server is unwilling to process the request because either an individual header field, or all the header fields collectively, are too large.", + ) + TEMPORARILY_UNAVAILABLE = ( + 480, + "Temporarily Unavailable", + "This status code indicates that the server is currently unable to handle the request due to a temporary overloading or maintenance of the server.", + ) + CALL_TRANSACTION_DOES_NOT_EXIST = ( + 481, + "Call/Transaction Does Not Exist", + "This status code indicates that the server has received a final response for the transaction which it is still attempting to complete.", + ) + LOOP_DETECTED = ( + 482, + "Loop Detected", + "This status code indicates that the server has detected an infinite loop while processing the request.", + ) + TOO_MANY_HOPS = ( + 483, + "Too Many Hops", + "This status code indicates that the server has exceeded the maximum number of hops allowed in the request URI.", + ) + ADDRESS_INCOMPLETE = ( + 484, + "Address Incomplete", + "This status code indicates that the server has received a final response for the transaction which it is still attempting to complete, but has an invalid value for one or more of the header fields included in the request message.", + ) + AMBIGUOUS = ( + 485, + "Ambiguous", + "This status code indicates that the server cannot decide on a response to the request because multiple responses are possible.", + ) + BUSY_HERE = ( + 486, + "Busy Here", + "This status code indicates that the server is busy here.", + ) + REQUEST_TERMINATED = ( + 487, + "Request Terminated", + "This status code indicates that the server has received a final response for the transaction which it is still attempting to complete, but has received a termination request for that transaction from the client.", + ) + NOT_ACCEPTABLE_HERE = ( + 488, + "Not Acceptable Here", + "This status code indicates that the server is not able to produce a response which is acceptable to the client, according to the proactive negotiation header fields received in the request, and the server is unwilling to supply a default reason phrase.", + ) + REQUEST_PENDING = ( + 491, + "Request Pending", + "This status code indicates that the server has received a final response for the transaction which it is still attempting to complete, but has not yet delivered that response to the client.", + ) + UNDECIPHERABLE = ( + 493, + "Undecipherable", + "This status code indicates that the server was unable to decrypt a message after performing the necessary decryption(s).", + ) + + SERVER_INTERNAL_ERROR = ( + 500, + "Server Internal Error", + "The server encountered an unexpected condition which prevented it from fulfilling the request.", + ) + NOT_IMPLEMENTED = ( + 501, + "Not Implemented", + "The server does not support the functionality required to fulfill the request.", + ) + BAD_GATEWAY = ( + 502, + "Bad Gateway", + "The server, while acting as a gateway or proxy, received an invalid response from the upstream server it accessed in attempting to fulfill the request.", + ) + SERVICE_UNAVAILABLE = ( + 503, + "Service Unavailable", + "The server is currently unable to handle the request due to a temporary overloading or maintenance of the server.", + ) + SERVER_TIME_OUT = ( + 504, + "Server Time-out", + "The server, while acting as a gateway or proxy, did not receive a timely response from the upstream server specified by the URI (e.g., HTTP, FTP, LDAP) or some other auxiliary server (e.g., DNS) it needed to access in attempting to complete the request.", + ) + VERSION_NOT_SUPPORTED = ( + 505, + "Version Not Supported", + "The server does not support, or refuses to support, the protocol version that was used in the request message.", + ) + MESSAGE_TOO_LARGE = ( + 513, + "Message Too Large", + "The server is unwilling to process the request because its header fields are too large.", + ) + + BUSY_EVERYWHERE = ( + 600, + "Busy Everywhere", + "The server is not able to process the request because it is busy. For example, this error might be given if a server is overloaded with requests and is unable to process one of the requests.", + ) + DECLINE = 603, "Decline", "The call has been declined." + DOES_NOT_EXIST_ANYWHERE = ( + 604, + "Does Not Exist Anywhere", + "The server has received a final response for the transaction which it is still attempting to complete, but has received a termination request for that transaction from a server which it does not control.", + ) + NOT_ACCEPTABLE_ANYWHERE = ( + 606, + "Not Acceptable", + "The server is not able to produce a response which is acceptable to the client, according to the proactive negotiation header fields received in the request, and the server is unwilling to supply a default reason phrase.", + ) + + +class SIPMethod(enum.StrEnum): + """SIP methods and descriptions as defined in [RFC 3261]. + + Extended in [RFC 3262], [RFC 3265], [RFC 3515], [RFC 3428], and [RFC 3311]. + + [RFC 3261]: https://tools.ietf.org/html/rfc3261 + [RFC 3262]: https://tools.ietf.org/html/rfc3262 + [RFC 3265]: https://tools.ietf.org/html/rfc3265 + [RFC 3515]: https://tools.ietf.org/html/rfc3515 + [RFC 3428]: https://tools.ietf.org/html/rfc3428 + [RFC 3311]: https://tools.ietf.org/html/rfc3311 + """ + + def __new__(cls, value: str, description: str = "") -> SIPMethod: + obj = str.__new__(cls, value) + obj._value_ = value + obj.description = description + return obj + + INVITE = "INVITE", "The client is requesting to initiate a call." + ACK = "ACK", "The client is acknowledging the receipt of a previous request." + BYE = "BYE", "The client is requesting to end the call." + CANCEL = "CANCEL", "The client is requesting to cancel a previous request." + REGISTER = ( + "REGISTER", + "The client requests that the server register itself with the server's registration agent.", + ) + OPTIONS = ( + "OPTIONS", + "The client requests information about the server's capabilities or configuration.", + ) + NOTIFY = "NOTIFY", "The client is requesting to send a notification to the server." + SUBSCRIBE = "SUBSCRIBE", "The client is requesting to subscribe to a resource." + PUBLISH = "PUBLISH", "The client is requesting to publish a resource." + REFER = ( + "REFER", + "The client is requesting that the server refer the client to another resource.", + ) + PRACK = ( + "PRACK", + "The client is requesting to confirm the receipt of a previous request.", + ) + INFO = "INFO", "The client is requesting information about a session." + MESSAGE = "MESSAGE", "The client is requesting to send a message." + UPDATE = "UPDATE", "The client is requesting to update a resource." class DigestAlgorithm(enum.StrEnum): diff --git a/voip/stun.py b/voip/stun.py index f33bd9a..e931506 100644 --- a/voip/stun.py +++ b/voip/stun.py @@ -5,8 +5,8 @@ import asyncio import dataclasses import enum +import ipaddress import logging -import socket import struct import uuid @@ -31,6 +31,52 @@ class STUNAttributeType(enum.IntEnum): XOR_MAPPED_ADDRESS = 0x0020 +def _parse_address( + value: bytes, xor_key: bytes +) -> tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int] | None: + """Decode a STUN MAPPED-ADDRESS or XOR-MAPPED-ADDRESS attribute value. + + When *xor_key* is non-empty the port and address bytes are XORed with + the key per RFC 5389 §15.2; pass an empty byte string for plain + MAPPED-ADDRESS attributes. + + 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)``) + for XOR-MAPPED-ADDRESS, or empty bytes for plain MAPPED-ADDRESS. + + Returns: + ``(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 + if len(value) < 4: + return None + family = value[1] + raw_port = struct.unpack(">H", value[2:4])[0] + port = raw_port ^ (MAGIC_COOKIE >> 16) if xor_key else raw_port + match family: + case 0x01 if len(value) >= 8: # IPv4 + raw_ip = value[4:8] + ip_bytes = ( + bytes(a ^ b for a, b in zip(raw_ip, xor_key[:4], strict=False)) + if xor_key + else raw_ip + ) + return ipaddress.IPv4Address(ip_bytes), port + case 0x02 if len(value) >= 20: # IPv6 + raw_ip = value[4:20] + ip_bytes = ( + bytes(a ^ b for a, b in zip(raw_ip, xor_key, strict=False)) + if xor_key + else raw_ip + ) + return ipaddress.IPv6Address(ip_bytes), port + case _: + return None + + @dataclasses.dataclass(kw_only=True, slots=True) class STUNProtocol(asyncio.DatagramProtocol): """ @@ -74,7 +120,11 @@ def packet_received(self, data: bytes, addr: tuple[str, int]) -> None: def connection_made(self, transport: asyncio.DatagramTransport) -> None: self.transport = transport if self.stun_server_address is None: - self.stun_connection_made(transport, transport.get_extra_info("sockname")) + # IPv6 sockets return a 4-tuple (host, port, flowinfo, scope_id); + # we only need the first two elements. + sockname = transport.get_extra_info("sockname") + host, port = sockname[0], sockname[1] + self.stun_connection_made(transport, (ipaddress.ip_address(host), port)) else: self._stun_transaction_id = uuid.uuid4().bytes[:12] self._send_stun_request() @@ -82,7 +132,7 @@ def connection_made(self, transport: asyncio.DatagramTransport) -> None: def stun_connection_made( self, transport: asyncio.DatagramTransport, - addr: tuple[str, int], + addr: tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int], ) -> None: """Called when the socket is ready and the reachable address is known. @@ -177,30 +227,21 @@ def _parse_stun_response(self, data: bytes) -> None: # Clear transaction ID so duplicate responses are ignored. self._stun_transaction_id = b"" offset = 20 - xor_mapped: tuple[str, int] | None = None - mapped: tuple[str, int] | None = None + xor_mapped: tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int] | None = ( + None + ) + mapped: tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int] | None = None + xor_key = struct.pack(">I", MAGIC_COOKIE) + response_tid while offset + 4 <= len(data): attribute_type, attribute_len = struct.unpack( ">HH", data[offset : offset + 4] ) attribute_value = data[offset + 4 : offset + 4 + attribute_len] - if ( - attribute_type == STUNAttributeType.XOR_MAPPED_ADDRESS - and len(attribute_value) >= 8 - and attribute_value[1] == 0x01 # IPv4 - ): - port = struct.unpack(">H", attribute_value[2:4])[0] ^ ( - MAGIC_COOKIE >> 16 - ) - ip_int = struct.unpack(">I", attribute_value[4:8])[0] ^ MAGIC_COOKIE - xor_mapped = (socket.inet_ntoa(struct.pack(">I", ip_int)), port) - elif ( - attribute_type == STUNAttributeType.MAPPED_ADDRESS - and len(attribute_value) >= 8 - and attribute_value[1] == 0x01 # IPv4 - ): - port = struct.unpack(">H", attribute_value[2:4])[0] - mapped = (socket.inet_ntoa(attribute_value[4:8]), port) + match attribute_type: + case STUNAttributeType.XOR_MAPPED_ADDRESS: + xor_mapped = _parse_address(attribute_value, xor_key) + case STUNAttributeType.MAPPED_ADDRESS: + mapped = _parse_address(attribute_value, b"") offset += 4 + ((attribute_len + 3) & ~3) # 4-byte aligned result = xor_mapped or mapped if result: