From 8fd5f9488eb7cc5813f6026e2d4b1d1ba08279ed Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 10:24:24 +0200 Subject: [PATCH 1/9] Fix Route header --- voip/sip/transactions.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py index 42ff5b6..e486b2e 100644 --- a/voip/sip/transactions.py +++ b/voip/sip/transactions.py @@ -695,7 +695,7 @@ async def send( "To": str(target), "Contact": sip.contact, "Call-ID": dialog.call_id, - "Route": f"", + "Route": f"", "Allow": sip.allow_header, "Content-Type": "application/sdp", }, From b5f718cd3a45385e08421213b416df88d36e75e2 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 11:13:48 +0200 Subject: [PATCH 2/9] Add authentication to invite transactions --- tests/sip/test_transactions.py | 186 +++++++++++++++++ voip/sip/transactions.py | 370 +++++++++++++++++++++------------ 2 files changed, 422 insertions(+), 134 deletions(-) create mode 100644 tests/sip/test_transactions.py diff --git a/tests/sip/test_transactions.py b/tests/sip/test_transactions.py new file mode 100644 index 0000000..ce247da --- /dev/null +++ b/tests/sip/test_transactions.py @@ -0,0 +1,186 @@ +"""Tests for SIP transaction authentication (RFC 3261 §22, RFC 8760).""" + +import asyncio + +from voip.sip import messages +from voip.sip.dialog import Dialog +from voip.sip.protocol import SessionInitiationProtocol +from voip.sip.transactions import DigestAuthMixin, InviteTransaction +from voip.sip.types import SIPMethod, SIPStatus, SipURI + +from .conftest import CallFixture + + +def _last_request(sip: SessionInitiationProtocol) -> messages.Request: + """Return the most recently sent SIP request captured by the fake transport.""" + return messages.Message.parse(sip.transport.sent[-1]) # type: ignore[arg-type] + + +def _make_challenge_response( + request: messages.Request, *, status_code: SIPStatus, authenticate: str +) -> messages.Response: + """Build a 401/407 response echoing *request*'s dialog headers.""" + return messages.Message.parse( # type: ignore[return-value] + ( + f"SIP/2.0 {int(status_code)} {status_code.phrase}\r\n" + f"Via: {request.headers.getlist('Via')[0]}\r\n" + f"From: {request.headers['From']}\r\n" + f"To: {request.headers['To']}\r\n" + f"Call-ID: {request.headers['Call-ID']}\r\n" + f"CSeq: {request.headers['CSeq']}\r\n" + f"{'Proxy-Authenticate' if status_code == SIPStatus.PROXY_AUTHENTICATION_REQUIRED else 'WWW-Authenticate'}: {authenticate}\r\n" + f"Content-Length: 0\r\n" + "\r\n" + ).encode() + ) + + +_OK_SDP = ( + b"v=0\r\n" + b"o=- 1 1 IN IP4 192.0.2.1\r\n" + b"s=-\r\n" + b"c=IN IP4 192.0.2.1\r\n" + b"t=0 0\r\n" + b"m=audio 5004 RTP/AVP 0\r\n" + b"a=rtpmap:0 PCMU/8000\r\n" +) + + +def _make_ok_response(request: messages.Request) -> messages.Response: + """Build a 200 OK with an SDP answer echoing *request*'s dialog headers.""" + return messages.Message.parse( # type: ignore[return-value] + ( + "SIP/2.0 200 OK\r\n" + f"Via: {request.headers.getlist('Via')[0]}\r\n" + f"From: {request.headers['From']}\r\n" + f"To: {request.headers['To']};tag=remote-tag-1\r\n" + f"Call-ID: {request.headers['Call-ID']}\r\n" + f"CSeq: {request.headers['CSeq']}\r\n" + "Contact: \r\n" + "Content-Type: application/sdp\r\n" + "\r\n" + ).encode() + + _OK_SDP + ) + + +async def _complete_invite(sip: SessionInitiationProtocol, send_task: asyncio.Task): + """Answer the retried INVITE with 200 OK so the outbound call resolves.""" + retry = _last_request(sip) + sip.response_received(_make_ok_response(retry)) + await asyncio.wait_for(send_task, timeout=1) + + +class TestInviteAuth: + """Outbound INVITE must answer 401/407 challenges with credentials.""" + + async def test_invite_retries_with_authorization_on_401(self, sip): + """A 401 challenge triggers a retried INVITE carrying a digest Authorization.""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) # let the INVITE be sent + + invite = _last_request(sip) + assert invite.method == SIPMethod.INVITE + + challenge = 'Digest realm="example.com", nonce="abc123", algorithm=MD5' + sip.response_received( + _make_challenge_response( + invite, status_code=SIPStatus.UNAUTHORIZED, authenticate=challenge + ) + ) + + # The retry INVITE is the last sent request (after the ACK to the 401). + retry = _last_request(sip) + assert retry.method == SIPMethod.INVITE + assert retry.headers["CSeq"] == "2 INVITE" + authorization = str(retry.headers["Authorization"]) + assert authorization.startswith("Digest ") + assert 'algorithm="MD5"' in authorization + # SipURI canonicalises the `+` in the user part to `%2B`. + digest_uri = str(target) + assert f'uri="{digest_uri}"' in authorization + + # The digest response matches the spec computation for the challenge. + expected = DigestAuthMixin.digest_response( + username=sip.aor.user, + password=sip.aor.password, + realm="example.com", + nonce="abc123", + method=SIPMethod.INVITE, + uri=digest_uri, + algorithm="MD5", + ) + assert f'response="{expected}"' in authorization + + # Original transaction is still pending; its result is chained to the retry. + assert not send_task.done() + await _complete_invite(sip, send_task) + + async def test_invite_retries_with_proxy_authorization_on_407(self, sip): + """A 407 challenge yields a Proxy-Authorization header on the retry.""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + + invite = _last_request(sip) + challenge = 'Digest realm="example.com", nonce="xyz789", algorithm=SHA-256' + sip.response_received( + _make_challenge_response( + invite, + status_code=SIPStatus.PROXY_AUTHENTICATION_REQUIRED, + authenticate=challenge, + ) + ) + + retry = _last_request(sip) + assert "Proxy-Authorization" in retry.headers + assert "Authorization" not in retry.headers + assert 'algorithm="SHA-256"' in str(retry.headers["Proxy-Authorization"]) + + await _complete_invite(sip, send_task) + + +class TestRegistrationAuth: + """Registration digest behaviour preserved after the mixin extraction.""" + + async def test_register_retries_with_authorization_on_401(self, sip): + """A 401 to REGISTER triggers a credentialed retry with an incremented CSeq.""" + # The initial REGISTER was sent by connection_made(). + register = _last_request(sip) + assert register.method == SIPMethod.REGISTER + + challenge = 'Digest realm="example.com", nonce="reg-nonce", algorithm=MD5' + sip.response_received( + _make_challenge_response( + register, status_code=SIPStatus.UNAUTHORIZED, authenticate=challenge + ) + ) + + retry = _last_request(sip) + assert retry.method == SIPMethod.REGISTER + assert retry.headers["CSeq"] == "2 REGISTER" + assert str(retry.headers["Authorization"]).startswith("Digest ") + + # Digest URI for REGISTER remains the registrar host (unchanged behaviour). + expected = DigestAuthMixin.digest_response( + username=sip.aor.user, + password=sip.aor.password, + realm="example.com", + nonce="reg-nonce", + method=SIPMethod.REGISTER, + uri=str(sip.aor.host), + algorithm="MD5", + ) + assert f'response="{expected}"' in str(retry.headers["Authorization"]) + assert f'uri="{sip.aor.host}"' in str(retry.headers["Authorization"]) diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py index e486b2e..087c62e 100644 --- a/voip/sip/transactions.py +++ b/voip/sip/transactions.py @@ -141,9 +141,27 @@ def from_request( ) -@dataclasses.dataclass(kw_only=True, slots=True) -class RegistrationTransaction(Transaction): - """SIP REGISTER client transaction [RFC 3261 §10].""" +class DigestAuthMixin: + """Digest authentication for SIP client transactions (RFC 3261 §22, RFC 8760). + + Mix into a [Transaction][voip.sip.transactions.Transaction] subclass whose + requests may receive a `401 Unauthorized` or `407 Proxy Authentication + Required` challenge. The mixin supplies the shared digest machinery + (challenge parsing, response computation, retry chaining); the host + transaction only implements + [retry_with_auth][voip.sip.transactions.DigestAuthMixin.retry_with_auth] + to rebuild and resend its request carrying the computed credentials. + + A challenge is answered by calling + [handle_auth_challenge][voip.sip.transactions.DigestAuthMixin.handle_auth_challenge] + from the host's `response_received`. The caller must drop the original + transaction from the registry first (e.g. via `ack` or `drop_transaction`); + `handle_auth_challenge` then registers a fresh retry transaction and chains + its result back to the original via + [forward_result][voip.sip.transactions.DigestAuthMixin.forward_result]. + """ + + __slots__ = () #: Map from `DigestAlgorithm` to the hashlib name. DIGEST_HASH_NAME: typing.ClassVar[dict[str, str]] = { @@ -155,120 +173,85 @@ class RegistrationTransaction(Transaction): DigestAlgorithm.SHA_512_256_SESS: "sha512_256", } - authorization: str | None = None - proxy_authorization: str | None = None - cseq: int = 1 - - def __post_init__(self): - super().__post_init__() - from .dialog import Dialog + def digest_uri(self) -> str: + """Return the Request-URI used for the digest `uri` parameter. - self.dialog = self.dialog or Dialog(uac=self.sip.aor) - headers = ( - self.headers - | self.dialog.headers - | { - "Contact": self.sip.contact, - "Expires": "3600", - "Max-Forwards": "70", - "Supported": "outbound", - } - ) - if self.authorization is not None: - headers["Authorization"] = self.authorization - if self.proxy_authorization is not None: - headers["Proxy-Authorization"] = self.proxy_authorization - self.request = Request.from_dialog( - dialog=self.dialog, - method=SIPMethod.REGISTER, - uri=types.SipURI(host=self.sip.aor.host, scheme=self.sip.aor.scheme), - headers=headers, - ) + Defaults to the request's Request-URI (RFC 3261 §22.1). Override to + customise per transaction. + """ + return str(self.request.uri) # type: ignore[attr-defined] - self.sip.send(self.request) + def handle_auth_challenge(self, response: Response) -> bool: + """Answer a 401/407 by resending the request with digest credentials. - def response_received(self, response: Response) -> None: - """Handle a REGISTER response including digest auth challenges (RFC 3261 §22). + Parses the challenge, computes the digest response, and delegates to + [retry_with_auth][voip.sip.transactions.DigestAuthMixin.retry_with_auth] + to resend. The original transaction must already be dropped by the + caller. Args: - response: The parsed SIP response. + response: The `401`/`407` challenge response. + + Returns: + `True` if a retry transaction was started. """ - self.sip.drop_transaction(self) - match response.status_code: - case SIPStatus.OK: - logger.info("Registration successful") - self.set_result(self.dialog) - return - case SIPStatus.UNAUTHORIZED | SIPStatus.PROXY_AUTHENTICATION_REQUIRED: - logger.debug( - "Auth challenge received (%s), retrying with credentials", - response.status_code, - ) - 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", "") - nonce = params.get("nonce", "") - opaque = params.get("opaque") - algorithm = params.get("algorithm", DigestAlgorithm.MD5) - qop_options = params.get("qop", "") - qop = ( - DigestQoP.AUTH.value - if DigestQoP.AUTH.value in qop_options.split(",") - else None - ) - nc = "00000001" - cnonce = secrets.token_hex(8) if qop else None - digest = self.digest_response( - username=self.sip.aor.user, - password=self.sip.aor.password, - realm=realm, - nonce=nonce, - method=SIPMethod.REGISTER, - uri=self.sip.aor.host, - algorithm=algorithm, - qop=qop, - nc=nc, - cnonce=cnonce, - ) - auth_value = ( - f'Digest username="{self.sip.aor.user}", realm="{realm}", ' - f'nonce="{nonce}", uri="{self.sip.aor.host}", ' - f'response="{digest}", algorithm="{algorithm}"' - ) - if qop: - auth_value += f', qop={qop}, nc={nc}, cnonce="{cnonce}"' - if opaque: - auth_value += f', opaque="{opaque}"' - if is_proxy: - tx = RegistrationTransaction( - sip=self.sip, - dialog=self.dialog, - cseq=2, - method=self.method, - proxy_authorization=auth_value, - ) - else: - tx = RegistrationTransaction( - sip=self.sip, - dialog=self.dialog, - cseq=2, - method=self.method, - authorization=auth_value, - ) - self.sip.register_transaction(tx) - tx.add_done_callback(self.forward_result) - case _: - raise NotImplementedError( - f"Unknown SIP status code: {response.status_code}" - ) + 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", "") + nonce = params.get("nonce", "") + opaque = params.get("opaque") + algorithm = params.get("algorithm", DigestAlgorithm.MD5) + qop_options = params.get("qop", "") + qop = ( + DigestQoP.AUTH.value + if DigestQoP.AUTH.value in qop_options.split(",") + else None + ) + nc = "00000001" + cnonce = secrets.token_hex(8) if qop else None + uri = self.digest_uri() + digest = self.digest_response( + username=self.sip.aor.user, # type: ignore[attr-defined] + password=self.sip.aor.password, # type: ignore[attr-defined] + realm=realm, + nonce=nonce, + method=self.method, # type: ignore[attr-defined] + uri=uri, + algorithm=algorithm, + qop=qop, + nc=nc, + cnonce=cnonce, + ) + auth_value = ( + f'Digest username="{self.sip.aor.user}", realm="{realm}", ' # type: ignore[attr-defined] + f'nonce="{nonce}", uri="{uri}", ' + f'response="{digest}", algorithm="{algorithm}"' + ) + if qop: + auth_value += f', qop={qop}, nc={nc}, cnonce="{cnonce}"' + if opaque: + auth_value += f', opaque="{opaque}"' + return self.retry_with_auth(response, auth_value, is_proxy) + + def retry_with_auth( + self, response: Response, auth_value: str, is_proxy: bool + ) -> bool: + """Rebuild and resend this request carrying *auth_value*. + + Override in host transactions: construct a new transaction with a fresh + branch and incremented CSeq, attach `auth_value` as `Authorization` + (or `Proxy-Authorization` when *is_proxy*), register it, and chain its + result back to the original via + [forward_result][voip.sip.transactions.DigestAuthMixin.forward_result]. + + Returns: + `True` if a retry was started. + """ + raise NotImplementedError def forward_result(self, fut: asyncio.Future) -> None: - """Forward the result of *fut* to this transaction (used for auth retry chaining).""" + """Forward the result of *fut* to this transaction (auth retry chaining).""" if not self.done(): if fut.cancelled(): self.cancel() @@ -279,10 +262,10 @@ def forward_result(self, fut: asyncio.Future) -> None: @staticmethod def parse_auth_challenge(header: str) -> dict[str, str]: - """Parse Digest challenge parameters from a WWW-Authenticate/Proxy-Authenticate header. + """Parse Digest challenge parameters from a WWW/Proxy-Authenticate header. Args: - header: The raw `WWW-Authenticate` or `Proxy-Authenticate` header value. + header: The raw `WWW-Authenticate` or `Proxy-Authenticate` value. Returns: A dict mapping parameter names to their unquoted values. @@ -356,7 +339,86 @@ def h(data: str) -> str: @dataclasses.dataclass(kw_only=True, slots=True) -class InviteTransaction(Transaction): +class RegistrationTransaction(DigestAuthMixin, Transaction): + """SIP REGISTER client transaction [RFC 3261 §10].""" + + authorization: str | None = None + proxy_authorization: str | None = None + cseq: int = 1 + + def __post_init__(self): + super().__post_init__() + from .dialog import Dialog + + self.dialog = self.dialog or Dialog(uac=self.sip.aor) + headers = ( + self.headers + | self.dialog.headers + | { + "Contact": self.sip.contact, + "Expires": "3600", + "Max-Forwards": "70", + "Supported": "outbound", + } + ) + if self.authorization is not None: + headers["Authorization"] = self.authorization + if self.proxy_authorization is not None: + headers["Proxy-Authorization"] = self.proxy_authorization + self.request = Request.from_dialog( + dialog=self.dialog, + method=SIPMethod.REGISTER, + uri=types.SipURI(host=self.sip.aor.host, scheme=self.sip.aor.scheme), + headers=headers, + ) + + self.sip.send(self.request) + + def response_received(self, response: Response) -> None: + """Handle a REGISTER response including digest auth challenges (RFC 3261 §22). + + Args: + response: The parsed SIP response. + """ + self.sip.drop_transaction(self) + match response.status_code: + case SIPStatus.OK: + logger.info("Registration successful") + self.set_result(self.dialog) + case SIPStatus.UNAUTHORIZED | SIPStatus.PROXY_AUTHENTICATION_REQUIRED: + logger.debug( + "Auth challenge received (%s), retrying with credentials", + response.status_code, + ) + self.handle_auth_challenge(response) + case _: + raise NotImplementedError( + f"Unknown SIP status code: {response.status_code}" + ) + + def retry_with_auth( + self, response: Response, auth_value: str, is_proxy: bool + ) -> bool: + """Resend the REGISTER with credentials (RFC 3261 §22).""" + tx = RegistrationTransaction( + sip=self.sip, + dialog=self.dialog, + cseq=self.cseq + 1, + method=self.method, + proxy_authorization=auth_value if is_proxy else None, + authorization=None if is_proxy else auth_value, + ) + self.sip.register_transaction(tx) + tx.add_done_callback(self.forward_result) + return True + + def digest_uri(self) -> str: + """Use the registrar host as the digest URI for REGISTER.""" + return str(self.sip.aor.host) + + +@dataclasses.dataclass(kw_only=True, slots=True) +class InviteTransaction(DigestAuthMixin, Transaction): """SIP INVITE transaction for inbound and outbound calls [RFC 3261 §17]. Handles the SIP signaling state machine for a single INVITE dialog. The @@ -653,26 +715,38 @@ async def send( ) tx.pending_call_class = session_class tx.pending_call_kwargs = session_kwargs + tx.request = tx._build_invite_request(target) + sip.register_transaction(tx) + sip.send(tx.request) + try: + return await tx + except asyncio.CancelledError: + sip.drop_transaction(tx) + raise - rtp_public = sip.rtp.public_address.result() + def _build_invite_request(self, target: types.SipURI) -> messages.Request: + """Build the outbound INVITE request (with SDP offer) for *target*. + + Factored out of [send][voip.sip.transactions.InviteTransaction.send] + so that an auth retry can rebuild the request with a fresh branch and + incremented CSeq via [retry_with_auth][voip.sip.transactions.InviteTransaction.retry_with_auth]. + """ + rtp_public = self.sip.rtp.public_address.result() session_id = str(secrets.randbelow(2**32) + 1) + addrtype = "IP6" if isinstance(rtp_public[0], ipaddress.IPv6Address) else "IP4" sdp_offer = SessionDescription( origin=Origin( username="-", sess_id=session_id, sess_version=session_id, nettype="IN", - addrtype=( - "IP6" if isinstance(rtp_public[0], ipaddress.IPv6Address) else "IP4" - ), + addrtype=addrtype, unicast_address=str(rtp_public[0]), ), timings=[Timing(start_time=0, stop_time=0)], connection=ConnectionData( nettype="IN", - addrtype=( - "IP6" if isinstance(rtp_public[0], ipaddress.IPv6Address) else "IP4" - ), + addrtype=addrtype, connection_address=str(rtp_public[0]), ), media=[ @@ -680,37 +754,65 @@ async def send( media="audio", port=rtp_public[1], proto="RTP/AVP", - fmt=session_class.sdp_formats(), + fmt=self.pending_call_class.sdp_formats(), attributes=[Attribute(name="sendrecv")], ) ], ) - tx.request = Request( + return Request( method=SIPMethod.INVITE, uri=target, headers={ "Max-Forwards": "70", - **tx.headers, - "From": dialog.from_header, + **self.headers, + "From": self.dialog.from_header, "To": str(target), - "Contact": sip.contact, - "Call-ID": dialog.call_id, - "Route": f"", - "Allow": sip.allow_header, + "Contact": self.sip.contact, + "Call-ID": self.dialog.call_id, + # "Route": f"", + "Allow": self.sip.allow_header, "Content-Type": "application/sdp", }, body=sdp_offer, ) - sip.register_transaction(tx) - sip.send(tx.request) - try: - return await tx - except asyncio.CancelledError: - sip.drop_transaction(tx) - raise + + def retry_with_auth( + self, response: Response, auth_value: str, is_proxy: bool + ) -> bool: + """Resend the INVITE with credentials after a 401/407 challenge. + + The original transaction is dropped by the caller (via `ack`); this + builds a fresh INVITE (new branch, incremented CSeq) carrying the + computed credentials and chains its result back to the original. + """ + tx = type(self)( + sip=self.sip, + method=SIPMethod.INVITE, + cseq=self.cseq + 1, + dialog=self.dialog, + ) + tx.pending_call_class = self.pending_call_class + tx.pending_call_kwargs = self.pending_call_kwargs + tx.request = tx._build_invite_request(self.request.uri) + header = "Proxy-Authorization" if is_proxy else "Authorization" + tx.request.headers[header] = auth_value + self.sip.register_transaction(tx) + self.sip.send(tx.request) + tx.add_done_callback(self.forward_result) + return True def response_received(self, response: Response) -> None: """Dispatch responses to an outbound INVITE.""" + if response.status_code in ( + SIPStatus.UNAUTHORIZED, + SIPStatus.PROXY_AUTHENTICATION_REQUIRED, + ): + # Acknowledge the challenge (RFC 3261 §13.2.2.4); ack() drops this + # transaction, then retry_with_auth registers a fresh INVITE with + # credentials and chains its result back here via forward_result. + self.ack(response) + self.handle_auth_challenge(response) + return match response.status_code // 100: case 1: # trying/ringing return From a2f711cfa7768003728a640a200db986310ed8e5 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 14:51:27 +0200 Subject: [PATCH 3/9] Fix non-200 acknowledgement --- tests/sip/test_transactions.py | 108 +++++++++++++++++++++++++++++++++ voip/sip/transactions.py | 57 +++++++++++------ 2 files changed, 148 insertions(+), 17 deletions(-) diff --git a/tests/sip/test_transactions.py b/tests/sip/test_transactions.py index ce247da..7e5d886 100644 --- a/tests/sip/test_transactions.py +++ b/tests/sip/test_transactions.py @@ -71,6 +71,114 @@ async def _complete_invite(sip: SessionInitiationProtocol, send_task: asyncio.Ta await asyncio.wait_for(send_task, timeout=1) +def _last_ack(sip: SessionInitiationProtocol) -> messages.Request: + """Return the most recently sent ACK.""" + for raw in reversed(sip.transport.sent): # type: ignore[attr-defined] + request = messages.Message.parse(raw) + if isinstance(request, messages.Request) and request.method == SIPMethod.ACK: + return request + raise AssertionError("no ACK was sent") + + +class TestInviteAck: + """The ACK terminating an outbound INVITE must be well-formed.""" + + async def test_ack_for_non_2xx_reuses_invite_via_and_mirrors_route(self, sip): + """A non-2xx final ACK reuses the INVITE Via/branch and mirrors its Route. + + Per RFC 3261 §17.1.1.3 the transactional ACK must mirror the INVITE's + Route header values (not the response's Record-Route) so the proxy + holding the INVITE server transaction matches it by branch and absorbs + it, rather than loose-routing it onward and retransmitting. + """ + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + invite = _last_request(sip) + invite_via = invite.headers.getlist("Via")[0] + + decline = messages.Message.parse( + ( + "SIP/2.0 603 Decline\r\n" + f"Via: {invite_via}\r\n" + f"From: {invite.headers['From']}\r\n" + f"To: {invite.headers['To']};tag=decline-tag\r\n" + f"Call-ID: {invite.headers['Call-ID']}\r\n" + f"CSeq: {invite.headers['CSeq']}\r\n" + "Record-Route: \r\n" + "Record-Route: \r\n" + "Content-Length: 0\r\n" + "\r\n" + ).encode() + ) + sip.response_received(decline) + + ack = _last_ack(sip) + # Reuses the INVITE's Via (same branch + rport); no non-standard `alias`. + assert ack.headers.getlist("Via")[0] == invite_via + assert "alias" not in ack.headers.getlist("Via")[0] + # The INVITE carried no Route, so the transactional ACK carries none. + # Record-Route in the response must NOT leak into the ACK here. + assert invite.headers.getlist("Route") == [] + assert ack.headers.getlist("Route") == [] + # Non-2xx ACK mirrors the INVITE Request-URI, not a Contact. + assert str(ack.uri) == str(target) + # The original transaction completes without establishing a dialog. + await asyncio.wait_for(send_task, timeout=1) + + async def test_ack_for_2xx_carries_dialog_route_set(self, sip): + """A 2xx ACK opens a fresh transaction and follows the dialog route set. + + On 2xx the route set is the reversed Record-Route (RFC 3261 §12.1.2) + and the Request-URI is the Contact. + """ + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + invite = _last_request(sip) + invite_via = invite.headers.getlist("Via")[0] + + ok = messages.Message.parse( + ( + "SIP/2.0 200 OK\r\n" + f"Via: {invite_via}\r\n" + f"From: {invite.headers['From']}\r\n" + f"To: {invite.headers['To']};tag=ok-tag\r\n" + f"Call-ID: {invite.headers['Call-ID']}\r\n" + f"CSeq: {invite.headers['CSeq']}\r\n" + "Record-Route: \r\n" + "Record-Route: \r\n" + "Contact: \r\n" + "Content-Type: application/sdp\r\n" + "\r\n" + ).encode() + + _OK_SDP + ) + sip.response_received(ok) + + ack = _last_ack(sip) + # 2xx ACK uses a fresh Via branch (different from the INVITE's). + assert ack.headers.getlist("Via")[0] != invite_via + # Dialog route set is the reversed Record-Route (first hop first). + assert ack.headers.getlist("Route") == [ + "", + "", + ] + # 2xx ACK targets the Contact, not the INVITE Request-URI. + assert str(ack.uri) == "sip:bob@192.0.2.1:5060" + await asyncio.wait_for(send_task, timeout=1) + + class TestInviteAuth: """Outbound INVITE must answer 401/407 challenges with credentials.""" diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py index 087c62e..6153842 100644 --- a/voip/sip/transactions.py +++ b/voip/sip/transactions.py @@ -769,7 +769,6 @@ def _build_invite_request(self, target: types.SipURI) -> messages.Request: "To": str(target), "Contact": self.sip.contact, "Call-ID": self.dialog.call_id, - # "Route": f"", "Allow": self.sip.allow_header, "Content-Type": "application/sdp", }, @@ -885,35 +884,59 @@ def _start_call(self, response: Response) -> None: self.sip.rtp.send(b"\x00", remote_rtp_address) def ack(self, response: Response) -> None: - """Send an ACK after receiving a terminal response. - - For 2xx responses, establishes the dialog and registers it with the - protocol. + """Send an ACK after a terminal response to the INVITE. + + For 2xx responses the dialog is established (remote tag, parties, route + set, remote target) and registered with the protocol; the ACK opens a + fresh transaction with a new Via branch and is routed along the dialog + route set (reversed `Record-Route`, RFC 3261 §12.1.2). + + For non-2xx final responses the ACK is the transactional ACK of the + INVITE client transaction (RFC 3261 §17.1.1.3): it reuses the INVITE's + Via header (same branch) and mirrors the INVITE's `Route` header + values and Request-URI, so the proxy holding the INVITE server + transaction matches and absorbs it (stopping retransmissions) rather + than loose-routing it onward. It must NOT derive routes from the + response's `Record-Route` — that forms a dialog route set only on 2xx. """ - if response.status_code // 100 == 2: + if is_success := response.status_code // 100 == 2: + # RFC 3261 §12.1.2: UAC dialog route set is Record-Route reversed. + routes = list(reversed(list(response.headers.getlist("Record-Route")))) self.dialog.remote_tag = response.remote_tag self.dialog.local_party = str(response.headers["From"]) self.dialog.remote_party = str(response.headers["To"]) self.dialog.outbound_cseq = self.cseq + 1 - # RFC 3261 §12.1.2: UAC route set is Record-Route in reverse order. - self.dialog.route_set = list( - reversed(list(response.headers.getlist("Record-Route"))) + self.dialog.route_set = routes + self.dialog.remote_contact = response.headers.get("Contact") or str( + self.request.uri ) self.sip.register_dialog(self.dialog) + else: + # RFC 3261 §17.1.1.3: transactional ACK mirrors the INVITE's Route. + routes = list(self.request.headers.getlist("Route")) self.sip.drop_transaction(self) - ack_branch = f"{Transaction.branch_prefix}-{uuid.uuid4()}" + # Non-2xx ACK: same Request-URI as the INVITE. 2xx ACK: the remote + # target (Contact), falling back to the INVITE Request-URI. contact = response.headers.get("Contact") ack_uri = ( - contact.split(";")[0].strip("<>") if contact else str(self.request.uri) + contact.split(";")[0].strip("<>") + if (is_success and contact) + else str(self.request.uri) ) - self.dialog.remote_contact = ack_uri + if is_success: + via = ( + f"SIP/2.0/{self.sip.aor.transport}" + f" {self.sip.public_address};rport" + f";branch={Transaction.branch_prefix}-{uuid.uuid4()}" + ) + else: + # Reuse the INVITE's Via (same branch + rport) for transaction + # matching at the proxy (RFC 3261 §17.1.1.3). + via = self.request.headers.getlist("Via")[0] ack_headers: SIPHeaderDict = SIPHeaderDict( { - "Via": ( - f"SIP/2.0/{self.sip.aor.transport}" - f" {self.sip.public_address};rport;branch={ack_branch};alias" - ), + "Via": via, "Max-Forwards": "70", "From": response.headers["From"], "To": response.headers["To"], @@ -922,7 +945,7 @@ def ack(self, response: Response) -> None: "Content-Length": "0", } ) - for route in self.dialog.route_set: + for route in routes: ack_headers.add("Route", route) self.sip.send( Request( From 1e86f3a59cb195710595b19464d4b513d1d15624 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 17:46:11 +0200 Subject: [PATCH 4/9] Use RSTP by default --- tests/sip/conftest.py | 4 +- tests/sip/test_transactions.py | 283 ++++++++++++++++++++++++++++++++- tests/test_rtp.py | 66 ++++++++ tests/test_srtp.py | 66 ++++++++ voip/__main__.py | 16 +- voip/rtp.py | 16 +- voip/sip/dialog.py | 5 + voip/sip/messages.py | 7 +- voip/sip/protocol.py | 8 +- voip/sip/transactions.py | 155 ++++++++++++++++-- voip/srtp.py | 52 ++++++ 11 files changed, 646 insertions(+), 32 deletions(-) create mode 100644 tests/test_srtp.py diff --git a/tests/sip/conftest.py b/tests/sip/conftest.py index ac1c1d2..61c3e2f 100644 --- a/tests/sip/conftest.py +++ b/tests/sip/conftest.py @@ -49,11 +49,11 @@ class CallFixture(Session): @classmethod def negotiate_codec(cls, remote_media: MediaDescription) -> MediaDescription: - """Return the first format from the offered media.""" + """Return the first format from the offered media, mirroring its proto.""" return MediaDescription( media="audio", port=5004, - proto="RTP/AVP", + proto=remote_media.proto, fmt=remote_media.fmt[:1] or [RTPPayloadFormat.from_pt(0)], ) diff --git a/tests/sip/test_transactions.py b/tests/sip/test_transactions.py index 7e5d886..368b2c9 100644 --- a/tests/sip/test_transactions.py +++ b/tests/sip/test_transactions.py @@ -7,6 +7,7 @@ from voip.sip.protocol import SessionInitiationProtocol from voip.sip.transactions import DigestAuthMixin, InviteTransaction from voip.sip.types import SIPMethod, SIPStatus, SipURI +from voip.srtp import SRTPSession from .conftest import CallFixture @@ -71,6 +72,57 @@ async def _complete_invite(sip: SessionInitiationProtocol, send_task: asyncio.Ta await asyncio.wait_for(send_task, timeout=1) +def _make_status_response( + request: messages.Request, status_code: SIPStatus +) -> messages.Response: + """Build a non-2xx final response echoing *request*'s dialog headers (no body).""" + return messages.Message.parse( # type: ignore[return-value] + ( + f"SIP/2.0 {int(status_code)} {status_code.phrase}\r\n" + f"Via: {request.headers.getlist('Via')[0]}\r\n" + f"From: {request.headers['From']}\r\n" + f"To: {request.headers['To']};tag={status_code.name.lower()}\r\n" + f"Call-ID: {request.headers['Call-ID']}\r\n" + f"CSeq: {request.headers['CSeq']}\r\n" + "Content-Length: 0\r\n" + "\r\n" + ).encode() + ) + + +def _make_srtp_ok_response( + request: messages.Request, remote_session: SRTPSession +) -> messages.Response: + """Build a 200 OK with an `RTP/SAVP` SDP answer carrying *remote_session*'s SDES key.""" + sdp = ( + ( + b"v=0\r\n" + b"o=- 1 1 IN IP4 192.0.2.1\r\n" + b"s=-\r\n" + b"c=IN IP4 192.0.2.1\r\n" + b"t=0 0\r\n" + b"m=audio 5004 RTP/SAVP 0\r\n" + b"a=rtpmap:0 PCMU/8000\r\n" + ) + + f"a=crypto:{remote_session.sdes_attribute}\r\n".encode() + + b"a=sendrecv\r\n" + ) + return messages.Message.parse( # type: ignore[return-value] + ( + "SIP/2.0 200 OK\r\n" + f"Via: {request.headers.getlist('Via')[0]}\r\n" + f"From: {request.headers['From']}\r\n" + f"To: {request.headers['To']};tag=srtp-tag\r\n" + f"Call-ID: {request.headers['Call-ID']}\r\n" + f"CSeq: {request.headers['CSeq']}\r\n" + "Contact: \r\n" + "Content-Type: application/sdp\r\n" + "\r\n" + ).encode() + + sdp + ) + + def _last_ack(sip: SessionInitiationProtocol) -> messages.Request: """Return the most recently sent ACK.""" for raw in reversed(sip.transport.sent): # type: ignore[attr-defined] @@ -259,7 +311,7 @@ async def test_invite_retries_with_proxy_authorization_on_407(self, sip): await _complete_invite(sip, send_task) -class TestRegistrationAuth: +class TestRegisterAuth: """Registration digest behaviour preserved after the mixin extraction.""" async def test_register_retries_with_authorization_on_401(self, sip): @@ -292,3 +344,232 @@ async def test_register_retries_with_authorization_on_401(self, sip): ) assert f'response="{expected}"' in str(retry.headers["Authorization"]) assert f'uri="{sip.aor.host}"' in str(retry.headers["Authorization"]) + + +class TestInviteSrtp: + """Outbound INVITE offers SRTP (RTP/SAVP + SDES) and falls back to RTP.""" + + async def test_outbound_invite_offers_srtp(self, sip): + """The default outbound INVITE advertises `RTP/SAVP` with an SDES `a=crypto:`.""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + invite = _last_request(sip) + media = invite.body.media[0] + assert media.proto == "RTP/SAVP" + crypto = next(a for a in media.attributes if a.name == "crypto") + assert crypto.value.startswith("1 AES_CM_128_HMAC_SHA1_80 inline:") + + # Complete the call so the task resolves cleanly. + sip.response_received(_make_srtp_ok_response(invite, SRTPSession.generate())) + await asyncio.wait_for(send_task, timeout=1) + + async def test_outbound_prefer_srtp_false_offers_plain_rtp(self, sip): + """`prefer_srtp=False` offers `RTP/AVP` with no `a=crypto:`.""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, + target=target, + dialog=dialog, + session_class=CallFixture, + prefer_srtp=False, + ) + ) + await asyncio.sleep(0) + invite = _last_request(sip) + media = invite.body.media[0] + assert media.proto == "RTP/AVP" + assert not any(a.name == "crypto" for a in media.attributes) + + sip.response_received(_make_ok_response(invite)) + await asyncio.wait_for(send_task, timeout=1) + + async def test_outbound_srtp_answer_installs_send_and_recv_sessions(self, sip): + """A 200 OK with `RTP/SAVP` + remote crypto installs send and recv sessions.""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + invite = _last_request(sip) + # The offer's crypto session becomes the call's send session. + offer_crypto = next( + a for a in invite.body.media[0].attributes if a.name == "crypto" + ) + offer_session = SRTPSession.from_sdes(offer_crypto.value) + + remote_session = SRTPSession.generate() + sip.response_received(_make_srtp_ok_response(invite, remote_session)) + await asyncio.wait_for(send_task, timeout=1) + + session = dialog.session + # Send session keys our outbound media (matches the offer's SDES key). + assert session.srtp is not None + assert session.srtp.master_key == offer_session.master_key + # Recv session keys the remote's media (parsed from the answer crypto). + assert session.srtp_recv is not None + assert session.srtp_recv.master_key == remote_session.master_key + # A packet encrypted by the remote is decryptable with the recv session. + from voip.rtp import RTPPacket # noqa: PLC0415 + + packet = RTPPacket( + payload_type=0, sequence_number=1, timestamp=160, ssrc=42, payload=b"hello" + ) + decrypted = session.srtp_recv.decrypt(remote_session.encrypt(bytes(packet))) + assert decrypted is not None + assert RTPPacket.parse(decrypted).payload == b"hello" + + async def test_outbound_488_falls_back_to_rtp(self, sip): + """A 488 to the SRTP offer triggers a fresh `RTP/AVP` INVITE (auto fallback).""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + invite_srtp = _last_request(sip) + assert invite_srtp.body.media[0].proto == "RTP/SAVP" + + sip.response_received( + _make_status_response(invite_srtp, SIPStatus.NOT_ACCEPTABLE_HERE) + ) + + invite_rtp = _last_request(sip) + assert invite_rtp.method == SIPMethod.INVITE + assert invite_rtp.headers["CSeq"] == "2 INVITE" + assert invite_rtp.body.media[0].proto == "RTP/AVP" + assert not any(a.name == "crypto" for a in invite_rtp.body.media[0].attributes) + assert not send_task.done() + + # The fallback INVITE succeeds as plain RTP. + sip.response_received(_make_ok_response(invite_rtp)) + await asyncio.wait_for(send_task, timeout=1) + assert dialog.session.srtp is None + assert dialog.session.srtp_recv is None + + async def test_outbound_avp_answer_to_savp_offer_is_plain_rtp(self, sip): + """A 200 OK answering SAVP with AVP downgrades to plain RTP.""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + invite = _last_request(sip) + assert invite.body.media[0].proto == "RTP/SAVP" + + # Answer with plain RTP/AVP (no crypto) — a downgrade. + sip.response_received(_make_ok_response(invite)) + await asyncio.wait_for(send_task, timeout=1) + assert dialog.session.srtp is None + assert dialog.session.srtp_recv is None + + async def test_retry_with_rtp_preserves_authorization(self, sip): + """Auth credentials survive the SRTP→RTP fallback re-INVITE.""" + target = SipURI.parse("sip:+15551234567@example.com:5060") + dialog = Dialog(uac=sip.aor, sip=sip) + send_task = asyncio.create_task( + InviteTransaction.send( + sip=sip, target=target, dialog=dialog, session_class=CallFixture + ) + ) + await asyncio.sleep(0) + invite = _last_request(sip) + + # 401 challenge → credentialed SRTP retry. + challenge = 'Digest realm="example.com", nonce="abc", algorithm=MD5' + sip.response_received( + _make_challenge_response( + invite, status_code=SIPStatus.UNAUTHORIZED, authenticate=challenge + ) + ) + retry = _last_request(sip) + assert retry.body.media[0].proto == "RTP/SAVP" + assert "Authorization" in retry.headers + + # 488 to the SRTP retry → fallback to RTP, Authorization carried over. + sip.response_received( + _make_status_response(retry, SIPStatus.NOT_ACCEPTABLE_HERE) + ) + fallback = _last_request(sip) + assert fallback.body.media[0].proto == "RTP/AVP" + assert "Authorization" in fallback.headers + assert not any(a.name == "crypto" for a in fallback.body.media[0].attributes) + + sip.response_received(_make_ok_response(fallback)) + await asyncio.wait_for(send_task, timeout=1) + + async def test_inbound_srtp_offer_keys_recv_from_offer_crypto(self, sip): + """Answering an SRTP offer installs send (ours) + recv (offer's) sessions. + + The 200 OK carries our own fresh SDES key (send); the call's recv + session is parsed from the offer's `a=crypto:` so the caller's media + can be decrypted (RFC 4568 — SDES keys each direction independently). + """ + remote_session = SRTPSession.generate() + invite_bytes = ( + ( + b"INVITE sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKinbound1\r\n" + b"From: sip:bob@biloxi.com;tag=from-tag-srtp\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: inbound-srtp@biloxi.com\r\n" + b"CSeq: 1 INVITE\r\n" + b"Content-Type: application/sdp\r\n" + b"\r\n" + b"v=0\r\n" + b"o=- 1 1 IN IP4 192.0.2.1\r\n" + b"s=-\r\n" + b"c=IN IP4 192.0.2.1\r\n" + b"t=0 0\r\n" + b"m=audio 0 RTP/SAVP 0\r\n" + b"a=rtpmap:0 PCMU/8000\r\n" + ) + + f"a=crypto:{remote_session.sdes_attribute}\r\n".encode() + + b"a=sendrecv\r\n" + ) + request = messages.Message.parse(invite_bytes) + + captured: dict = {} + + class AnsweringDialog(Dialog): + """Answer inbound calls with the test CallFixture session.""" + + def call_received(self) -> None: + captured["dialog"] = self + self.answer(session_class=CallFixture) + + sip.dialog_class = AnsweringDialog + recv_task = asyncio.create_task( + InviteTransaction.receive(request=request, sip=sip) + ) + await asyncio.sleep(0) # call_received() runs and sends the 200 OK + + ok = messages.Message.parse(sip.transport.sent[-1]) + assert ok.status_code == SIPStatus.OK + ok_media = ok.body.media[0] + assert ok_media.proto == "RTP/SAVP" + ok_crypto = next(a for a in ok_media.attributes if a.name == "crypto") + send_session = SRTPSession.from_sdes(ok_crypto.value) + + session = captured["dialog"].session + assert session.srtp is not None + assert session.srtp.master_key == send_session.master_key + assert session.srtp_recv is not None + assert session.srtp_recv.master_key == remote_session.master_key + + recv_task.cancel() diff --git a/tests/test_rtp.py b/tests/test_rtp.py index 839e97d..f8681dd 100644 --- a/tests/test_rtp.py +++ b/tests/test_rtp.py @@ -479,6 +479,72 @@ def packet_received(self, packet: RTPPacket, addr) -> None: assert received == [] assert any("authentication failed" in r.message for r in caplog.records) + @pytest.mark.asyncio + async def test_srtp_recv_session_decrypts_inbound_media(self): + """The mux decrypts with `srtp_recv` (remote key), not the send `srtp`.""" + from voip.srtp import SRTPSession # noqa: PLC0415 + + received: list[RTPPacket] = [] + + @dataclasses.dataclass + class SRTPCapture(Session): + def packet_received(self, packet: RTPPacket, addr) -> None: + received.append(packet) + + mux = RealtimeTransportProtocol() + send_session = SRTPSession.generate() # keys our outbound media + recv_session = SRTPSession.generate() # keys the remote's inbound media + handler = SRTPCapture( + rtp=mux, + dialog=Dialog(), + media=make_media(), + srtp=send_session, + srtp_recv=recv_session, + caller=CallerID(""), + ) + mux.register_call(None, handler) + + # A packet encrypted with the remote (recv) key is delivered decrypted. + remote_packet = make_rtp_packet(payload=b"from-remote") + mux.datagram_received(recv_session.encrypt(remote_packet), ("1.2.3.4", 5004)) + assert len(received) == 1 + assert received[0].payload == b"from-remote" + + # A packet encrypted with the send key is NOT decryptable by the mux + # (it uses the recv key) — auth fails and the packet is discarded. + own_packet = make_rtp_packet(payload=b"our-own") + mux.datagram_received(send_session.encrypt(own_packet), ("1.2.3.4", 5004)) + assert len(received) == 1 # still only the recv-keyed packet + + @pytest.mark.asyncio + async def test_srtp_falls_back_to_send_session_for_decrypt(self): + """With only `srtp` set (legacy symmetric), the mux decrypts with it.""" + from voip.srtp import SRTPSession # noqa: PLC0415 + + received: list[RTPPacket] = [] + + @dataclasses.dataclass + class SRTPCapture(Session): + def packet_received(self, packet: RTPPacket, addr) -> None: + received.append(packet) + + mux = RealtimeTransportProtocol() + session = SRTPSession.generate() + handler = SRTPCapture( + rtp=mux, + dialog=Dialog(), + media=make_media(), + srtp=session, + caller=CallerID(""), + ) + mux.register_call(None, handler) + # srtp_recv unset → mux falls back to `srtp` for decryption. + mux.datagram_received( + session.encrypt(make_rtp_packet(payload=b"x")), ("1.2.3.4", 5004) + ) + assert len(received) == 1 + assert received[0].payload == b"x" + class TestSession: def test_caller__defaults_to_empty_string(self): diff --git a/tests/test_srtp.py b/tests/test_srtp.py new file mode 100644 index 0000000..92b4677 --- /dev/null +++ b/tests/test_srtp.py @@ -0,0 +1,66 @@ +"""Tests for the SRTP SDES key parser (voip.srtp, RFC 4568).""" + +import base64 + +import pytest +from voip.srtp import CIPHER_SUITE, SRTPSession + + +class TestFromSdes: + """`SRTPSession.from_sdes` parses an `a=crypto:` value into a session.""" + + def test_round_trips_generated_attribute(self): + """Parsing our own `sdes_attribute` reproduces the master key and salt.""" + session = SRTPSession.generate() + parsed = SRTPSession.from_sdes(session.sdes_attribute) + assert parsed.master_key == session.master_key + assert parsed.master_salt == session.master_salt + + def test_decrypts_media_encrypted_with_parsed_session(self): + """A session built from a remote `a=crypto:` decrypts that remote's SRTP.""" + from voip.rtp import RTPPacket # noqa: PLC0415 + + remote = SRTPSession.generate() + recv = SRTPSession.from_sdes(remote.sdes_attribute) + packet = RTPPacket( + payload_type=0, sequence_number=7, timestamp=160, ssrc=99, payload=b"hello" + ) + assert recv.decrypt(remote.encrypt(bytes(packet))) is not None + + def test_parses_tag_suite_and_inline_key(self): + """The tag and cipher suite are accepted; the inline key is base64.""" + session = SRTPSession.generate() + key_salt = base64.b64encode(session.master_key + session.master_salt).decode() + value = f"7 {CIPHER_SUITE} inline:{key_salt}" + parsed = SRTPSession.from_sdes(value) + assert parsed.master_key == session.master_key + assert parsed.master_salt == session.master_salt + + def test_ignores_lifetime_and_mki_trailers(self): + """`|lifetime` and `~MKI` suffixes after the inline key are ignored.""" + session = SRTPSession.generate() + key_salt = base64.b64encode(session.master_key + session.master_salt).decode() + value = f"1 {CIPHER_SUITE} inline:{key_salt}|2^32~12345" + parsed = SRTPSession.from_sdes(value) + assert parsed.master_key == session.master_key + assert parsed.master_salt == session.master_salt + + def test_rejects_unsupported_cipher_suite(self): + """An unknown cipher suite raises `ValueError` (only one suite is implemented).""" + session = SRTPSession.generate() + key_salt = base64.b64encode(session.master_key + session.master_salt).decode() + value = f"1 AES_CM_128_HMAC_SHA1_32 inline:{key_salt}" + with pytest.raises(ValueError, match="Unsupported SRTP cipher suite"): + SRTPSession.from_sdes(value) + + def test_rejects_malformed_value(self): + """A value missing the `inline:` parameter is rejected.""" + with pytest.raises(ValueError, match="Malformed SDES"): + SRTPSession.from_sdes("1 AES_CM_128_HMAC_SHA1_80 something:else") + + def test_rejects_short_inline_key(self): + """An inline key shorter than key+salt is rejected.""" + short = base64.b64encode(b"\x00" * 10).decode() + value = f"1 {CIPHER_SUITE} inline:{short}" + with pytest.raises(ValueError, match="too short"): + SRTPSession.from_sdes(value) diff --git a/voip/__main__.py b/voip/__main__.py index e6cdcd1..2ef9c23 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -172,8 +172,14 @@ def mcp(aor: SipURI, stun_server: NetworkAddress, no_verify_tls: bool, transport default=False, help="Disable TLS certificate verification (insecure; for testing only).", ) +@click.option( + "--no-srtp", + is_flag=True, + default=False, + help="Offer plain RTP only; do not offer SRTP or fall back from it.", +) @click.pass_context -def sip(ctx, aor, stun_server, no_verify_tls): +def sip(ctx, aor, stun_server, no_verify_tls, no_srtp): """Session Initiation Protocol (SIP).""" ctx.ensure_object(dict) ctx.obj.update( @@ -181,6 +187,7 @@ def sip(ctx, aor, stun_server, no_verify_tls): proxy_addr=aor.maddr, stun_server=stun_server, no_verify_tls=no_verify_tls, + prefer_srtp=not no_srtp, ) @@ -222,7 +229,9 @@ async def run(): stun_server=obj["stun_server"], ) await OutboundDialog(sip=protocol).dial( - parse_uri(dial, aor), session_class=EchoCall + parse_uri(dial, aor), + session_class=EchoCall, + prefer_srtp=obj["prefer_srtp"], ) await protocol.disconnected_event.wait() @@ -291,6 +300,7 @@ async def run(): await OutboundDialog(sip=protocol).dial( parse_uri(dial, aor), session_class=TranscribingCall, + prefer_srtp=obj["prefer_srtp"], stt_model=WhisperModel(stt_model), ) await protocol.disconnected_event.wait() @@ -417,6 +427,7 @@ async def run(): await OutboundDialog(sip=protocol).dial( parse_uri(dial, aor), session_class=AgentCallWithOutput, + prefer_srtp=obj["prefer_srtp"], stt_model=WhisperModel(stt_model), llm_model=llm_model, voice=voice, @@ -458,6 +469,7 @@ async def run(): await OutboundDialog(sip=protocol).dial( parse_uri(target, aor), session_class=SayCall, + prefer_srtp=obj["prefer_srtp"], text=prompt, voice=voice, ) diff --git a/voip/rtp.py b/voip/rtp.py index ccd837b..c334a33 100644 --- a/voip/rtp.py +++ b/voip/rtp.py @@ -106,7 +106,11 @@ class Session: dialog: SIP dialog state for this call leg. media: Negotiated SDP media description for this call leg. caller: Caller identifier as received in the SIP From header. - srtp: Optional SRTP session for encrypting and decrypting media. + srtp: Optional SRTP session used to encrypt outbound media (our SDES + send key). + srtp_recv: Optional SRTP session used to decrypt inbound media (the + remote's SDES key). Falls back to `srtp` when unset, preserving + the legacy symmetric single-session behaviour. """ rtp: RealtimeTransportProtocol @@ -114,6 +118,7 @@ class Session: media: MediaDescription caller: CallerID srtp: SRTPSession | None = None + srtp_recv: SRTPSession | None = None def packet_received(self, packet: RTPPacket, addr: NetworkAddress) -> None: """Handle a parsed RTP packet. Override in subclasses to process media. @@ -327,14 +332,17 @@ def packet_received(self, data: bytes, addr: NetworkAddress) -> None: When the matched handler carries an SRTP session, the packet is authenticated and decrypted before being forwarded; packets that fail - authentication are logged at WARNING level and discarded. + authentication are logged at WARNING level and discarded. Decryption + uses the handler's receive session (`srtp_recv`) when set, falling + back to `srtp` so a single symmetric session still works. """ handler = self.calls.get(addr) if handler is None: handler = self.calls.get(None) if handler is not None: - if handler.srtp is not None: - decrypted = handler.srtp.decrypt(data) + recv_session = handler.srtp_recv or handler.srtp + if recv_session is not None: + decrypted = recv_session.decrypt(data) if decrypted is None: logger.warning( "SRTP authentication failed for packet from %s:%s, discarding", diff --git a/voip/sip/dialog.py b/voip/sip/dialog.py index 67d9456..1a3eca3 100644 --- a/voip/sip/dialog.py +++ b/voip/sip/dialog.py @@ -181,6 +181,7 @@ async def dial( target: SipURI, *, session_class: type[Session], + prefer_srtp: bool = True, **session_kwargs: typing.Any, ) -> None: """ @@ -189,6 +190,9 @@ async def dial( Args: target: SIP or tel URI of the remote party (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`). session_class: Session subclass to create for this call. + prefer_srtp: Offer SRTP (`RTP/SAVP` + SDES) and fall back to plain + RTP if the far end rejects it. Defaults to `True`; pass + `False` to offer plain RTP only. **session_kwargs: Extra keyword arguments forwarded to `session_class`. [RFC 3261 §13.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-13.1 @@ -200,6 +204,7 @@ async def dial( target=target, dialog=self, session_class=session_class, + prefer_srtp=prefer_srtp, **session_kwargs, ) diff --git a/voip/sip/messages.py b/voip/sip/messages.py index 4eacbab..559ae26 100644 --- a/voip/sip/messages.py +++ b/voip/sip/messages.py @@ -110,6 +110,9 @@ def __bytes__(self) -> bytes: (self._first_line().encode(), bytes(self.headers), raw_body) ) + def __repr__(self): + return self._first_line() + @property def branch(self) -> str: """Branch parameter from the top Via header (RFC 3261 §20.42).""" @@ -135,7 +138,7 @@ def sequence(self) -> int: def _first_line(self) -> str: ... -@dataclasses.dataclass(kw_only=True) +@dataclasses.dataclass(slots=True, kw_only=True, repr=False) class Request(Message): """ A SIP request message [RFC 3261 §7.1]. @@ -158,7 +161,7 @@ def from_dialog(cls, *, dialog: Dialog, headers, **kwargs) -> Request: ) -@dataclasses.dataclass(kw_only=True) +@dataclasses.dataclass(slots=True, kw_only=True, repr=False) class Response(Message): """ A SIP response message [RFC 3261 §7.2]. diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index c092547..6ec48d6 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -21,7 +21,7 @@ from .transactions import ( ByeTransaction, InviteTransaction, - RegistrationTransaction, + RegisterTransaction, Transaction, ) from .types import ( @@ -41,7 +41,7 @@ "SIP", "SessionInitiationProtocol", "InviteTransaction", - "RegistrationTransaction", + "RegisterTransaction", ] @@ -293,7 +293,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: # type: ig ) try: loop = asyncio.get_running_loop() - tx = RegistrationTransaction(sip=self, method=SIPMethod.REGISTER) + tx = RegisterTransaction(sip=self, method=SIPMethod.REGISTER) self.register_transaction(tx) loop.create_task(self.handle_registration(tx)) if not isinstance(transport, asyncio.DatagramTransport): @@ -311,7 +311,7 @@ async def send_keepalive(self) -> None: logger.info("PING", extra={"addr": self.public_address}) self.transport.write(PING) - async def handle_registration(self, tx: RegistrationTransaction) -> None: + async def handle_registration(self, tx: RegisterTransaction) -> None: await tx self.on_registered() diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py index 6153842..881f7be 100644 --- a/voip/sip/transactions.py +++ b/voip/sip/transactions.py @@ -43,7 +43,7 @@ __all__ = [ "ByeTransaction", "InviteTransaction", - "RegistrationTransaction", + "RegisterTransaction", ] @@ -339,7 +339,7 @@ def h(data: str) -> str: @dataclasses.dataclass(kw_only=True, slots=True) -class RegistrationTransaction(DigestAuthMixin, Transaction): +class RegisterTransaction(DigestAuthMixin, Transaction): """SIP REGISTER client transaction [RFC 3261 §10].""" authorization: str | None = None @@ -392,15 +392,13 @@ def response_received(self, response: Response) -> None: ) self.handle_auth_challenge(response) case _: - raise NotImplementedError( - f"Unknown SIP status code: {response.status_code}" - ) + raise NotImplementedError(f"Unexpected SIP response: {response!r}") def retry_with_auth( self, response: Response, auth_value: str, is_proxy: bool ) -> bool: """Resend the REGISTER with credentials (RFC 3261 §22).""" - tx = RegistrationTransaction( + tx = RegisterTransaction( sip=self.sip, dialog=self.dialog, cseq=self.cseq + 1, @@ -447,6 +445,13 @@ class MySession(SessionInitiationProtocol): pending_call_kwargs: dict[str, typing.Any] = dataclasses.field( default_factory=dict, repr=False ) + #: Offer SRTP (`RTP/SAVP` + SDES `a=crypto:`) in the outbound INVITE and + #: fall back to plain RTP on rejection. Flipped to `False` for the RTP + #: fallback retry and by `prefer_srtp=False` on the public API. + offer_srtp: bool = True + #: SRTP send session generated for the current offer; reused as the + #: call's send `srtp` once the answer confirms SRTP. `None` for RTP. + srtp_offer: SRTPSession | None = dataclasses.field(default=None, repr=False) @classmethod async def receive( @@ -589,8 +594,29 @@ def call_received(self) -> None: fmt=[RTPPayloadFormat.from_pt(0)], ) - use_srtp = negotiated_media.proto == "RTP/SAVP" - srtp_session = SRTPSession.generate() if use_srtp else None + # SRTP when the offer is `RTP/SAVP` (or `SAVPF`) and carries an SDES + # `a=crypto:` key. We generate a fresh send session (its key goes into + # our 200-OK `a=crypto:`) and parse the offer's crypto to decrypt the + # caller's media — SDES keys each direction independently (RFC 4568). + is_srtp = negotiated_media.proto.startswith("RTP/SAVP") + srtp_send = SRTPSession.generate() if is_srtp else None + offer_crypto = ( + next( + ( + attr + for attr in remote_audio.attributes + if attr.name == "crypto" and attr.value + ), + None, + ) + if remote_audio is not None + else None + ) + srtp_recv = ( + SRTPSession.from_sdes(offer_crypto.value) + if is_srtp and offer_crypto is not None + else None + ) self.dialog.local_party = ( f"{self.request.headers['To']};tag={self.dialog.local_tag}" @@ -603,10 +629,12 @@ def call_received(self) -> None: rtp=self.sip.rtp, caller=caller, media=negotiated_media, - srtp=srtp_session, + srtp=srtp_send, + srtp_recv=srtp_recv, dialog=self.dialog, **session_kwargs, ) + self.dialog.session = session if remote_audio is not None and remote_audio.port != 0: media_connection = remote_audio.connection session_connection = ( @@ -630,9 +658,9 @@ def call_received(self) -> None: session_id = str(secrets.randbelow(2**32) + 1) rtp_public = self.sip.rtp.public_address.result() sdp_media_attributes = [Attribute(name="sendrecv")] - if srtp_session is not None: + if srtp_send is not None: sdp_media_attributes.append( - Attribute(name="crypto", value=srtp_session.sdes_attribute) + Attribute(name="crypto", value=srtp_send.sdes_attribute) ) self.send_response( Response.from_request( @@ -686,6 +714,7 @@ async def send( target: types.SipURI, dialog: Dialog, session_class: type[Session], + prefer_srtp: bool = True, **session_kwargs: typing.Any, ) -> Dialog: """Initiate an outgoing call to *target* [RFC 3261 §13.1]. @@ -695,6 +724,9 @@ async def send( target: SIP or tel URI of the callee (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`). dialog: The dialog to associate with this call. session_class: Session implementation that will be initialized for the call. + prefer_srtp: Offer SRTP (`RTP/SAVP` + SDES `a=crypto:`) and fall + back to plain RTP if the far end rejects it. Defaults to + `True`; pass `False` to offer plain RTP only. **session_kwargs: Additional keyword arguments forwarded to the call class constructor. @@ -715,6 +747,7 @@ async def send( ) tx.pending_call_class = session_class tx.pending_call_kwargs = session_kwargs + tx.offer_srtp = prefer_srtp tx.request = tx._build_invite_request(target) sip.register_transaction(tx) sip.send(tx.request) @@ -730,10 +763,25 @@ def _build_invite_request(self, target: types.SipURI) -> messages.Request: Factored out of [send][voip.sip.transactions.InviteTransaction.send] so that an auth retry can rebuild the request with a fresh branch and incremented CSeq via [retry_with_auth][voip.sip.transactions.InviteTransaction.retry_with_auth]. + + When [offer_srtp][voip.sip.transactions.InviteTransaction.offer_srtp] + is set (default) the offer advertises `RTP/SAVP` with an SDES + `a=crypto:` attribute; otherwise it advertises plain `RTP/AVP`. """ rtp_public = self.sip.rtp.public_address.result() session_id = str(secrets.randbelow(2**32) + 1) addrtype = "IP6" if isinstance(rtp_public[0], ipaddress.IPv6Address) else "IP4" + if self.offer_srtp: + proto = "RTP/SAVP" + self.srtp_offer = SRTPSession.generate() + attributes = [ + Attribute(name="sendrecv"), + Attribute(name="crypto", value=self.srtp_offer.sdes_attribute), + ] + else: + proto = "RTP/AVP" + self.srtp_offer = None + attributes = [Attribute(name="sendrecv")] sdp_offer = SessionDescription( origin=Origin( username="-", @@ -753,9 +801,9 @@ def _build_invite_request(self, target: types.SipURI) -> messages.Request: MediaDescription( media="audio", port=rtp_public[1], - proto="RTP/AVP", + proto=proto, fmt=self.pending_call_class.sdp_formats(), - attributes=[Attribute(name="sendrecv")], + attributes=attributes, ) ], ) @@ -782,7 +830,40 @@ def retry_with_auth( The original transaction is dropped by the caller (via `ack`); this builds a fresh INVITE (new branch, incremented CSeq) carrying the - computed credentials and chains its result back to the original. + computed credentials and chains its result back to the original. The + SRTP/RTP offer mode is preserved across the retry. + """ + header = "Proxy-Authorization" if is_proxy else "Authorization" + return self._retry_invite( + offer_srtp=self.offer_srtp, auth_headers={header: auth_value} + ) + + def retry_with_rtp(self, response: Response) -> bool: + """Fall back from SRTP to plain RTP after the far end rejects SRTP. + + The original (SRTP-offering) transaction is dropped by the caller (via + `ack`); this builds a fresh INVITE (new branch, incremented CSeq) + advertising `RTP/AVP` with no `a=crypto:`, carrying over any + `Authorization`/`Proxy-Authorization` already obtained, and chains its + result back to the original. Triggered for `488`, `606` and `415` + responses to an SRTP offer. + """ + auth_headers = { + header: self.request.headers[header] + for header in ("Authorization", "Proxy-Authorization") + if header in self.request.headers + } + return self._retry_invite(offer_srtp=False, auth_headers=auth_headers) + + def _retry_invite(self, *, offer_srtp: bool, auth_headers: dict[str, str]) -> bool: + """Build, register and send a fresh INVITE retry, chaining its result. + + Shared by [retry_with_auth][voip.sip.transactions.InviteTransaction.retry_with_auth] + (which preserves the current SRTP/RTP mode) and + [retry_with_rtp][voip.sip.transactions.InviteTransaction.retry_with_rtp] + (which switches to plain RTP). The new transaction carries *auth_headers* + (e.g. `Authorization`/`Proxy-Authorization`) so credentials obtained on a + prior attempt are not lost when falling back. """ tx = type(self)( sip=self.sip, @@ -792,9 +873,10 @@ def retry_with_auth( ) tx.pending_call_class = self.pending_call_class tx.pending_call_kwargs = self.pending_call_kwargs + tx.offer_srtp = offer_srtp tx.request = tx._build_invite_request(self.request.uri) - header = "Proxy-Authorization" if is_proxy else "Authorization" - tx.request.headers[header] = auth_value + for header, value in auth_headers.items(): + tx.request.headers[header] = value self.sip.register_transaction(tx) self.sip.send(tx.request) tx.add_done_callback(self.forward_result) @@ -817,6 +899,15 @@ def response_received(self, response: Response) -> None: return case 2: # OK self._start_call(response) + # SRTP offer rejected as "not acceptable" → fall back to plain RTP. + if self.offer_srtp and response.status_code in ( + SIPStatus.NOT_ACCEPTABLE_HERE, # 488 + SIPStatus.NOT_ACCEPTABLE_ANYWHERE, # 606 + SIPStatus.UNSUPPORTED_MEDIA_TYPE, # 415 + ): + self.ack(response) + self.retry_with_rtp(response) + return self.ack(response) self.complete() @@ -852,12 +943,42 @@ def _start_call(self, response: Response) -> None: fmt=[RTPPayloadFormat.from_pt(0)], ) + # The answer confirms SRTP only when it mirrors our `RTP/SAVP` offer + # and supplies its own SDES `a=crypto:` key. Our generated offer + # session keys the outbound (send) media; the remote crypto keys the + # inbound (recv) media. A downgrade to `RTP/AVP` yields plain RTP. + remote_crypto = ( + next( + ( + attr + for attr in remote_audio.attributes + if attr.name == "crypto" and attr.value + ), + None, + ) + if remote_audio is not None + else None + ) + is_srtp = ( + remote_audio is not None + and remote_audio.proto.startswith("RTP/SAVP") + and remote_crypto is not None + and self.srtp_offer is not None + ) + srtp_send = self.srtp_offer if is_srtp else None + srtp_recv = ( + SRTPSession.from_sdes(remote_crypto.value) # type: ignore[arg-type] + if is_srtp and remote_crypto is not None + else None + ) + if self.pending_call_class is not None: self.dialog.session = self.pending_call_class( rtp=self.sip.rtp, caller=CallerID(str(self.sip.aor)), media=negotiated_media, - srtp=None, + srtp=srtp_send, + srtp_recv=srtp_recv, dialog=self.dialog, **self.pending_call_kwargs, ) diff --git a/voip/srtp.py b/voip/srtp.py index 6e92c0d..5dae89b 100644 --- a/voip/srtp.py +++ b/voip/srtp.py @@ -99,6 +99,58 @@ def generate(cls) -> SRTPSession: master_salt=os.urandom(_SALT_SIZE), ) + @classmethod + def from_sdes(cls, value: str) -> SRTPSession: + """Build an SRTP session from a remote SDP `a=crypto:` value (RFC 4568). + + Parses the SDES crypto attribute format:: + + inline:[|][~] + + Only `AES_CM_128_HMAC_SHA1_80` is implemented; any other cipher suite + raises `ValueError`. The `inline:` keying material is base64-decoded + into the 16-byte master key followed by the 14-byte master salt; + trailing lifetime/MKI fields are ignored. + + Args: + value: The `a=crypto:` attribute value (without the `a=crypto:` + prefix), as received from the remote SDP. + + Returns: + An [`SRTPSession`][voip.srtp.SRTPSession] keyed with the remote + master key and salt — use it as the receive session to decrypt + the remote's media. + + Raises: + ValueError: When the value is malformed or names an unsupported + cipher suite. + """ + parts = value.split() + if len(parts) < 3 or not parts[2].startswith("inline:"): + raise ValueError(f"Malformed SDES crypto attribute: {value!r}") + _tag, suite, inline = parts[0], parts[1], parts[2] + # The inline parameter may carry trailing |lifetime or ~MKI suffixes. + key_material_b64, _, _trailing = inline.removeprefix("inline:").partition("|") + key_material_b64 = key_material_b64.split("~", 1)[0] + if suite != CIPHER_SUITE: + raise ValueError( + f"Unsupported SRTP cipher suite {suite!r}; " + f"only {CIPHER_SUITE!r} is implemented" + ) + try: + key_material = base64.b64decode(key_material_b64, validate=True) + except (ValueError, base64.binascii.Error) as exc: # noqa: PERF203 + raise ValueError(f"Malformed SDES inline key: {inline!r}") from exc + if len(key_material) < _KEY_SIZE + _SALT_SIZE: + raise ValueError( + f"SDES inline key too short ({len(key_material)} bytes); " + f"expected at least {_KEY_SIZE + _SALT_SIZE}" + ) + return cls( + master_key=key_material[:_KEY_SIZE], + master_salt=key_material[_KEY_SIZE : _KEY_SIZE + _SALT_SIZE], + ) + @property def sdes_attribute(self) -> str: """SDP `a=crypto:` attribute value for SDES key exchange (RFC 4568). From 18ebc6e845f93cc57b54b30ec63421d65764e074 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 18:46:24 +0200 Subject: [PATCH 5/9] Condense --- tests/sip/test_transactions.py | 22 ---------------------- voip/__main__.py | 13 +------------ voip/sip/dialog.py | 8 +++----- voip/sip/transactions.py | 12 +++++------- 4 files changed, 9 insertions(+), 46 deletions(-) diff --git a/tests/sip/test_transactions.py b/tests/sip/test_transactions.py index 368b2c9..dffd423 100644 --- a/tests/sip/test_transactions.py +++ b/tests/sip/test_transactions.py @@ -369,28 +369,6 @@ async def test_outbound_invite_offers_srtp(self, sip): sip.response_received(_make_srtp_ok_response(invite, SRTPSession.generate())) await asyncio.wait_for(send_task, timeout=1) - async def test_outbound_prefer_srtp_false_offers_plain_rtp(self, sip): - """`prefer_srtp=False` offers `RTP/AVP` with no `a=crypto:`.""" - target = SipURI.parse("sip:+15551234567@example.com:5060") - dialog = Dialog(uac=sip.aor, sip=sip) - send_task = asyncio.create_task( - InviteTransaction.send( - sip=sip, - target=target, - dialog=dialog, - session_class=CallFixture, - prefer_srtp=False, - ) - ) - await asyncio.sleep(0) - invite = _last_request(sip) - media = invite.body.media[0] - assert media.proto == "RTP/AVP" - assert not any(a.name == "crypto" for a in media.attributes) - - sip.response_received(_make_ok_response(invite)) - await asyncio.wait_for(send_task, timeout=1) - async def test_outbound_srtp_answer_installs_send_and_recv_sessions(self, sip): """A 200 OK with `RTP/SAVP` + remote crypto installs send and recv sessions.""" target = SipURI.parse("sip:+15551234567@example.com:5060") diff --git a/voip/__main__.py b/voip/__main__.py index 2ef9c23..c98c195 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -172,14 +172,8 @@ def mcp(aor: SipURI, stun_server: NetworkAddress, no_verify_tls: bool, transport default=False, help="Disable TLS certificate verification (insecure; for testing only).", ) -@click.option( - "--no-srtp", - is_flag=True, - default=False, - help="Offer plain RTP only; do not offer SRTP or fall back from it.", -) @click.pass_context -def sip(ctx, aor, stun_server, no_verify_tls, no_srtp): +def sip(ctx, aor, stun_server, no_verify_tls): """Session Initiation Protocol (SIP).""" ctx.ensure_object(dict) ctx.obj.update( @@ -187,7 +181,6 @@ def sip(ctx, aor, stun_server, no_verify_tls, no_srtp): proxy_addr=aor.maddr, stun_server=stun_server, no_verify_tls=no_verify_tls, - prefer_srtp=not no_srtp, ) @@ -231,7 +224,6 @@ async def run(): await OutboundDialog(sip=protocol).dial( parse_uri(dial, aor), session_class=EchoCall, - prefer_srtp=obj["prefer_srtp"], ) await protocol.disconnected_event.wait() @@ -300,7 +292,6 @@ async def run(): await OutboundDialog(sip=protocol).dial( parse_uri(dial, aor), session_class=TranscribingCall, - prefer_srtp=obj["prefer_srtp"], stt_model=WhisperModel(stt_model), ) await protocol.disconnected_event.wait() @@ -427,7 +418,6 @@ async def run(): await OutboundDialog(sip=protocol).dial( parse_uri(dial, aor), session_class=AgentCallWithOutput, - prefer_srtp=obj["prefer_srtp"], stt_model=WhisperModel(stt_model), llm_model=llm_model, voice=voice, @@ -469,7 +459,6 @@ async def run(): await OutboundDialog(sip=protocol).dial( parse_uri(target, aor), session_class=SayCall, - prefer_srtp=obj["prefer_srtp"], text=prompt, voice=voice, ) diff --git a/voip/sip/dialog.py b/voip/sip/dialog.py index 1a3eca3..bb97490 100644 --- a/voip/sip/dialog.py +++ b/voip/sip/dialog.py @@ -181,18 +181,17 @@ async def dial( target: SipURI, *, session_class: type[Session], - prefer_srtp: bool = True, **session_kwargs: typing.Any, ) -> None: """ Initiate an outbound call to *target*. + Offers SRTP (`RTP/SAVP` + SDES) and falls back to plain RTP if the far + end rejects it (488/606/415) or answers with `RTP/AVP`. + Args: target: SIP or tel URI of the remote party (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`). session_class: Session subclass to create for this call. - prefer_srtp: Offer SRTP (`RTP/SAVP` + SDES) and fall back to plain - RTP if the far end rejects it. Defaults to `True`; pass - `False` to offer plain RTP only. **session_kwargs: Extra keyword arguments forwarded to `session_class`. [RFC 3261 §13.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-13.1 @@ -204,7 +203,6 @@ async def dial( target=target, dialog=self, session_class=session_class, - prefer_srtp=prefer_srtp, **session_kwargs, ) diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py index 881f7be..c9a9001 100644 --- a/voip/sip/transactions.py +++ b/voip/sip/transactions.py @@ -446,8 +446,8 @@ class MySession(SessionInitiationProtocol): default_factory=dict, repr=False ) #: Offer SRTP (`RTP/SAVP` + SDES `a=crypto:`) in the outbound INVITE and - #: fall back to plain RTP on rejection. Flipped to `False` for the RTP - #: fallback retry and by `prefer_srtp=False` on the public API. + #: fall back to plain RTP on rejection. Defaults to `True`; flipped to + #: `False` internally by the RTP fallback retry. offer_srtp: bool = True #: SRTP send session generated for the current offer; reused as the #: call's send `srtp` once the answer confirms SRTP. `None` for RTP. @@ -714,19 +714,18 @@ async def send( target: types.SipURI, dialog: Dialog, session_class: type[Session], - prefer_srtp: bool = True, **session_kwargs: typing.Any, ) -> Dialog: """Initiate an outgoing call to *target* [RFC 3261 §13.1]. + Offers SRTP (`RTP/SAVP` + SDES `a=crypto:`) and falls back to plain RTP + if the far end rejects it (488/606/415) or answers with `RTP/AVP`. + Args: sip: The SIP session to send from. target: SIP or tel URI of the callee (e.g. `"sip:+15551234567@carrier.com"` or `"tel:+15551234567"`). dialog: The dialog to associate with this call. session_class: Session implementation that will be initialized for the call. - prefer_srtp: Offer SRTP (`RTP/SAVP` + SDES `a=crypto:`) and fall - back to plain RTP if the far end rejects it. Defaults to - `True`; pass `False` to offer plain RTP only. **session_kwargs: Additional keyword arguments forwarded to the call class constructor. @@ -747,7 +746,6 @@ async def send( ) tx.pending_call_class = session_class tx.pending_call_kwargs = session_kwargs - tx.offer_srtp = prefer_srtp tx.request = tx._build_invite_request(target) sip.register_transaction(tx) sip.send(tx.request) From e515c9f69e66a1393c35520282b8d1c13e8bd45a Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 19:28:16 +0200 Subject: [PATCH 6/9] Drop space --- voip/rtp.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/voip/rtp.py b/voip/rtp.py index c334a33..756b1e8 100644 --- a/voip/rtp.py +++ b/voip/rtp.py @@ -332,7 +332,7 @@ def packet_received(self, data: bytes, addr: NetworkAddress) -> None: When the matched handler carries an SRTP session, the packet is authenticated and decrypted before being forwarded; packets that fail - authentication are logged at WARNING level and discarded. Decryption + authentication are logged at WARNING level and discarded. Decryption uses the handler's receive session (`srtp_recv`) when set, falling back to `srtp` so a single symmetric session still works. """ From 0995f9326b4706d38b5aef3350e206d30778684a Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 20:28:48 +0200 Subject: [PATCH 7/9] Improve typing --- voip/sip/messages.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/voip/sip/messages.py b/voip/sip/messages.py index 559ae26..f2fc9f6 100644 --- a/voip/sip/messages.py +++ b/voip/sip/messages.py @@ -146,8 +146,8 @@ class Request(Message): [RFC 3261 §7.1]: https://datatracker.ietf.org/doc/html/rfc3261#section-7.1 """ - method: SIPMethod | str - uri: SipURI | str + method: SIPMethod + uri: SipURI def _first_line(self) -> str: return f"{self.method} {self.uri} {self.version}" From 4ed82de7da99eab03b0d53c4270f62627e9b8890 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 20:39:15 +0200 Subject: [PATCH 8/9] Small fixes --- tests/test_srtp.py | 15 +++++++++++++++ voip/srtp.py | 10 +++++++++- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/tests/test_srtp.py b/tests/test_srtp.py index 92b4677..65faf81 100644 --- a/tests/test_srtp.py +++ b/tests/test_srtp.py @@ -45,6 +45,21 @@ def test_ignores_lifetime_and_mki_trailers(self): assert parsed.master_key == session.master_key assert parsed.master_salt == session.master_salt + def test_from_sdes__uses_first_of_multiple_inline_keys(self): + """RFC 4568 §5.1.1: several inline key-params joined by ';' may be present.""" + session = SRTPSession.generate() + backup = SRTPSession.generate() + active_key_salt = base64.b64encode( + session.master_key + session.master_salt + ).decode() + backup_key_salt = base64.b64encode( + backup.master_key + backup.master_salt + ).decode() + value = f"1 {CIPHER_SUITE} inline:{active_key_salt};inline:{backup_key_salt}" + parsed = SRTPSession.from_sdes(value) + assert parsed.master_key == session.master_key + assert parsed.master_salt == session.master_salt + def test_rejects_unsupported_cipher_suite(self): """An unknown cipher suite raises `ValueError` (only one suite is implemented).""" session = SRTPSession.generate() diff --git a/voip/srtp.py b/voip/srtp.py index 5dae89b..371c862 100644 --- a/voip/srtp.py +++ b/voip/srtp.py @@ -112,6 +112,11 @@ def from_sdes(cls, value: str) -> SRTPSession: into the 16-byte master key followed by the 14-byte master salt; trailing lifetime/MKI fields are ignored. + RFC 4568 §5.1.1 allows several `inline:` key-params separated by `;` + within the key-params field (e.g. ``inline:KEY1;inline:KEY2``). The + first key-param is the *active* key used for SRTP processing, while + any further key-params are pre-rolled backup keys and are ignored here. + Args: value: The `a=crypto:` attribute value (without the `a=crypto:` prefix), as received from the remote SDP. @@ -128,7 +133,10 @@ def from_sdes(cls, value: str) -> SRTPSession: parts = value.split() if len(parts) < 3 or not parts[2].startswith("inline:"): raise ValueError(f"Malformed SDES crypto attribute: {value!r}") - _tag, suite, inline = parts[0], parts[1], parts[2] + _tag, suite, key_params = parts[0], parts[1], parts[2] + # RFC 4568 §5.1.1: several inline key-params may be joined by ';'; + # the first is the active key, the rest are pre-rolled backups. + inline = key_params.split(";", 1)[0] # The inline parameter may carry trailing |lifetime or ~MKI suffixes. key_material_b64, _, _trailing = inline.removeprefix("inline:").partition("|") key_material_b64 = key_material_b64.split("~", 1)[0] From 46046b6410098ed0aaaf7b43048797fc14b308cc Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Fri, 19 Jun 2026 21:32:01 +0200 Subject: [PATCH 9/9] Refactor --- tests/codecs/test_opus.py | 48 ++++++++--------- tests/sip/test_types.py | 16 +++--- tests/test_mcp.py | 14 ++--- tests/test_stun.py | 30 +++++------ voip/__main__.py | 10 ++-- voip/ai.py | 38 +++++++------- voip/audio.py | 32 ++++++------ voip/codecs/opus.py | 16 +++--- voip/codecs/pcma.py | 4 +- voip/codecs/pcmu.py | 8 +-- voip/mcp.py | 14 ++--- voip/sdp/messages.py | 16 +++--- voip/sdp/types.py | 4 +- voip/sip/messages.py | 18 +++---- voip/sip/protocol.py | 28 +++++----- voip/sip/transactions.py | 36 +++++++------ voip/sip/types.py | 14 ++--- voip/srtp.py | 105 +++++++++++++++++--------------------- voip/stun.py | 29 +++++------ 19 files changed, 235 insertions(+), 245 deletions(-) diff --git a/tests/codecs/test_opus.py b/tests/codecs/test_opus.py index 67958e1..9d5786d 100644 --- a/tests/codecs/test_opus.py +++ b/tests/codecs/test_opus.py @@ -12,64 +12,64 @@ class TestOggCRC32: def test_ogg_crc32__empty_bytes(self): - """_ogg_crc32 of empty bytes is zero.""" - assert Opus._ogg_crc32(b"") == 0 + """ogg_crc32 of empty bytes is zero.""" + assert Opus.ogg_crc32(b"") == 0 def test_ogg_crc32__known_value(self): - """_ogg_crc32 produces a deterministic 32-bit value.""" - crc = Opus._ogg_crc32(b"OggS") + """ogg_crc32 produces a deterministic 32-bit value.""" + crc = Opus.ogg_crc32(b"OggS") assert 0 <= crc <= 0xFFFFFFFF class TestOggPage: def test_ogg_page__starts_with_capture_pattern(self): - """_ogg_page output starts with the Ogg capture pattern 'OggS'.""" - page = Opus._ogg_page(0x02, 0, 0x12345678, 0, [b"hello"]) + """ogg_page output starts with the Ogg capture pattern 'OggS'.""" + page = Opus.ogg_page(0x02, 0, 0x12345678, 0, [b"hello"]) assert page[:4] == b"OggS" def test_ogg_page__contains_packet_data(self): - """_ogg_page embeds the provided packet bytes.""" - page = Opus._ogg_page(0x02, 0, 0, 0, [b"payload"]) + """ogg_page embeds the provided packet bytes.""" + page = Opus.ogg_page(0x02, 0, 0, 0, [b"payload"]) assert b"payload" in page def test_ogg_page__large_packet_uses_255_lacing(self): - """_ogg_page correctly laces a packet exceeding 254 bytes.""" - page = Opus._ogg_page(0x00, 0, 0, 0, [b"x" * 256]) + """ogg_page correctly laces a packet exceeding 254 bytes.""" + page = Opus.ogg_page(0x00, 0, 0, 0, [b"x" * 256]) assert page[:4] == b"OggS" assert len(page) > 256 class TestOggContainer: def test_ogg_container__starts_with_ogg_magic(self): - """_ogg_container output starts with the Ogg capture pattern 'OggS'.""" - assert Opus._ogg_container(b"packet").startswith(b"OggS") + """ogg_container output starts with the Ogg capture pattern 'OggS'.""" + assert Opus.ogg_container(b"packet").startswith(b"OggS") def test_ogg_container__contains_opus_head(self): - """_ogg_container includes the OpusHead identification header.""" - assert b"OpusHead" in Opus._ogg_container(b"packet") + """ogg_container includes the OpusHead identification header.""" + assert b"OpusHead" in Opus.ogg_container(b"packet") def test_ogg_container__contains_opus_tags(self): - """_ogg_container includes the OpusTags comment header.""" - assert b"OpusTags" in Opus._ogg_container(b"packet") + """ogg_container includes the OpusTags comment header.""" + assert b"OpusTags" in Opus.ogg_container(b"packet") def test_ogg_container__non_empty_for_single_packet(self): - """_ogg_container produces a non-empty Ogg container for a single Opus packet.""" - assert len(Opus._ogg_container(b"x" * 100)) > 100 + """ogg_container produces a non-empty Ogg container for a single Opus packet.""" + assert len(Opus.ogg_container(b"x" * 100)) > 100 def test_ogg_container__empty_payload(self): - """_ogg_container produces a valid Ogg container even for empty payload.""" - result = Opus._ogg_container(b"") + """ogg_container produces a valid Ogg container even for empty payload.""" + result = Opus.ogg_container(b"") assert b"OggS" in result def test_ogg_container__produces_three_pages(self): - """_ogg_container produces exactly three Ogg pages: BOS, tags, and data.""" - result = Opus._ogg_container(b"x" * 10) + """ogg_container produces exactly three Ogg pages: BOS, tags, and data.""" + result = Opus.ogg_container(b"x" * 10) assert result.count(b"OggS") == 3 class TestOpusDecode: def test_decode__wraps_in_ogg_format(self): - """Decode passes the payload through _ogg_container before calling decode_pcm.""" + """Decode passes the payload through ogg_container before calling decode_pcm.""" with patch.object( Opus, "decode_pcm", return_value=np.zeros(16000, dtype=np.float32) ) as mock_decode_pcm: @@ -146,7 +146,7 @@ def test_packetize__frame_count(self): Regression test: the previous implementation appended a flush packet (`codec.encode(None)`) after all frames, producing N+1 RTP packets - for N frames of audio. `_dispatch_next_packet` sends every yielded + for N frames of audio. `dispatch_next_packet` sends every yielded payload at a fixed 20 ms interval, so the extra packet shifted the receiver's playback timeline by one ptime (20 ms), causing audible timing glitches. diff --git a/tests/sip/test_types.py b/tests/sip/test_types.py index 4603182..6482b01 100644 --- a/tests/sip/test_types.py +++ b/tests/sip/test_types.py @@ -405,29 +405,29 @@ def test_host__tel_absent(self): class TestMaskCaller: def test_mask_caller__with_display_name(self): """Mask all but last four chars of a quoted display name.""" - from voip.sip.types import _mask_caller + from voip.sip.types import mask_caller assert ( - _mask_caller('"08001234567" ;tag=abc') + mask_caller('"08001234567" ;tag=abc') == "*******4567" ) def test_mask_caller__bare_uri(self): """Mask user part from a bare SIP URI.""" - from voip.sip.types import _mask_caller + from voip.sip.types import mask_caller - assert _mask_caller("sip:alice@example.com") == "*lice" + assert mask_caller("sip:alice@example.com") == "*lice" def test_mask_caller__short_name(self): """Return the name unchanged when it is four characters or fewer.""" - from voip.sip.types import _mask_caller + from voip.sip.types import mask_caller - assert _mask_caller("sip:bob@example.com") == "bob" + assert mask_caller("sip:bob@example.com") == "bob" def test_mask_caller__long_name(self): """Mask all but last four characters of a long username.""" - from voip.sip.types import _mask_caller + from voip.sip.types import mask_caller - result = _mask_caller("sip:verylonguser@example.com") + result = mask_caller("sip:verylonguser@example.com") assert result.endswith("user") assert result.startswith("*") diff --git a/tests/test_mcp.py b/tests/test_mcp.py index d8de9e1..f0221b5 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -71,8 +71,8 @@ def make_agent_call( agent = MCPAgentCall(**kwargs) if messages is not None: - # Inject synthetic _messages (bypasses AgentCall's auto-population). - object.__setattr__(agent, "_messages", messages) + # Inject synthetic messages (bypasses AgentCall's auto-population). + object.__setattr__(agent, "messages", messages) return agent @@ -121,7 +121,7 @@ async def test_transcription_received__appends_user_message(self) -> None: with patch("asyncio.create_task"): agent.transcription_received("How are you?") - user_msgs = [m for m in agent._messages if m["role"] == "user"] + user_msgs = [m for m in agent.messages if m["role"] == "user"] assert any(m["content"] == "How are you?" for m in user_msgs) async def test_transcription_received__cancels_pending_task(self) -> None: @@ -129,7 +129,7 @@ async def test_transcription_received__cancels_pending_task(self) -> None: agent = make_agent_call() old_task = MagicMock(spec=asyncio.Task) old_task.done.return_value = False - object.__setattr__(agent, "_response_task", old_task) + object.__setattr__(agent, "response_task", old_task) with patch.object(agent, "cancel_outbound_audio"): with patch("asyncio.create_task") as mock_create: @@ -143,7 +143,7 @@ async def test_transcription_received__skips_cancel_when_done(self) -> None: agent = make_agent_call() old_task = MagicMock(spec=asyncio.Task) old_task.done.return_value = True - object.__setattr__(agent, "_response_task", old_task) + object.__setattr__(agent, "response_task", old_task) with patch.object(agent, "cancel_outbound_audio"): with patch("asyncio.create_task"): @@ -159,7 +159,7 @@ async def test_transcription_received__skips_cancel_when_done(self) -> None: class TestRespond: async def test_respond__speaks_reply(self) -> None: - """respond() speaks the LLM reply and appends it to _messages.""" + """respond() speaks the LLM reply and appends it to messages.""" ctx = make_mock_context("Nice to meet you.") agent = make_agent_call( ctx=ctx, @@ -172,7 +172,7 @@ async def test_respond__speaks_reply(self) -> None: await agent.respond() mock_send.assert_awaited_once_with("Nice to meet you.") - assert {"role": "assistant", "content": "Nice to meet you."} in agent._messages + assert {"role": "assistant", "content": "Nice to meet you."} in agent.messages async def test_respond__filters_system_from_sampling(self) -> None: """System messages are not forwarded to ctx.sample.""" diff --git a/tests/test_stun.py b/tests/test_stun.py index 0737200..037060a 100644 --- a/tests/test_stun.py +++ b/tests/test_stun.py @@ -11,7 +11,7 @@ STUNAttributeType, STUNMessageType, STUNProtocol, - _parse_address, + parse_address, ) @@ -91,12 +91,12 @@ def test_xor_mapped_address__value(self): 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 + 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 + assert parse_address(value, b"") is None class TestSTUNProtocol: @@ -237,7 +237,7 @@ def stun_connection_made(self, transport, 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.stun_transaction_id = transaction_id proto.datagram_received(response, ("::1", 3478)) assert len(received) == 1 assert received[0] == (ipaddress.IPv6Address("2001:db8::1"), 54321) @@ -257,7 +257,7 @@ def stun_connection_made(self, transport, 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.stun_transaction_id = transaction_id proto.datagram_received(response, ("::1", 3478)) assert len(received) == 1 assert received[0] == (ipaddress.IPv6Address("::1"), 12345) @@ -277,7 +277,7 @@ def stun_connection_made(self, transport, 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.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) @@ -296,7 +296,7 @@ def stun_connection_made(self, transport, 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.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) @@ -332,22 +332,22 @@ def test_error_received__logs_warning(self, caplog): 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.""" + """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 + 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.""" + """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 + 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.""" + """parse_stun_response() ignores responses with a mismatched transaction ID.""" transaction_id = b"\x06" * 12 received: list = [] @@ -359,10 +359,10 @@ def stun_connection_made(self, transport, addr): 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 + 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) + proto.parse_stun_response(response) assert received == [] diff --git a/voip/__main__.py b/voip/__main__.py index c98c195..d2d916d 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -27,7 +27,7 @@ logger = logging.getLogger("voip") -def _parse_sip_uri(ctx, param, value) -> SipURI: +def parse_sip_uri(ctx, param, value) -> SipURI: """Parse a SIP URI.""" try: return SipURI.parse(value) @@ -112,7 +112,7 @@ def voip(ctx, verbose: int = 0): "aor", metavar="AOR", envvar="SIP_AOR", - callback=_parse_sip_uri, + callback=parse_sip_uri, ) @click.option( "--stun-server", @@ -154,7 +154,7 @@ def mcp(aor: SipURI, stun_server: NetworkAddress, no_verify_tls: bool, transport "aor", metavar="AOR", envvar="SIP_AOR", - callback=_parse_sip_uri, + callback=parse_sip_uri, ) @click.option( "--stun-server", @@ -372,7 +372,7 @@ def transcription_received(self, text: str) -> None: super().transcription_received(text) async def send_audio(self, audio: np.ndarray) -> None: - for msg in self._messages[self.msg_count :]: + for msg in self.messages[self.msg_count :]: click.echo( click.style( f"Agent: {msg['content']}", @@ -383,7 +383,7 @@ async def send_audio(self, audio: np.ndarray) -> None: await super().send_audio(audio) async def respond(self) -> None: - self.msg_count = len(self._messages) + self.msg_count = len(self.messages) await super().respond() class AgentDialog(dialog.Dialog): diff --git a/voip/ai.py b/voip/ai.py index dd900b7..737d4d4 100644 --- a/voip/ai.py +++ b/voip/ai.py @@ -114,13 +114,13 @@ class TTSMixin: ) voice: pathlib.Path | str | torch.Tensor = dataclasses.field(default="marius") - _voice_state: dict[str, dict[str, torch.Tensor]] = dataclasses.field( + voice_state: dict[str, dict[str, torch.Tensor]] = dataclasses.field( init=False, repr=False ) def __post_init__(self) -> None: super().__post_init__() - self._voice_state = self.tts_model.get_state_for_audio_prompt(self.voice) + self.voice_state = self.tts_model.get_state_for_audio_prompt(self.voice) async def send_speech(self, text: str) -> None: """Synthesise `text` and transmit it as outbound RTP audio. @@ -130,7 +130,7 @@ async def send_speech(self, text: str) -> None: """ await self.send_audio( self.resample( - self.tts_model.generate_audio(self._voice_state, text).numpy(), + self.tts_model.generate_audio(self.voice_state, text).numpy(), self.tts_model.sample_rate, self.codec.sample_rate_hz, ) @@ -182,11 +182,11 @@ class AgentCall(TTSMixin, TranscribeCall): salutation: str = dataclasses.field(default="Hi.") audio_interrupt_duration: datetime.timedelta = datetime.timedelta(seconds=0.75) - _messages: list[dict] = dataclasses.field(init=False, repr=False) - _response_task: asyncio.Task | None = dataclasses.field( + messages: list[dict] = dataclasses.field(init=False, repr=False) + response_task: asyncio.Task | None = dataclasses.field( init=False, repr=False, default=None ) - _cancel_audio_handle: asyncio.Handle | None = dataclasses.field( + cancel_audio_handle: asyncio.Handle | None = dataclasses.field( init=False, repr=False, default=None ) @@ -204,37 +204,37 @@ class AgentCall(TTSMixin, TranscribeCall): def __post_init__(self) -> None: super().__post_init__() - self._messages = [ + self.messages = [ { "role": "system", "content": self.system_prompt, } ] if self.salutation: - self._messages.append({"role": "assistant", "content": self.salutation}) + self.messages.append({"role": "assistant", "content": self.salutation}) asyncio.create_task(self.send_speech(self.salutation)) def transcription_received(self, text: str) -> None: self.cancel_outbound_audio() - self._messages.append({"role": "user", "content": text}) - if self._response_task is not None and not self._response_task.done(): - self._response_task.cancel() - self._response_task = asyncio.create_task(self.respond()) + self.messages.append({"role": "user", "content": text}) + if self.response_task is not None and not self.response_task.done(): + self.response_task.cancel() + self.response_task = asyncio.create_task(self.respond()) async def respond(self) -> None: response = await ollama.AsyncClient().chat( model=self.llm_model, - messages=self._messages, + messages=self.messages, ) if reply := self.emoji_pattern.sub("", response.message.content or ""): - self._messages.append({"role": "assistant", "content": reply}) + self.messages.append({"role": "assistant", "content": reply}) logger.debug("Agent reply: %r", reply) await self.send_speech(reply) def on_audio_speech(self) -> None: loop = asyncio.get_event_loop() - if self._cancel_audio_handle is None: - self._cancel_audio_handle = loop.call_later( + if self.cancel_audio_handle is None: + self.cancel_audio_handle = loop.call_later( self.audio_interrupt_duration.total_seconds(), self.cancel_outbound_audio, ) @@ -242,6 +242,6 @@ def on_audio_speech(self) -> None: def on_audio_silence(self) -> None: super().on_audio_silence() - if self._cancel_audio_handle is not None: - self._cancel_audio_handle.cancel() - self._cancel_audio_handle = None + if self.cancel_audio_handle is not None: + self.cancel_audio_handle.cancel() + self.cancel_audio_handle = None diff --git a/voip/audio.py b/voip/audio.py index 630e11f..cbe1dc3 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -228,7 +228,7 @@ def on_audio_sent(self) -> None: [SayCall][voip.ai.SayCall] finishes speaking. """ - def _dispatch_next_packet( + def dispatch_next_packet( self, packets: Iterator[bytes], remote_addr: tuple[str, int], @@ -246,7 +246,7 @@ def _dispatch_next_packet( loop = asyncio.get_running_loop() self.outbound_handle = loop.call_at( next_deadline, - self._dispatch_next_packet, + self.dispatch_next_packet, packets, remote_addr, next_deadline, @@ -273,7 +273,7 @@ async def send_audio(self, audio: np.ndarray) -> None: self.cancel_outbound_audio() loop = asyncio.get_running_loop() next_send_at = loop.time() - self._dispatch_next_packet( + self.dispatch_next_packet( self.codec.packetize(audio), remote_addr, next_send_at, @@ -343,43 +343,43 @@ async def voice_received(self, audio: np.ndarray) -> None: default=datetime.timedelta(milliseconds=200) ) - _speech_buffer: np.ndarray = dataclasses.field( + speech_buffer: np.ndarray = dataclasses.field( init=False, repr=False, default_factory=lambda: np.empty((0,), dtype=np.float32) ) - _flush_voice_buffer_handle: asyncio.TimerHandle | None = dataclasses.field( + flush_voice_buffer_handle: asyncio.TimerHandle | None = dataclasses.field( init=False, repr=False, default=None ) def audio_received(self, *, audio: np.ndarray, rms: float) -> None: - self._speech_buffer = np.concatenate((self._speech_buffer, audio)) + self.speech_buffer = np.concatenate((self.speech_buffer, audio)) if rms > self.voice_rms_threshold: self.on_audio_speech() else: self.on_audio_silence() def on_audio_speech(self) -> None: - if self._flush_voice_buffer_handle is not None: - self._flush_voice_buffer_handle.cancel() - self._flush_voice_buffer_handle = None + if self.flush_voice_buffer_handle is not None: + self.flush_voice_buffer_handle.cancel() + self.flush_voice_buffer_handle = None def on_audio_silence(self) -> None: - if self._flush_voice_buffer_handle is None: + if self.flush_voice_buffer_handle is None: loop = asyncio.get_event_loop() - self._flush_voice_buffer_handle = loop.call_later( + self.flush_voice_buffer_handle = loop.call_later( self.silence_gap.total_seconds(), self.flush_voice_buffer, ) def flush_voice_buffer(self) -> None: - self._flush_voice_buffer_handle = None + self.flush_voice_buffer_handle = None # Ensure at least one second of audio to avoid cutting words in half. if not ( - len(self._speech_buffer) + len(self.speech_buffer) < self.sampling_rate_hz * self.silence_gap.total_seconds() - or self.rms(self._speech_buffer) < self.utterances_rms_threshold + or self.rms(self.speech_buffer) < self.utterances_rms_threshold ): - asyncio.create_task(self.voice_received(self._speech_buffer.copy())) - self._speech_buffer = np.empty((0,), dtype=np.float32) + asyncio.create_task(self.voice_received(self.speech_buffer.copy())) + self.speech_buffer = np.empty((0,), dtype=np.float32) async def voice_received(self, audio: np.ndarray) -> None: """Handle the flushed speech buffer. Override in subclasses. diff --git a/voip/codecs/opus.py b/voip/codecs/opus.py index 7be3433..5aa708f 100644 --- a/voip/codecs/opus.py +++ b/voip/codecs/opus.py @@ -50,7 +50,7 @@ class Opus(PyAVCodec): channels: ClassVar[int] = 2 @staticmethod - def _ogg_crc32(data: bytes) -> int: + def ogg_crc32(data: bytes) -> int: """Compute an Ogg CRC32 checksum (polynomial 0x04C11DB7). Args: @@ -67,7 +67,7 @@ def _ogg_crc32(data: bytes) -> int: return crc & 0xFFFFFFFF @classmethod - def _ogg_page( + def ogg_page( cls, header_type: int, granule_position: int, @@ -106,10 +106,10 @@ def _ogg_page( len(lacing), ) + bytes(lacing) page = header + b"".join(packets) - return page[:22] + struct.pack(" bytes: + def ogg_container(cls, packet: bytes) -> bytes: """Wrap a raw Opus RTP payload in a minimal Ogg Opus container. Produces a three-page Ogg stream: BOS (OpusHead), comment @@ -141,9 +141,9 @@ def _ogg_container(cls, packet: bytes) -> bytes: ) return b"".join( [ - cls._ogg_page(0x02, 0, serial_number, 0, [opus_head]), # BOS - cls._ogg_page(0x00, 0, serial_number, 1, [opus_tags]), - cls._ogg_page(0x04, cls.frame_size, serial_number, 2, [packet]), + cls.ogg_page(0x02, 0, serial_number, 0, [opus_head]), # BOS + cls.ogg_page(0x00, 0, serial_number, 1, [opus_tags]), + cls.ogg_page(0x04, cls.frame_size, serial_number, 2, [packet]), ] ) @@ -155,7 +155,7 @@ def decode( *, input_rate_hz: int | None = None, ) -> np.ndarray: - return cls.decode_pcm(cls._ogg_container(payload), "ogg", output_rate_hz) + return cls.decode_pcm(cls.ogg_container(payload), "ogg", output_rate_hz) @classmethod def encode(cls, samples: np.ndarray) -> bytes: diff --git a/voip/codecs/pcma.py b/voip/codecs/pcma.py index d27ff0c..9157206 100644 --- a/voip/codecs/pcma.py +++ b/voip/codecs/pcma.py @@ -16,7 +16,7 @@ # G.711 A-law segment upper bounds (16-bit PCM magnitude, inclusive per segment). # Vectorised segment lookup via np.searchsorted uses side='left' to count thresholds # strictly exceeded (v > threshold), giving the correct 0–7 segment index. -_ALAW_SEG_UBOUND: np.ndarray = np.array( +ALAW_SEG_UBOUND: np.ndarray = np.array( (0xFF, 0x1FF, 0x3FF, 0x7FF, 0xFFF, 0x1FFF, 0x3FFF, 0x7FFF), dtype=np.int32 ) @@ -78,7 +78,7 @@ def encode(cls, samples: np.ndarray) -> bytes: # Find the segment index (0–7) via vectorised binary search on the upper bounds. # side='left' counts thresholds strictly less than magnitude, i.e. exceeded. seg = np.minimum( - np.searchsorted(_ALAW_SEG_UBOUND, magnitude, side="left").astype(np.int32), + np.searchsorted(ALAW_SEG_UBOUND, magnitude, side="left").astype(np.int32), 7, ) # Extract the 4-bit mantissa in 16-bit space. diff --git a/voip/codecs/pcmu.py b/voip/codecs/pcmu.py index 4010db1..bad46a1 100644 --- a/voip/codecs/pcmu.py +++ b/voip/codecs/pcmu.py @@ -13,8 +13,8 @@ __all__ = ["PCMU"] -_MU_LAW_BIAS: int = 0x84 # 132: G.711 mu-law bias constant -_MU_LAW_CLIP: int = 32635 # maximum biased magnitude (14-bit saturate) +MU_LAW_BIAS: int = 0x84 # 132: G.711 mu-law bias constant +MU_LAW_CLIP: int = 32635 # maximum biased magnitude (14-bit saturate) class PCMU(RTPCodec): @@ -49,7 +49,7 @@ def decode( exp = ((raw >> 4) & 0x07).astype(np.int32) mantissa = (raw & 0x0F).astype(np.int32) # ITU-T G.711 §A: magnitude = (((mantissa << 3) + BIAS) << exp) - BIAS - magnitude = (((mantissa << 3) + _MU_LAW_BIAS) << exp) - _MU_LAW_BIAS + magnitude = (((mantissa << 3) + MU_LAW_BIAS) << exp) - MU_LAW_BIAS linear = (magnitude.astype(np.float32) / 32768.0).astype(np.float32) source_rate_hz = ( input_rate_hz if input_rate_hz is not None else cls.sample_rate_hz @@ -62,7 +62,7 @@ def decode( def encode(cls, samples: np.ndarray) -> bytes: pcm = np.clip(np.round(samples * 32768.0), -32768, 32767).astype(np.int32) sign = np.where(pcm >= 0, 0x80, 0x00).astype(np.uint8) - biased = np.minimum(np.abs(pcm) + _MU_LAW_BIAS, _MU_LAW_CLIP) + biased = np.minimum(np.abs(pcm) + MU_LAW_BIAS, MU_LAW_CLIP) exp = np.clip( np.floor(np.log2(np.maximum(biased, 1))).astype(np.int32) - 7, 0, 7 ) diff --git a/voip/mcp.py b/voip/mcp.py index cd6e677..51c3e3b 100644 --- a/voip/mcp.py +++ b/voip/mcp.py @@ -52,16 +52,16 @@ class MCPAgentCall(ai.AgentCall): def transcript(self) -> str: return "\n".join( f"{'Caller' if msg['role'] == 'user' else 'Agent'}: {msg['content']}" - for msg in self._messages + for msg in self.messages if msg["role"] != "system" ) def transcription_received(self, text: str) -> None: self.cancel_outbound_audio() - self._messages.append({"role": "user", "content": text}) - if self._response_task is not None and not self._response_task.done(): - self._response_task.cancel() - self._response_task = asyncio.create_task(self.respond()) + self.messages.append({"role": "user", "content": text}) + if self.response_task is not None and not self.response_task.done(): + self.response_task.cancel() + self.response_task = asyncio.create_task(self.respond()) async def respond(self) -> None: sampling_messages = [ @@ -69,7 +69,7 @@ async def respond(self) -> None: role=typing.cast(typing.Literal["user", "assistant"], msg["role"]), content=TextContent(type="text", text=msg["content"]), ) - for msg in self._messages + for msg in self.messages if msg["role"] != "system" ] result = await self.ctx.sample( @@ -77,7 +77,7 @@ async def respond(self) -> None: system_prompt=self.system_prompt, ) if result.text and (reply := result.text.strip()): - self._messages.append({"role": "assistant", "content": reply}) + self.messages.append({"role": "assistant", "content": reply}) await self.send_speech(reply) diff --git a/voip/sdp/messages.py b/voip/sdp/messages.py index 20acf84..7d18b7a 100644 --- a/voip/sdp/messages.py +++ b/voip/sdp/messages.py @@ -72,10 +72,10 @@ def parse(cls, data: bytes | str) -> SessionDescription: sdp = cls() current_media: MediaDescription | None = None for line in text.splitlines(): - current_media = sdp._apply_line(line.rstrip("\r"), current_media) + current_media = sdp.apply_line(line.rstrip("\r"), current_media) return sdp - def _apply_line( + def apply_line( self, line: str, current_media: MediaDescription | None ) -> MediaDescription | None: """Apply a single SDP line to this session, return the active MediaDescription.""" @@ -94,10 +94,10 @@ def _apply_line( and isinstance(parsed, Attribute) and current_media is not None ): - if self._apply_media_attribute(parsed, current_media): + if self.apply_media_attribute(parsed, current_media): return current_media if field.media_attr is not None and current_media is not None: - return self._apply_to_media( + return self.apply_to_media( current_media, field.media_attr, parsed, field.is_list ) if field.is_list: @@ -107,7 +107,7 @@ def _apply_line( return current_media @staticmethod - def _apply_media_attribute(attr: Attribute, media: MediaDescription) -> bool: + def apply_media_attribute(attr: Attribute, media: MediaDescription) -> bool: """Fold a media-level a= attribute into *media* if it is a format-specific attribute. Returns `True` when the attribute was consumed (`a=rtpmap` or @@ -117,7 +117,7 @@ def _apply_media_attribute(attr: Attribute, media: MediaDescription) -> bool: return media.apply_attribute(attr) @staticmethod - def _apply_to_media( + def apply_to_media( media: MediaDescription, attr: str, value: object, is_list: bool ) -> MediaDescription: """Apply a parsed field value to a MediaDescription, return it unchanged.""" @@ -131,9 +131,9 @@ def __bytes__(self) -> bytes: return str(self).encode() def __str__(self) -> str: - return "\r\n".join(self._lines()) + "\r\n" + return "\r\n".join(self.lines()) + "\r\n" - def _lines(self) -> Generator[str]: + def lines(self) -> Generator[str]: """Yield each SDP line in canonical field order.""" for field in FIELD_MAP: if field.session_attr == "media": diff --git a/voip/sdp/types.py b/voip/sdp/types.py index 51c2a0f..307b424 100644 --- a/voip/sdp/types.py +++ b/voip/sdp/types.py @@ -383,7 +383,7 @@ def apply_attribute(self, attr: Attribute) -> bool: return True return False - def _lines(self) -> Generator[str]: + def lines(self) -> Generator[str]: """Yield each SDP line in canonical field order.""" yield f"m={self.media} {self.port} {self.proto} {' '.join(str(f.payload_type) for f in self.fmt)}" match self.title: @@ -403,7 +403,7 @@ def _lines(self) -> Generator[str]: yield from (f"a={a}" for a in self.attributes) def __bytes__(self) -> bytes: - return "\r\n".join(self._lines()).encode() + return "\r\n".join(self.lines()).encode() @classmethod def parse(cls, data: bytes | str) -> MediaDescription: diff --git a/voip/sip/messages.py b/voip/sip/messages.py index f2fc9f6..fe89cca 100644 --- a/voip/sip/messages.py +++ b/voip/sip/messages.py @@ -81,7 +81,7 @@ def parse(cls, data: bytes) -> Request | Response: status_code=int(status_code_str), phrase=reason.decode("ascii"), headers=headers, - body=cls._parse_body(headers, body), + body=cls.parse_body(headers, body), version=version.decode("ascii"), ) try: @@ -92,12 +92,12 @@ def parse(cls, data: bytes) -> Request | Response: method=method.decode("ascii"), uri=uri.decode("ascii"), headers=headers, - body=cls._parse_body(headers, body), + body=cls.parse_body(headers, body), version=version.decode("ascii"), ) @staticmethod - def _parse_body(headers: dict[str, str], body: bytes) -> SessionDescription | None: + def parse_body(headers: dict[str, str], body: bytes) -> SessionDescription | None: """Parse the body according to the Content-Type header.""" if headers.get("Content-Type") == "application/sdp" and body: return SessionDescription.parse(body) @@ -106,12 +106,10 @@ def _parse_body(headers: dict[str, str], body: bytes) -> SessionDescription | No def __bytes__(self) -> bytes: if raw_body := bytes(self.body) if self.body is not None else b"": self.headers["Content-Length"] = str(len(raw_body)) - return b"\r\n".join( - (self._first_line().encode(), bytes(self.headers), raw_body) - ) + return b"\r\n".join((self.first_line().encode(), bytes(self.headers), raw_body)) def __repr__(self): - return self._first_line() + return self.first_line() @property def branch(self) -> str: @@ -135,7 +133,7 @@ def sequence(self) -> int: return int(self.headers["CSeq"].split()[0]) @abc.abstractmethod - def _first_line(self) -> str: ... + def first_line(self) -> str: ... @dataclasses.dataclass(slots=True, kw_only=True, repr=False) @@ -149,7 +147,7 @@ class Request(Message): method: SIPMethod uri: SipURI - def _first_line(self) -> str: + def first_line(self) -> str: return f"{self.method} {self.uri} {self.version}" @classmethod @@ -172,7 +170,7 @@ class Response(Message): status_code: SIPStatus | int phrase: str - def _first_line(self) -> str: + def first_line(self) -> str: return f"{self.version} {self.status_code} {self.phrase}" @classmethod diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index 6ec48d6..38b4443 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -101,10 +101,10 @@ async def main(): keepalive_task: asyncio.Task | None = dataclasses.field(init=False, default=None) public_address: NetworkAddress = None - _dialogs: dict[tuple[str, str], Dialog] = dataclasses.field( + dialogs: dict[tuple[str, str], Dialog] = dataclasses.field( init=False, default_factory=dict ) - _transactions: dict[str, Transaction] = dataclasses.field( + transactions: dict[str, Transaction] = dataclasses.field( init=False, default_factory=dict ) disconnected_event: asyncio.Event = dataclasses.field( @@ -258,7 +258,7 @@ def register_dialog(self, dialog: Dialog) -> None: if dialog.remote_tag is None: logger.warning("Dialog without remote tag cannot be registered: %r", dialog) else: - self._dialogs[dialog.local_tag, dialog.remote_tag] = dialog + self.dialogs[dialog.local_tag, dialog.remote_tag] = dialog def drop_dialog(self, dialog: Dialog) -> None: """Remove *dialog* from the registry.""" @@ -266,18 +266,18 @@ def drop_dialog(self, dialog: Dialog) -> None: logger.warning("Dialog without remote tag cannot be removed: %r", dialog) else: try: - del self._dialogs[dialog.local_tag, dialog.remote_tag] + del self.dialogs[dialog.local_tag, dialog.remote_tag] except KeyError: logger.warning("Dialog not found for removal: %r", dialog) def register_transaction(self, tx: Transaction) -> None: """Register *tx* by its branch parameter.""" - self._transactions[tx.branch] = tx + self.transactions[tx.branch] = tx def drop_transaction(self, tx: Transaction) -> None: """Remove *tx* from the registry.""" try: - del self._transactions[tx.branch] + del self.transactions[tx.branch] except KeyError: logger.warning("Transaction not found for removal: %r", tx) @@ -317,18 +317,18 @@ async def handle_registration(self, tx: RegisterTransaction) -> None: def data_received(self, data: bytes) -> None: self.recv_buffer.extend(data) - for frame in self._extract_frames(): - self._dispatch_frame(frame) + for frame in self.extract_frames(): + self.dispatch_frame(frame) def datagram_received(self, data: bytes, addr: tuple) -> None: # type: ignore[override] """Dispatch a complete UDP SIP datagram.""" - self._dispatch_frame(data) + self.dispatch_frame(data) def error_received(self, exc: Exception) -> None: # type: ignore[override] """Log a UDP transport error.""" logger.warning("UDP error received", exc_info=exc) - def _extract_frames(self) -> typing.Generator[memoryview | bytes]: # noqa: C901 + def extract_frames(self) -> typing.Generator[memoryview | bytes]: # noqa: C901 while self.recv_buffer: if self.recv_buffer[0:1] != b"\r": # SIP message: wait for the header-body separator. @@ -364,7 +364,7 @@ def _extract_frames(self) -> typing.Generator[memoryview | bytes]: # noqa: C901 # Single CR or other incomplete sequence – wait for more data. break - def _dispatch_frame(self, frame: memoryview | bytes) -> None: + def dispatch_frame(self, frame: memoryview | bytes) -> None: peer = NetworkAddress(*self.transport.get_extra_info("peername")[:2]) if frame == PONG: logger.info("PONG", extra={"addr": peer}) @@ -458,7 +458,7 @@ def request_received(self, request: Request) -> None: case SIPMethod.ACK: # For non-2xx ACKs the INVITE tx is still present; route by dialog. try: - tx = self._dialogs[ + tx = self.dialogs[ request.remote_tag, request.local_tag ].invite_transaction except KeyError: @@ -475,7 +475,7 @@ def request_received(self, request: Request) -> None: asyncio.create_task(ByeTransaction.receive(request=request, sip=self)) case SIPMethod.CANCEL: try: - tx = self._transactions[request.branch] + tx = self.transactions[request.branch] except KeyError: self.send( Response.from_request( @@ -505,7 +505,7 @@ def response_received(self, response: Response) -> None: response: The parsed SIP response. """ try: - tx = self._transactions[response.branch] + tx = self.transactions[response.branch] except KeyError: logger.warning( "Received response with unknown branch %r: %r", diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py index c9a9001..3c60d56 100644 --- a/voip/sip/transactions.py +++ b/voip/sip/transactions.py @@ -128,7 +128,7 @@ def from_request( sip: SessionInitiationProtocol, ): try: - dialog = sip._dialogs[request.remote_tag, request.local_tag] + dialog = sip.dialogs[request.remote_tag, request.local_tag] except KeyError: dialog = sip.dialog_class.from_request(request) return cls( @@ -326,16 +326,18 @@ def digest_response( if is_sess and cnonce is None: raise ValueError(f"algorithm={algorithm!r} requires a cnonce value") - def h(data: str) -> str: - return hashlib.new(hash_name, data.encode()).hexdigest() - - ha1 = h(f"{username}:{realm}:{password}") + ha1 = cls.hash(hash_name, f"{username}:{realm}:{password}") if is_sess: - ha1 = h(f"{ha1}:{nonce}:{cnonce}") - ha2 = h(f"{method}:{uri}") + ha1 = cls.hash(hash_name, f"{ha1}:{nonce}:{cnonce}") + ha2 = cls.hash(hash_name, f"{method}:{uri}") if qop in (DigestQoP.AUTH, DigestQoP.AUTH_INT): - return h(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}") - return h(f"{ha1}:{nonce}:{ha2}") + return cls.hash(hash_name, f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}") + return cls.hash(hash_name, f"{ha1}:{nonce}:{ha2}") + + @staticmethod + def hash(hash_name: str, data: str) -> str: + """Return the hex digest of *data* using the named hash algorithm.""" + return hashlib.new(hash_name, data.encode()).hexdigest() @dataclasses.dataclass(kw_only=True, slots=True) @@ -746,7 +748,7 @@ async def send( ) tx.pending_call_class = session_class tx.pending_call_kwargs = session_kwargs - tx.request = tx._build_invite_request(target) + tx.request = tx.build_invite_request(target) sip.register_transaction(tx) sip.send(tx.request) try: @@ -755,7 +757,7 @@ async def send( sip.drop_transaction(tx) raise - def _build_invite_request(self, target: types.SipURI) -> messages.Request: + def build_invite_request(self, target: types.SipURI) -> messages.Request: """Build the outbound INVITE request (with SDP offer) for *target*. Factored out of [send][voip.sip.transactions.InviteTransaction.send] @@ -832,7 +834,7 @@ def retry_with_auth( SRTP/RTP offer mode is preserved across the retry. """ header = "Proxy-Authorization" if is_proxy else "Authorization" - return self._retry_invite( + return self.retry_invite( offer_srtp=self.offer_srtp, auth_headers={header: auth_value} ) @@ -851,9 +853,9 @@ def retry_with_rtp(self, response: Response) -> bool: for header in ("Authorization", "Proxy-Authorization") if header in self.request.headers } - return self._retry_invite(offer_srtp=False, auth_headers=auth_headers) + return self.retry_invite(offer_srtp=False, auth_headers=auth_headers) - def _retry_invite(self, *, offer_srtp: bool, auth_headers: dict[str, str]) -> bool: + def retry_invite(self, *, offer_srtp: bool, auth_headers: dict[str, str]) -> bool: """Build, register and send a fresh INVITE retry, chaining its result. Shared by [retry_with_auth][voip.sip.transactions.InviteTransaction.retry_with_auth] @@ -872,7 +874,7 @@ def _retry_invite(self, *, offer_srtp: bool, auth_headers: dict[str, str]) -> bo tx.pending_call_class = self.pending_call_class tx.pending_call_kwargs = self.pending_call_kwargs tx.offer_srtp = offer_srtp - tx.request = tx._build_invite_request(self.request.uri) + tx.request = tx.build_invite_request(self.request.uri) for header, value in auth_headers.items(): tx.request.headers[header] = value self.sip.register_transaction(tx) @@ -896,7 +898,7 @@ def response_received(self, response: Response) -> None: case 1: # trying/ringing return case 2: # OK - self._start_call(response) + self.start_call(response) # SRTP offer rejected as "not acceptable" → fall back to plain RTP. if self.offer_srtp and response.status_code in ( SIPStatus.NOT_ACCEPTABLE_HERE, # 488 @@ -909,7 +911,7 @@ def response_received(self, response: Response) -> None: self.ack(response) self.complete() - def _start_call(self, response: Response) -> None: + def start_call(self, response: Response) -> None: """Complete call setup after a 200 OK is received. Negotiates the codec from the remote SDP answer, creates the call diff --git a/voip/sip/types.py b/voip/sip/types.py index af833d0..897f4d1 100644 --- a/voip/sip/types.py +++ b/voip/sip/types.py @@ -150,17 +150,17 @@ def parse(cls, value: str) -> SipURI: 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"))) + parameters=dict(cls.parse_parameters(match.group("parameters"))) if match.group("parameters") else {}, - headers=dict(cls._parse_headers(match.group("headers")[1:])) + 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]]: + 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) @@ -169,7 +169,7 @@ def _parse_parameters(cls, params: str) -> Iterator[tuple[str, str | None]]: yield urllib.parse.unquote(part), None @classmethod - def _parse_headers(cls, headers: str) -> Iterator[tuple[str, str]]: + def parse_headers(cls, headers: str) -> Iterator[tuple[str, str]]: for part in headers.split("&"): if "=" in part: name, val = part.split("=", 1) @@ -486,7 +486,7 @@ class DigestQoP(enum.StrEnum): AUTH_INT = "auth-int" -def _mask_caller(header: str) -> str: +def mask_caller(header: str) -> str: """Return a privacy-safe label from a SIP From/To header value. Strips the `tag=` parameter, extracts the display name or SIP user part, @@ -494,9 +494,9 @@ def _mask_caller(header: str) -> str: Examples: ``` - >>> _mask_caller('"08001234567" ;tag=abc') + >>> mask_caller('"08001234567" ;tag=abc') '*******4567' - >>> _mask_caller('sip:alice@example.com') + >>> mask_caller('sip:alice@example.com') '*lice' ``` """ diff --git a/voip/srtp.py b/voip/srtp.py index 371c862..097d026 100644 --- a/voip/srtp.py +++ b/voip/srtp.py @@ -26,14 +26,14 @@ CIPHER_SUITE = "AES_CM_128_HMAC_SHA1_80" #: AES key size in bytes (128-bit). -_KEY_SIZE = 16 +KEY_SIZE = 16 #: Master salt size in bytes (112-bit). -_SALT_SIZE = 14 +SALT_SIZE = 14 #: HMAC-SHA1-80 authentication tag size in bytes. -_AUTH_TAG_SIZE = 10 +AUTH_TAG_SIZE = 10 -def _prf(master_key: bytes, label: int, master_salt: bytes, length: int) -> bytes: +def prf(master_key: bytes, label: int, master_salt: bytes, length: int) -> bytes: """Compute the SRTP pseudo-random function (RFC 3711 §4.3.1). Uses AES Counter Mode with IV derived from the label and master salt. @@ -76,27 +76,27 @@ class SRTPSession: master_key: bytes master_salt: bytes - _session_key: bytes = dataclasses.field(init=False) - _session_auth_key: bytes = dataclasses.field(init=False) - _session_salt: bytes = dataclasses.field(init=False) + session_key: bytes = dataclasses.field(init=False) + session_auth_key: bytes = dataclasses.field(init=False) + session_salt: bytes = dataclasses.field(init=False) #: Rollover counter and highest sent sequence number for encryption. - _send_roc: int = dataclasses.field(init=False, default=0) - _last_send_seq: int = dataclasses.field(init=False, default=-1) + send_roc: int = dataclasses.field(init=False, default=0) + last_send_seq: int = dataclasses.field(init=False, default=-1) #: Rollover counter and highest received sequence number for decryption. - _recv_roc: int = dataclasses.field(init=False, default=0) - _last_recv_seq: int = dataclasses.field(init=False, default=-1) + recv_roc: int = dataclasses.field(init=False, default=0) + last_recv_seq: int = dataclasses.field(init=False, default=-1) def __post_init__(self) -> None: - self._session_key = _prf(self.master_key, 0x00, self.master_salt, _KEY_SIZE) - self._session_auth_key = _prf(self.master_key, 0x01, self.master_salt, 20) - self._session_salt = _prf(self.master_key, 0x02, self.master_salt, _SALT_SIZE) + self.session_key = prf(self.master_key, 0x00, self.master_salt, KEY_SIZE) + self.session_auth_key = prf(self.master_key, 0x01, self.master_salt, 20) + self.session_salt = prf(self.master_key, 0x02, self.master_salt, SALT_SIZE) @classmethod def generate(cls) -> SRTPSession: """Generate a new SRTP session with a cryptographically random key and salt.""" return cls( - master_key=os.urandom(_KEY_SIZE), - master_salt=os.urandom(_SALT_SIZE), + master_key=os.urandom(KEY_SIZE), + master_salt=os.urandom(SALT_SIZE), ) @classmethod @@ -134,10 +134,7 @@ def from_sdes(cls, value: str) -> SRTPSession: if len(parts) < 3 or not parts[2].startswith("inline:"): raise ValueError(f"Malformed SDES crypto attribute: {value!r}") _tag, suite, key_params = parts[0], parts[1], parts[2] - # RFC 4568 §5.1.1: several inline key-params may be joined by ';'; - # the first is the active key, the rest are pre-rolled backups. inline = key_params.split(";", 1)[0] - # The inline parameter may carry trailing |lifetime or ~MKI suffixes. key_material_b64, _, _trailing = inline.removeprefix("inline:").partition("|") key_material_b64 = key_material_b64.split("~", 1)[0] if suite != CIPHER_SUITE: @@ -149,14 +146,14 @@ def from_sdes(cls, value: str) -> SRTPSession: key_material = base64.b64decode(key_material_b64, validate=True) except (ValueError, base64.binascii.Error) as exc: # noqa: PERF203 raise ValueError(f"Malformed SDES inline key: {inline!r}") from exc - if len(key_material) < _KEY_SIZE + _SALT_SIZE: + if len(key_material) < KEY_SIZE + SALT_SIZE: raise ValueError( f"SDES inline key too short ({len(key_material)} bytes); " - f"expected at least {_KEY_SIZE + _SALT_SIZE}" + f"expected at least {KEY_SIZE + SALT_SIZE}" ) return cls( - master_key=key_material[:_KEY_SIZE], - master_salt=key_material[_KEY_SIZE : _KEY_SIZE + _SALT_SIZE], + master_key=key_material[:KEY_SIZE], + master_salt=key_material[KEY_SIZE : KEY_SIZE + SALT_SIZE], ) @property @@ -174,27 +171,27 @@ def sdes_attribute(self) -> str: key_salt = base64.b64encode(self.master_key + self.master_salt).decode() return f"1 {CIPHER_SUITE} inline:{key_salt}" - def _compute_iv(self, ssrc: int, index: int) -> bytes: + def compute_iv(self, ssrc: int, index: int) -> bytes: """Compute the 128-bit AES-CM IV for a given SSRC and packet index. IV = (session_salt * 2^16) XOR (SSRC * 2^64) XOR (index * 2^16) per RFC 3711 §4.1.1. """ iv_int = ( - (int.from_bytes(self._session_salt, "big") << 16) + (int.from_bytes(self.session_salt, "big") << 16) ^ (ssrc << 64) ^ (index << 16) ) return iv_int.to_bytes(16, "big") - def _auth_tag(self, packet_no_tag: bytes, roc: int) -> bytes: + def auth_tag(self, packet_no_tag: bytes, roc: int) -> bytes: """Compute the 10-byte HMAC-SHA1 authentication tag (RFC 3711 §4.2).""" roc_bytes = struct.pack(">I", roc) - mac = hmac.HMAC(self._session_auth_key, hashes.SHA1()) # noqa: S303 + mac = hmac.HMAC(self.session_auth_key, hashes.SHA1()) # noqa: S303 mac.update(packet_no_tag + roc_bytes) - return mac.finalize()[:_AUTH_TAG_SIZE] + return mac.finalize()[:AUTH_TAG_SIZE] - def _estimate_recv_index(self, seq: int) -> tuple[int, int]: + def estimate_recv_index(self, seq: int) -> tuple[int, int]: """Estimate the packet index and new ROC for a received sequence number. Implements the index estimation algorithm from RFC 3711 §3.3.1. @@ -206,10 +203,9 @@ def _estimate_recv_index(self, seq: int) -> tuple[int, int]: A `(index, roc_guess)` tuple where `index` is the estimated 48-bit packet index and `roc_guess` is the ROC value used. """ - s_l = self._last_recv_seq - roc = self._recv_roc + s_l = self.last_recv_seq + roc = self.recv_roc if s_l < 0: - # No packets received yet; use the current ROC. return (roc << 16) | seq, roc if s_l < 0x8000: # s_l < 2^15 if seq - s_l > 0x8000: @@ -243,21 +239,18 @@ def encrypt(self, packet: bytes) -> bytes: ssrc = struct.unpack(">I", header[8:12])[0] seq = struct.unpack(">H", header[2:4])[0] - # Detect rollover: the sequence number wrapped from ~65535 back to ~0. - # For the send side, sequence numbers always increase monotonically so - # any decrease indicates a rollover. - if self._last_send_seq >= 0 and seq < self._last_send_seq: - self._send_roc = (self._send_roc + 1) % (1 << 32) - self._last_send_seq = seq + if self.last_send_seq >= 0 and seq < self.last_send_seq: + self.send_roc = (self.send_roc + 1) % (1 << 32) + self.last_send_seq = seq - index = (self._send_roc << 16) | seq - iv = self._compute_iv(ssrc, index) - cipher = Cipher(algorithms.AES(self._session_key), modes.CTR(iv)) + index = (self.send_roc << 16) | seq + iv = self.compute_iv(ssrc, index) + cipher = Cipher(algorithms.AES(self.session_key), modes.CTR(iv)) enc = cipher.encryptor() encrypted_payload = enc.update(payload) + enc.finalize() srtp_no_tag = header + encrypted_payload - return srtp_no_tag + self._auth_tag(srtp_no_tag, self._send_roc) + return srtp_no_tag + self.auth_tag(srtp_no_tag, self.send_roc) def decrypt(self, packet: bytes) -> bytes | None: """Decrypt and authenticate an SRTP packet. @@ -273,33 +266,31 @@ def decrypt(self, packet: bytes) -> bytes | None: Decrypted RTP packet bytes, or `None` when authentication fails or the packet is too short. """ - if len(packet) < 12 + _AUTH_TAG_SIZE: + if len(packet) < 12 + AUTH_TAG_SIZE: return None - srtp_no_tag = packet[:-_AUTH_TAG_SIZE] - received_tag = packet[-_AUTH_TAG_SIZE:] + srtp_no_tag = packet[:-AUTH_TAG_SIZE] + received_tag = packet[-AUTH_TAG_SIZE:] header = packet[:12] ssrc = struct.unpack(">I", header[8:12])[0] seq = struct.unpack(">H", header[2:4])[0] - index, roc_guess = self._estimate_recv_index(seq) + index, roc_guess = self.estimate_recv_index(seq) - expected_tag = self._auth_tag(srtp_no_tag, roc_guess) + expected_tag = self.auth_tag(srtp_no_tag, roc_guess) if not _hmac_stdlib.compare_digest(received_tag, expected_tag): return None - # Authentication passed — update the highest received sequence number - # and ROC per RFC 3711 §3.3.1. - if roc_guess == self._recv_roc: - if self._last_recv_seq < 0 or seq > self._last_recv_seq: - self._last_recv_seq = seq - elif roc_guess == (self._recv_roc + 1) % (1 << 32): - self._recv_roc = roc_guess - self._last_recv_seq = seq + if roc_guess == self.recv_roc: + if self.last_recv_seq < 0 or seq > self.last_recv_seq: + self.last_recv_seq = seq + elif roc_guess == (self.recv_roc + 1) % (1 << 32): + self.recv_roc = roc_guess + self.last_recv_seq = seq encrypted_payload = srtp_no_tag[12:] - iv = self._compute_iv(ssrc, index) - cipher = Cipher(algorithms.AES(self._session_key), modes.CTR(iv)) + iv = self.compute_iv(ssrc, index) + cipher = Cipher(algorithms.AES(self.session_key), modes.CTR(iv)) dec = cipher.decryptor() payload = dec.update(encrypted_payload) + dec.finalize() return header + payload diff --git a/voip/stun.py b/voip/stun.py index 422ce5e..cfa69e7 100644 --- a/voip/stun.py +++ b/voip/stun.py @@ -32,7 +32,7 @@ class STUNAttributeType(enum.IntEnum): XOR_MAPPED_ADDRESS = 0x0020 -def _parse_address( +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. @@ -115,7 +115,7 @@ def packet_received(self, data: bytes, addr: tuple[str, int]) -> None: stun_server_address: NetworkAddress | None = NetworkAddress( "stun.cloudflare.com", 3478 ) - _stun_transaction_id: bytes = dataclasses.field(init=False, default=b"") + stun_transaction_id: bytes = dataclasses.field(init=False, default=b"") transport: asyncio.DatagramTransport | None = dataclasses.field( init=False, default=None ) @@ -128,8 +128,8 @@ def connection_made(self, transport: asyncio.DatagramTransport) -> None: transport, NetworkAddress(host=ipaddress.ip_address(host), port=port) ) else: - self._stun_transaction_id = uuid.uuid4().bytes[:12] - self._send_stun_request() + self.stun_transaction_id = uuid.uuid4().bytes[:12] + self.send_stun_request() def stun_connection_made( self, @@ -173,10 +173,10 @@ def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: # RFC 7983: first byte in [0, 3] indicates a STUN packet. if ( len(data) >= 20 - and self._stun_transaction_id - and data[8:20] == self._stun_transaction_id + and self.stun_transaction_id + and data[8:20] == self.stun_transaction_id ): - self._parse_stun_response(data) + self.parse_stun_response(data) return self.packet_received(data, NetworkAddress(*addr[:2])) @@ -195,7 +195,7 @@ def packet_received(self, data: bytes, addr: NetworkAddress) -> None: addr: Source `(host, port)` of the datagram. """ - def _send_stun_request(self) -> None: + def send_stun_request(self) -> None: """Send a STUN Binding Request through the protocol's own transport. Sends the request through the transport bound to this protocol so the @@ -209,12 +209,12 @@ def _send_stun_request(self) -> None: STUNMessageType.BINDING_REQUEST, 0, MAGIC_COOKIE, - self._stun_transaction_id, + self.stun_transaction_id, ) logger.debug("Sending STUN Binding Request to %s:%s", *self.stun_server_address) self.transport.sendto(request, self.stun_server_address) - def _parse_stun_response(self, data: bytes) -> None: + def parse_stun_response(self, data: bytes) -> None: """Parse a STUN Binding Success Response and invoke :meth:`stun_connection_made`.""" logger.debug("Parsing STUN response (len=%d)", len(data)) if len(data) < 20: @@ -224,11 +224,10 @@ def _parse_stun_response(self, data: bytes) -> None: if ( magic_cookie != MAGIC_COOKIE or message_type != STUNMessageType.BINDING_SUCCESS_RESPONSE - or response_tid != self._stun_transaction_id + or response_tid != self.stun_transaction_id ): return - # Clear transaction ID so duplicate responses are ignored. - self._stun_transaction_id = b"" + self.stun_transaction_id = b"" offset = 20 xor_mapped: tuple[ipaddress.IPv4Address | ipaddress.IPv6Address, int] | None = ( None @@ -242,9 +241,9 @@ def _parse_stun_response(self, data: bytes) -> None: attribute_value = data[offset + 4 : offset + 4 + attribute_len] match attribute_type: case STUNAttributeType.XOR_MAPPED_ADDRESS: - xor_mapped = _parse_address(attribute_value, xor_key) + xor_mapped = parse_address(attribute_value, xor_key) case STUNAttributeType.MAPPED_ADDRESS: - mapped = _parse_address(attribute_value, b"") + mapped = parse_address(attribute_value, b"") offset += 4 + ((attribute_len + 3) & ~3) # 4-byte aligned try: host, port = xor_mapped or mapped