From 3deb87c12bab67efccdf0536e950ab734f3fec15 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 26 Mar 2026 12:14:23 +0000 Subject: [PATCH 1/4] Initial plan From 8c955f84754562be772b7936edb24639f77ec9c2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 26 Mar 2026 12:30:22 +0000 Subject: [PATCH 2/4] Add TCP stream buffering to data_received via extract_frames/dispatch_frame Co-authored-by: codingjoe <1772890+codingjoe@users.noreply.github.com> Agent-Logs-Url: https://github.com/codingjoe/VoIP/sessions/6938e2c0-d1ba-4740-8f03-50b5ed9a8b7b --- tests/sip/test_protocol.py | 524 +++++++++++++++++++++++++++++++++++++ voip/sip/protocol.py | 124 ++++++--- 2 files changed, 608 insertions(+), 40 deletions(-) diff --git a/tests/sip/test_protocol.py b/tests/sip/test_protocol.py index 60b8bbf..06e3d2f 100644 --- a/tests/sip/test_protocol.py +++ b/tests/sip/test_protocol.py @@ -4,11 +4,13 @@ import asyncio import datetime +import ipaddress from voip.sip.messages import Message, Response from voip.sip.protocol import SessionInitiationProtocol from voip.sip.transactions import InviteTransaction from voip.sip.types import SIPMethod, SipUri +from voip.types import NetworkAddress from .conftest import INVITE_BYTES, FakeTransport @@ -333,3 +335,525 @@ def test_connection_lost__without_keepalive_task(self, sip): sip.keepalive_task = None sip.connection_lost(None) assert sip.transport is None + + # ------------------------------------------------------------------ + # Helpers shared by extract_frames / dispatch_frame tests + # ------------------------------------------------------------------ + + def _make_session(self, rtp, fake_transport=None): + """Return a SessionInitiationProtocol wired with a fake transport. + + Does **not** call `connection_made` to avoid triggering + `RegistrationTransaction.__post_init__` inside a running event loop, + which raises `TypeError` due to a `super()` / `slots=True` interaction + in Python 3.13. + """ + session = SessionInitiationProtocol( + aor=SipUri.parse("sips:alice@example.com"), + rtp=rtp, + transaction_class=InviteTransaction, + ) + transport = fake_transport or FakeTransport() + session.transport = transport + session.local_address = NetworkAddress(ipaddress.ip_address("127.0.0.1"), 5061) + return session + + # ------------------------------------------------------------------ + # extract_frames + # ------------------------------------------------------------------ + + def test_extract_frames__empty_buffer(self, rtp): + """Return an empty list when the receive buffer is empty.""" + session = self._make_session(rtp) + assert session.extract_frames() == [] + + def test_extract_frames__complete_message(self, rtp): + """Extract a single complete SIP message and clear the buffer.""" + session = self._make_session(rtp) + session.recv_buffer.extend(INVITE_BYTES) + frames = session.extract_frames() + assert frames == [INVITE_BYTES] + assert len(session.recv_buffer) == 0 + + def test_extract_frames__partial_headers(self, rtp): + """Keep partial message bytes in the buffer until complete.""" + session = self._make_session(rtp) + session.recv_buffer.extend(INVITE_BYTES[:20]) + assert session.extract_frames() == [] + assert len(session.recv_buffer) == 20 + + def test_extract_frames__two_coalesced_messages(self, rtp): + """Extract two SIP messages delivered in a single TCP segment.""" + second = ( + b"OPTIONS sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKopt1\r\n" + b"From: sip:bob@biloxi.com;tag=t99\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: opt-coalesced@biloxi.com\r\n" + b"CSeq: 1 OPTIONS\r\n" + b"\r\n" + ) + session = self._make_session(rtp) + session.recv_buffer.extend(INVITE_BYTES + second) + frames = session.extract_frames() + assert len(frames) == 2 + assert frames[0] == INVITE_BYTES + assert frames[1] == second + assert len(session.recv_buffer) == 0 + + def test_extract_frames__message_with_body(self, rtp): + """Extract a SIP message that includes a Content-Length body.""" + body = b"v=0\r\no=- 1 1 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\n" + headers = ( + b"INVITE sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKbody1\r\n" + b"From: sip:bob@biloxi.com;tag=tb1\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: body-call@biloxi.com\r\n" + b"CSeq: 1 INVITE\r\n" + b"Content-Type: application/sdp\r\n" + b"Content-Length: " + str(len(body)).encode() + b"\r\n" + b"\r\n" + ) + message = headers + body + session = self._make_session(rtp) + session.recv_buffer.extend(message) + frames = session.extract_frames() + assert frames == [message] + assert len(session.recv_buffer) == 0 + + def test_extract_frames__incomplete_body(self, rtp): + """Keep bytes in the buffer when only part of the body has arrived.""" + body = b"v=0\r\no=- 1 1 IN IP4 127.0.0.1\r\n" + headers = ( + b"INVITE sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKbody2\r\n" + b"From: sip:bob@biloxi.com;tag=tb2\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: body2-call@biloxi.com\r\n" + b"CSeq: 1 INVITE\r\n" + b"Content-Length: " + str(len(body)).encode() + b"\r\n" + b"\r\n" + ) + session = self._make_session(rtp) + session.recv_buffer.extend(headers + body[:5]) + assert session.extract_frames() == [] + assert len(session.recv_buffer) == len(headers) + 5 + + def test_extract_frames__ping(self, rtp): + """Extract an RFC 5626 PING (CRLF CRLF) keepalive frame.""" + session = self._make_session(rtp) + session.recv_buffer.extend(b"\r\n\r\n") + frames = session.extract_frames() + assert frames == [b"\r\n\r\n"] + assert len(session.recv_buffer) == 0 + + def test_extract_frames__pong(self, rtp): + """Extract an RFC 5626 PONG (CRLF) keepalive frame.""" + session = self._make_session(rtp) + session.recv_buffer.extend(b"\r\n") + frames = session.extract_frames() + assert frames == [b"\r\n"] + assert len(session.recv_buffer) == 0 + + def test_extract_frames__partial_keepalive_wait(self, rtp): + """Buffer a partial PING (CRLF CR) without dispatching until the 4th byte arrives.""" + session = self._make_session(rtp) + session.recv_buffer.extend(b"\r\n\r") + assert session.extract_frames() == [] + assert len(session.recv_buffer) == 3 + + def test_extract_frames__ping_followed_by_message(self, rtp): + """Extract a PING and a SIP message from the same buffer.""" + session = self._make_session(rtp) + session.recv_buffer.extend(b"\r\n\r\n" + INVITE_BYTES) + frames = session.extract_frames() + assert len(frames) == 2 + assert frames[0] == b"\r\n\r\n" + assert frames[1] == INVITE_BYTES + + def test_extract_frames__pong_followed_by_message(self, rtp): + """Extract a PONG and a SIP message coalesced in the same buffer.""" + session = self._make_session(rtp) + session.recv_buffer.extend(b"\r\n" + INVITE_BYTES) + frames = session.extract_frames() + assert len(frames) == 2 + assert frames[0] == b"\r\n" + assert frames[1] == INVITE_BYTES + + def test_extract_frames__invalid_content_length(self, rtp): + """Treat an unparseable Content-Length as zero (no body).""" + message = ( + b"OPTIONS sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKcl0\r\n" + b"From: sip:bob@biloxi.com;tag=tclx\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: clx-call@biloxi.com\r\n" + b"CSeq: 1 OPTIONS\r\n" + b"Content-Length: notanumber\r\n" + b"\r\n" + ) + session = self._make_session(rtp) + session.recv_buffer.extend(message) + frames = session.extract_frames() + assert len(frames) == 1 + + def test_extract_frames__single_cr_waits(self, rtp): + """Keep a lone CR in the buffer without dispatching (incomplete keepalive).""" + session = self._make_session(rtp) + session.recv_buffer.extend(b"\r") + assert session.extract_frames() == [] + assert session.recv_buffer == bytearray(b"\r") + + # ------------------------------------------------------------------ + # dispatch_frame + # ------------------------------------------------------------------ + + def test_dispatch_frame__pong(self, rtp, fake_transport, caplog): + """Log PONG on receiving a CRLF frame without sending a reply.""" + import logging + + session = self._make_session(rtp, fake_transport) + with caplog.at_level(logging.INFO): + session.dispatch_frame(b"\r\n") + assert "PONG" in caplog.text + assert fake_transport.sent == [] + + def test_dispatch_frame__ping__sends_pong(self, rtp, fake_transport): + """Reply with a CRLF PONG when a PING (CRLF CRLF) frame is dispatched.""" + session = self._make_session(rtp, fake_transport) + session.dispatch_frame(b"\r\n\r\n") + assert b"\r\n" in fake_transport.sent + + def test_dispatch_frame__sip_request(self, rtp, fake_transport): + """Dispatch a SIP request frame to request_received.""" + session = self._make_session(rtp, fake_transport) + before = len(session.transactions) + session.dispatch_frame(INVITE_BYTES) + assert len(session.transactions) > before + + def test_dispatch_frame__sip_response(self, rtp, fake_transport): + """Dispatch a SIP response frame to response_received without error.""" + session = self._make_session(rtp, fake_transport) + branch = "z9hG4bKresp-test" + session.transactions[branch] = InviteTransaction( + sip=session, + method=SIPMethod.INVITE, + branch=branch, + cseq=1, + ) + response_bytes = ( + f"SIP/2.0 200 OK\r\n" + f"Via: SIP/2.0/TLS example.com;branch={branch}\r\n" + f"From: sip:alice@example.com;tag=local\r\n" + f"To: sip:example.com;tag=remote\r\n" + f"Call-ID: resp-test@example.com\r\n" + f"CSeq: 1 INVITE\r\n" + f"\r\n" + ).encode() + session.dispatch_frame(response_bytes) + + # ------------------------------------------------------------------ + # data_received – stream reassembly + # ------------------------------------------------------------------ + + def test_data_received__split_message(self, rtp, fake_transport): + """Reassemble a SIP request delivered in two TCP segments.""" + session = self._make_session(rtp, fake_transport) + split = len(INVITE_BYTES) // 2 + before = len(session.transactions) + session.data_received(INVITE_BYTES[:split]) + assert len(session.transactions) == before # incomplete – not dispatched yet + session.data_received(INVITE_BYTES[split:]) + assert len(session.transactions) > before # now dispatched + + def test_data_received__coalesced_messages(self, rtp, fake_transport): + """Dispatch two SIP requests coalesced into one TCP segment.""" + second = ( + b"OPTIONS sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKopt2\r\n" + b"From: sip:bob@biloxi.com;tag=t88\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: opt-coalesced2@biloxi.com\r\n" + b"CSeq: 2 OPTIONS\r\n" + b"\r\n" + ) + session = self._make_session(rtp, fake_transport) + before = len(session.transactions) + session.data_received(INVITE_BYTES + second) + # INVITE creates a transaction; OPTIONS is answered directly (no tx added) + assert len(session.transactions) > before + + def test_data_received__body_split_across_segments(self, rtp, fake_transport): + """Reassemble a SIP request with a body split across two TCP segments.""" + body = b"v=0\r\no=- 1 1 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\n" + headers = ( + b"INVITE sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKbodysplit\r\n" + b"From: sip:bob@biloxi.com;tag=tbs\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: body-split@biloxi.com\r\n" + b"CSeq: 1 INVITE\r\n" + b"Content-Type: application/sdp\r\n" + b"Content-Length: " + str(len(body)).encode() + b"\r\n" + b"\r\n" + ) + session = self._make_session(rtp, fake_transport) + before = len(session.transactions) + session.data_received(headers + body[:5]) + assert len(session.transactions) == before # body incomplete + session.data_received(body[5:]) + assert len(session.transactions) > before # body complete, dispatched + + # ------------------------------------------------------------------ + # send_keepalive (via _make_session to avoid RegistrationTransaction) + # ------------------------------------------------------------------ + + async def test_send_keepalive__sends_ping_without_sip_fixture(self, rtp): + """send_keepalive writes a PING (CRLF CRLF) after the interval elapses.""" + fake_transport = FakeTransport() + session = self._make_session(rtp, fake_transport) + session.keepalive_interval = datetime.timedelta(milliseconds=10) + task = asyncio.create_task(session.send_keepalive()) + await asyncio.sleep(0.05) + task.cancel() + assert b"\r\n\r\n" in fake_transport.sent + + async def test_send_keepalive__stops_when_transport_cleared(self, rtp): + """send_keepalive exits when transport is set to None.""" + session = self._make_session(rtp) + session.transport = None + session.keepalive_interval = datetime.timedelta(milliseconds=1) + await session.send_keepalive() + + # ------------------------------------------------------------------ + # send / close (via _make_session) + # ------------------------------------------------------------------ + + def test_send__writes_bytes_without_sip_fixture(self, rtp, fake_transport): + """send() serialises and writes a SIP message to the transport.""" + session = self._make_session(rtp, fake_transport) + response = Response(status_code=200, phrase="OK") + session.send(response) + assert bytes(response) in fake_transport.sent + + def test_send__no_op_when_transport_is_none(self, rtp): + """send() is a no-op when transport is None.""" + session = self._make_session(rtp) + session.transport = None + session.send(Response(status_code=200, phrase="OK")) + + def test_close__closes_transport_without_sip_fixture(self, rtp, fake_transport): + """close() closes the underlying transport.""" + session = self._make_session(rtp, fake_transport) + session.close() + assert fake_transport.closed is True + + def test_close__no_op_when_transport_is_none(self, rtp): + """close() is a no-op when transport is None.""" + session = self._make_session(rtp) + session.transport = None + session.close() + + # ------------------------------------------------------------------ + # allowed_methods / allow_header / method_not_allowed (via _make_session) + # ------------------------------------------------------------------ + + def test_allowed_methods__includes_options_without_sip_fixture(self, rtp): + """OPTIONS is always included in allowed_methods.""" + session = self._make_session(rtp) + assert SIPMethod.OPTIONS in session.allowed_methods + + def test_allow_header__is_comma_separated_without_sip_fixture(self, rtp): + """allow_header returns a comma-separated string of methods.""" + session = self._make_session(rtp) + header = session.allow_header + assert "OPTIONS" in header + assert "," in header + + def test_method_not_allowed__sends_405_without_sip_fixture(self, rtp, fake_transport): + """method_not_allowed() sends a 405 Method Not Allowed response.""" + session = self._make_session(rtp, fake_transport) + request = Message.parse( + b"PUBLISH sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKpub3\r\n" + b"From: sip:bob@biloxi.com;tag=t5\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: pub3-call@biloxi.com\r\n" + b"CSeq: 1 PUBLISH\r\n" + b"\r\n" + ) + session.method_not_allowed(request) + assert any(b"405" in data for data in fake_transport.sent) + + # ------------------------------------------------------------------ + # request_received (via _make_session) + # ------------------------------------------------------------------ + + def test_request_received__options_sends_200_without_sip_fixture( + self, rtp, fake_transport + ): + """OPTIONS request is answered with 200 OK.""" + session = self._make_session(rtp, fake_transport) + request = Message.parse( + b"OPTIONS sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKopt3\r\n" + b"From: sip:bob@biloxi.com;tag=t6\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: opt3-call@biloxi.com\r\n" + b"CSeq: 3 OPTIONS\r\n" + b"\r\n" + ) + session.request_received(request) + assert any(b"200" in data for data in fake_transport.sent) + + def test_request_received__invite_creates_transaction_without_sip_fixture( + self, rtp, fake_transport + ): + """INVITE request creates an InviteTransaction.""" + session = self._make_session(rtp, fake_transport) + request = Message.parse(INVITE_BYTES) + before = len(session.transactions) + session.request_received(request) + assert len(session.transactions) > before + + def test_request_received__unsupported_method_sends_405_without_sip_fixture( + self, rtp, fake_transport + ): + """Unsupported method triggers method_not_allowed (405).""" + session = self._make_session(rtp, fake_transport) + request = Message.parse( + b"PUBLISH sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKpub4\r\n" + b"From: sip:bob@biloxi.com;tag=t7\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: pub4-call@biloxi.com\r\n" + b"CSeq: 1 PUBLISH\r\n" + b"\r\n" + ) + session.request_received(request) + assert any(b"405" in data for data in fake_transport.sent) + + def test_request_received__cancel_dispatches_to_transaction_without_sip_fixture( + self, rtp, fake_transport + ): + """CANCEL is forwarded to the matching INVITE transaction.""" + session = self._make_session(rtp, fake_transport) + invite = Message.parse(INVITE_BYTES) + session.request_received(invite) + tx = session.transactions[invite.branch] + session.dialogs[(tx.dialog.remote_tag, tx.dialog.local_tag)] = tx.dialog + cancel = Message.parse( + b"CANCEL sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKabc123\r\n" + b"From: sip:bob@biloxi.com;tag=from-tag-1\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: test-call-id@biloxi.com\r\n" + b"CSeq: 1 CANCEL\r\n" + b"\r\n" + ) + session.request_received(cancel) + assert any(b"200" in data for data in fake_transport.sent) + + def test_request_received__cancel_gone_when_no_transaction_without_sip_fixture( + self, rtp, fake_transport + ): + """CANCEL with no matching transaction returns 410 Gone.""" + session = self._make_session(rtp, fake_transport) + cancel = Message.parse( + b"CANCEL sip:alice@example.com SIP/2.0\r\n" + b"Via: SIP/2.0/TLS 192.0.2.1:5061;branch=z9hG4bKnone2\r\n" + b"From: sip:bob@biloxi.com;tag=t8\r\n" + b"To: sip:alice@example.com\r\n" + b"Call-ID: no-tx2@biloxi.com\r\n" + b"CSeq: 1 CANCEL\r\n" + b"\r\n" + ) + session.request_received(cancel) + assert any(b"410" in data for data in fake_transport.sent) + + # ------------------------------------------------------------------ + # response_received (via _make_session) + # ------------------------------------------------------------------ + + def test_response_received__delegates_to_transaction_without_sip_fixture( + self, rtp, fake_transport + ): + """Response is forwarded to the matching transaction.""" + session = self._make_session(rtp, fake_transport) + branch = "z9hG4bKdel-test" + session.transactions[branch] = InviteTransaction( + sip=session, + method=SIPMethod.INVITE, + branch=branch, + cseq=1, + ) + response = Message.parse( + f"SIP/2.0 200 OK\r\n" + f"Via: SIP/2.0/TLS example.com;branch={branch}\r\n" + f"From: sip:alice@example.com;tag=lt\r\n" + f"To: sip:example.com;tag=rt\r\n" + f"Call-ID: del-test@example.com\r\n" + f"CSeq: 1 INVITE\r\n" + f"\r\n".encode() + ) + session.response_received(response) + + def test_response_received__warns_on_unknown_branch_without_sip_fixture( + self, rtp, fake_transport, caplog + ): + """Log a warning when the response branch is not in transactions.""" + import logging + + session = self._make_session(rtp, fake_transport) + response = Message.parse( + b"SIP/2.0 200 OK\r\n" + b"Via: SIP/2.0/TLS example.com;branch=z9hG4bKunknown\r\n" + b"From: sip:alice@example.com;tag=lt2\r\n" + b"To: sip:example.com;tag=rt2\r\n" + b"Call-ID: unknown-branch@example.com\r\n" + b"CSeq: 1 INVITE\r\n" + b"\r\n" + ) + with caplog.at_level(logging.WARNING): + session.response_received(response) + assert "unknown branch" in caplog.text + + # ------------------------------------------------------------------ + # connection_lost (via _make_session) + # ------------------------------------------------------------------ + + async def test_connection_lost__cancels_keepalive_without_sip_fixture(self, rtp): + """connection_lost() cancels and clears the keepalive task.""" + session = self._make_session(rtp) + session.keepalive_task = asyncio.create_task(asyncio.sleep(9999)) + session.connection_lost(None) + assert session.keepalive_task is None + + def test_connection_lost__clears_transport_without_sip_fixture(self, rtp): + """connection_lost() sets transport to None.""" + session = self._make_session(rtp) + session.connection_lost(None) + assert session.transport is None + + def test_connection_lost__sets_disconnected_event_without_sip_fixture(self, rtp): + """connection_lost() sets the disconnected_event.""" + session = self._make_session(rtp) + session.connection_lost(None) + assert session.disconnected_event.is_set() + + def test_connection_lost__logs_exception_without_sip_fixture(self, rtp, caplog): + """connection_lost() logs an error when an exception is provided.""" + import logging + + session = self._make_session(rtp) + with caplog.at_level(logging.ERROR): + session.connection_lost(OSError("reset")) + assert session.transport is None + + def test_connection_lost__no_keepalive_task_without_sip_fixture(self, rtp): + """connection_lost() is safe when keepalive_task is None.""" + session = self._make_session(rtp) + session.keepalive_task = None + session.connection_lost(None) + assert session.transport is None diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index 783cb8b..58549dd 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -100,6 +100,7 @@ class MySession(SessionInitiationProtocol): ) transport: asyncio.Transport | None = dataclasses.field(init=False, default=None) is_secure: bool = dataclasses.field(init=False, default=False) + recv_buffer: bytearray = dataclasses.field(init=False, default_factory=bytearray) def connection_made(self, transport: asyncio.Transport) -> None: # type: ignore[override] """Store the TLS/TCP transport and start RTP mux + carrier registration.""" @@ -126,53 +127,96 @@ async def send_keepalive(self) -> None: self.transport.write(b"\r\n\r\n") def data_received(self, data: bytes) -> None: - match data: + """Buffer incoming bytes and dispatch each complete frame.""" + self.recv_buffer.extend(data) + for frame in self.extract_frames(): + self.dispatch_frame(frame) + + def extract_frames(self) -> list[bytes]: + """Extract complete SIP messages and keepalive frames from [`recv_buffer`][voip.sip.protocol.SessionInitiationProtocol.recv_buffer]. + + TCP is a stream protocol; a single [`data_received`][voip.sip.protocol.SessionInitiationProtocol.data_received] + call may carry a partial message, an exact message, or several coalesced messages. + This method uses the `Content-Length` header for body framing per + [RFC 3261 §18.3](https://datatracker.ietf.org/doc/html/rfc3261#section-18.3) + and recognises [RFC 5626](https://datatracker.ietf.org/doc/html/rfc5626) + PING (`\\r\\n\\r\\n`) and PONG (`\\r\\n`) keepalive sequences. + + Each call mutates [`recv_buffer`][voip.sip.protocol.SessionInitiationProtocol.recv_buffer] + in-place, consuming only the bytes that belong to complete frames. + + Returns: + List of complete frame byte sequences ready for parsing or keepalive dispatch. + """ + frames = [] + while self.recv_buffer: + if self.recv_buffer[0:1] != b"\r": + # SIP message: wait for the header-body separator. + header_end = self.recv_buffer.find(b"\r\n\r\n") + if header_end == -1: + break # incomplete headers – wait for more data + content_length = 0 + for line in self.recv_buffer[:header_end].split(b"\r\n")[1:]: + name, sep, value = line.partition(b":") + if sep and name.strip().lower() == b"content-length": + try: + content_length = int(value.strip()) + except ValueError: + pass + break + message_end = header_end + 4 + content_length + if len(self.recv_buffer) < message_end: + break # incomplete body – wait for more data + frames.append(bytes(self.recv_buffer[:message_end])) + del self.recv_buffer[:message_end] + elif len(self.recv_buffer) >= 4 and self.recv_buffer[:4] == b"\r\n\r\n": + # RFC 5626 PING + frames.append(b"\r\n\r\n") + del self.recv_buffer[:4] + elif len(self.recv_buffer) >= 3 and self.recv_buffer[2:3] == b"\r": + # Third byte is CR – could be the start of \r\n\r\n; wait for 4th byte. + break + elif self.recv_buffer[:2] == b"\r\n": + # RFC 5626 PONG (2-byte sequence confirmed not to be a PING prefix). + frames.append(b"\r\n") + del self.recv_buffer[:2] + else: + # Single CR or other incomplete sequence – wait for more data. + break + return frames + + def dispatch_frame(self, frame: bytes) -> None: + """Dispatch a single complete frame (SIP message or keepalive). + + Args: + frame: A complete frame as returned by + [`extract_frames`][voip.sip.protocol.SessionInitiationProtocol.extract_frames]. + """ + peer = NetworkAddress(*self.transport.get_extra_info("peername")) + match frame: case b"\r\n": - logger.info( - "PONG", - extra={ - "addr": NetworkAddress( - *self.transport.get_extra_info("peername") - ) - }, - ) - return + logger.info("PONG", extra={"addr": peer}) case b"\r\n\r\n": - logger.info( - "PING", - extra={ - "addr": NetworkAddress( - *self.transport.get_extra_info("peername") - ) - }, - ) + logger.info("PING", extra={"addr": peer}) if self.transport: logger.info("PONG", extra={"addr": self.local_address}) self.transport.write(b"\r\n") - return - match Message.parse(data): - case Request() as request: - logger.info( - "Request received: %r", - request, - extra={ - "addr": NetworkAddress( - *self.transport.get_extra_info("peername") + case _: + match Message.parse(frame): + case Request() as request: + logger.info( + "Request received: %r", + request, + extra={"addr": peer}, ) - }, - ) - self.request_received(request) - case Response() as response: - logger.info( - "Response received %r", - response, - extra={ - "addr": NetworkAddress( - *self.transport.get_extra_info("peername") + self.request_received(request) + case Response() as response: + logger.info( + "Response received %r", + response, + extra={"addr": peer}, ) - }, - ) - self.response_received(response) + self.response_received(response) def send(self, message: Response | Request) -> None: """Serialize and send a SIP message over the TLS/TCP connection.""" From 78daf19325e80b10dbe2bf0c32e52dd9c423d8c2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 26 Mar 2026 14:34:38 +0000 Subject: [PATCH 3/4] Named PING/PONG constants + extract_frames generator yielding memoryview Co-authored-by: codingjoe <1772890+codingjoe@users.noreply.github.com> Agent-Logs-Url: https://github.com/codingjoe/VoIP/sessions/6d0e224a-c1c2-494c-ad1c-2e90bd633722 --- tests/sip/test_protocol.py | 46 +++++++++--------- voip/sip/protocol.py | 99 ++++++++++++++++++++++---------------- 2 files changed, 81 insertions(+), 64 deletions(-) diff --git a/tests/sip/test_protocol.py b/tests/sip/test_protocol.py index 06e3d2f..33892a4 100644 --- a/tests/sip/test_protocol.py +++ b/tests/sip/test_protocol.py @@ -7,7 +7,7 @@ import ipaddress from voip.sip.messages import Message, Response -from voip.sip.protocol import SessionInitiationProtocol +from voip.sip.protocol import PING, PONG, SessionInitiationProtocol from voip.sip.transactions import InviteTransaction from voip.sip.types import SIPMethod, SipUri from voip.types import NetworkAddress @@ -363,15 +363,15 @@ def _make_session(self, rtp, fake_transport=None): # ------------------------------------------------------------------ def test_extract_frames__empty_buffer(self, rtp): - """Return an empty list when the receive buffer is empty.""" + """Return an empty iterator when the receive buffer is empty.""" session = self._make_session(rtp) - assert session.extract_frames() == [] + assert [bytes(f) for f in session.extract_frames()] == [] def test_extract_frames__complete_message(self, rtp): """Extract a single complete SIP message and clear the buffer.""" session = self._make_session(rtp) session.recv_buffer.extend(INVITE_BYTES) - frames = session.extract_frames() + frames = [bytes(f) for f in session.extract_frames()] assert frames == [INVITE_BYTES] assert len(session.recv_buffer) == 0 @@ -379,7 +379,7 @@ def test_extract_frames__partial_headers(self, rtp): """Keep partial message bytes in the buffer until complete.""" session = self._make_session(rtp) session.recv_buffer.extend(INVITE_BYTES[:20]) - assert session.extract_frames() == [] + assert [bytes(f) for f in session.extract_frames()] == [] assert len(session.recv_buffer) == 20 def test_extract_frames__two_coalesced_messages(self, rtp): @@ -395,7 +395,7 @@ def test_extract_frames__two_coalesced_messages(self, rtp): ) session = self._make_session(rtp) session.recv_buffer.extend(INVITE_BYTES + second) - frames = session.extract_frames() + frames = [bytes(f) for f in session.extract_frames()] assert len(frames) == 2 assert frames[0] == INVITE_BYTES assert frames[1] == second @@ -418,7 +418,7 @@ def test_extract_frames__message_with_body(self, rtp): message = headers + body session = self._make_session(rtp) session.recv_buffer.extend(message) - frames = session.extract_frames() + frames = [bytes(f) for f in session.extract_frames()] assert frames == [message] assert len(session.recv_buffer) == 0 @@ -437,48 +437,48 @@ def test_extract_frames__incomplete_body(self, rtp): ) session = self._make_session(rtp) session.recv_buffer.extend(headers + body[:5]) - assert session.extract_frames() == [] + assert [bytes(f) for f in session.extract_frames()] == [] assert len(session.recv_buffer) == len(headers) + 5 def test_extract_frames__ping(self, rtp): """Extract an RFC 5626 PING (CRLF CRLF) keepalive frame.""" session = self._make_session(rtp) - session.recv_buffer.extend(b"\r\n\r\n") - frames = session.extract_frames() - assert frames == [b"\r\n\r\n"] + session.recv_buffer.extend(PING) + frames = [bytes(f) for f in session.extract_frames()] + assert frames == [PING] assert len(session.recv_buffer) == 0 def test_extract_frames__pong(self, rtp): """Extract an RFC 5626 PONG (CRLF) keepalive frame.""" session = self._make_session(rtp) - session.recv_buffer.extend(b"\r\n") - frames = session.extract_frames() - assert frames == [b"\r\n"] + session.recv_buffer.extend(PONG) + frames = [bytes(f) for f in session.extract_frames()] + assert frames == [PONG] assert len(session.recv_buffer) == 0 def test_extract_frames__partial_keepalive_wait(self, rtp): """Buffer a partial PING (CRLF CR) without dispatching until the 4th byte arrives.""" session = self._make_session(rtp) session.recv_buffer.extend(b"\r\n\r") - assert session.extract_frames() == [] + assert [bytes(f) for f in session.extract_frames()] == [] assert len(session.recv_buffer) == 3 def test_extract_frames__ping_followed_by_message(self, rtp): """Extract a PING and a SIP message from the same buffer.""" session = self._make_session(rtp) - session.recv_buffer.extend(b"\r\n\r\n" + INVITE_BYTES) - frames = session.extract_frames() + session.recv_buffer.extend(PING + INVITE_BYTES) + frames = [bytes(f) for f in session.extract_frames()] assert len(frames) == 2 - assert frames[0] == b"\r\n\r\n" + assert frames[0] == PING assert frames[1] == INVITE_BYTES def test_extract_frames__pong_followed_by_message(self, rtp): """Extract a PONG and a SIP message coalesced in the same buffer.""" session = self._make_session(rtp) - session.recv_buffer.extend(b"\r\n" + INVITE_BYTES) - frames = session.extract_frames() + session.recv_buffer.extend(PONG + INVITE_BYTES) + frames = [bytes(f) for f in session.extract_frames()] assert len(frames) == 2 - assert frames[0] == b"\r\n" + assert frames[0] == PONG assert frames[1] == INVITE_BYTES def test_extract_frames__invalid_content_length(self, rtp): @@ -495,14 +495,14 @@ def test_extract_frames__invalid_content_length(self, rtp): ) session = self._make_session(rtp) session.recv_buffer.extend(message) - frames = session.extract_frames() + frames = [bytes(f) for f in session.extract_frames()] assert len(frames) == 1 def test_extract_frames__single_cr_waits(self, rtp): """Keep a lone CR in the buffer without dispatching (incomplete keepalive).""" session = self._make_session(rtp) session.recv_buffer.extend(b"\r") - assert session.extract_frames() == [] + assert [bytes(f) for f in session.extract_frames()] == [] assert session.recv_buffer == bytearray(b"\r") # ------------------------------------------------------------------ diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index 58549dd..b4bb595 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -26,6 +26,12 @@ logger = logging.getLogger("voip.sip") +#: RFC 5626 §4.4 keepalive PING sequence. +PING: typing.Final[bytes] = b"\r\n\r\n" + +#: RFC 5626 §4.4 keepalive PONG reply. +PONG: typing.Final[bytes] = b"\r\n" + __all__ = [ "SIP", "SessionInitiationProtocol", @@ -124,7 +130,7 @@ async def send_keepalive(self) -> None: if self.transport is None: return logger.info("PING", extra={"addr": self.local_address}) - self.transport.write(b"\r\n\r\n") + self.transport.write(PING) def data_received(self, data: bytes) -> None: """Buffer incoming bytes and dispatch each complete frame.""" @@ -132,7 +138,7 @@ def data_received(self, data: bytes) -> None: for frame in self.extract_frames(): self.dispatch_frame(frame) - def extract_frames(self) -> list[bytes]: + def extract_frames(self) -> typing.Generator[memoryview | bytes, None, None]: """Extract complete SIP messages and keepalive frames from [`recv_buffer`][voip.sip.protocol.SessionInitiationProtocol.recv_buffer]. TCP is a stream protocol; a single [`data_received`][voip.sip.protocol.SessionInitiationProtocol.data_received] @@ -140,15 +146,26 @@ def extract_frames(self) -> list[bytes]: This method uses the `Content-Length` header for body framing per [RFC 3261 §18.3](https://datatracker.ietf.org/doc/html/rfc3261#section-18.3) and recognises [RFC 5626](https://datatracker.ietf.org/doc/html/rfc5626) - PING (`\\r\\n\\r\\n`) and PONG (`\\r\\n`) keepalive sequences. + [`PING`][voip.sip.protocol.PING] and [`PONG`][voip.sip.protocol.PONG] keepalive sequences. Each call mutates [`recv_buffer`][voip.sip.protocol.SessionInitiationProtocol.recv_buffer] in-place, consuming only the bytes that belong to complete frames. - Returns: - List of complete frame byte sequences ready for parsing or keepalive dispatch. + For SIP messages a [`memoryview`][] into the buffer is yielded — zero copy until + [`dispatch_frame`][voip.sip.protocol.SessionInitiationProtocol.dispatch_frame] converts + to [`bytes`][] for parsing. The view is explicitly released and the consumed bytes + removed from the buffer before the next frame is extracted. + + !!! warning + Each yielded [`memoryview`][] is only valid until the generator is advanced to + the next frame. Callers **must not** hold a reference to the view after the + current loop iteration (i.e. after [`dispatch_frame`][voip.sip.protocol.SessionInitiationProtocol.dispatch_frame] + returns). Use `bytes(frame)` to materialise the data if a longer-lived copy is needed. + + Yields: + A [`memoryview`][] for each complete SIP message, or the [`PING`][voip.sip.protocol.PING] / + [`PONG`][voip.sip.protocol.PONG] constant for keepalive frames. """ - frames = [] while self.recv_buffer: if self.recv_buffer[0:1] != b"\r": # SIP message: wait for the header-body separator. @@ -167,56 +184,56 @@ def extract_frames(self) -> list[bytes]: message_end = header_end + 4 + content_length if len(self.recv_buffer) < message_end: break # incomplete body – wait for more data - frames.append(bytes(self.recv_buffer[:message_end])) + frame = memoryview(self.recv_buffer)[:message_end] + yield frame + frame.release() del self.recv_buffer[:message_end] - elif len(self.recv_buffer) >= 4 and self.recv_buffer[:4] == b"\r\n\r\n": - # RFC 5626 PING - frames.append(b"\r\n\r\n") + elif len(self.recv_buffer) >= 4 and self.recv_buffer[:4] == PING: + yield PING del self.recv_buffer[:4] elif len(self.recv_buffer) >= 3 and self.recv_buffer[2:3] == b"\r": - # Third byte is CR – could be the start of \r\n\r\n; wait for 4th byte. + # Third byte is CR – could be the start of PING; wait for 4th byte. break - elif self.recv_buffer[:2] == b"\r\n": - # RFC 5626 PONG (2-byte sequence confirmed not to be a PING prefix). - frames.append(b"\r\n") + elif self.recv_buffer[:2] == PONG: + yield PONG del self.recv_buffer[:2] else: # Single CR or other incomplete sequence – wait for more data. break - return frames - def dispatch_frame(self, frame: bytes) -> None: + def dispatch_frame(self, frame: memoryview | bytes) -> None: """Dispatch a single complete frame (SIP message or keepalive). Args: - frame: A complete frame as returned by + frame: A [`memoryview`][] (SIP message) or keepalive constant + ([`PING`][voip.sip.protocol.PING] / [`PONG`][voip.sip.protocol.PONG]) + as yielded by [`extract_frames`][voip.sip.protocol.SessionInitiationProtocol.extract_frames]. """ peer = NetworkAddress(*self.transport.get_extra_info("peername")) - match frame: - case b"\r\n": - logger.info("PONG", extra={"addr": peer}) - case b"\r\n\r\n": - logger.info("PING", extra={"addr": peer}) - if self.transport: - logger.info("PONG", extra={"addr": self.local_address}) - self.transport.write(b"\r\n") - case _: - match Message.parse(frame): - case Request() as request: - logger.info( - "Request received: %r", - request, - extra={"addr": peer}, - ) - self.request_received(request) - case Response() as response: - logger.info( - "Response received %r", - response, - extra={"addr": peer}, - ) - self.response_received(response) + if frame == PONG: + logger.info("PONG", extra={"addr": peer}) + elif frame == PING: + logger.info("PING", extra={"addr": peer}) + if self.transport: + logger.info("PONG", extra={"addr": self.local_address}) + self.transport.write(PONG) + else: + match Message.parse(bytes(frame)): + case Request() as request: + logger.info( + "Request received: %r", + request, + extra={"addr": peer}, + ) + self.request_received(request) + case Response() as response: + logger.info( + "Response received %r", + response, + extra={"addr": peer}, + ) + self.response_received(response) def send(self, message: Response | Request) -> None: """Serialize and send a SIP message over the TLS/TCP connection.""" From 609055b6db1e64e10134db906acf7c558d5d2e70 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Thu, 26 Mar 2026 15:52:24 +0100 Subject: [PATCH 4/4] Drop the extras --- tests/sip/test_protocol.py | 38 ++++++++++++++++--------------- voip/sip/protocol.py | 46 +++++--------------------------------- 2 files changed, 25 insertions(+), 59 deletions(-) diff --git a/tests/sip/test_protocol.py b/tests/sip/test_protocol.py index 33892a4..fb1c8a2 100644 --- a/tests/sip/test_protocol.py +++ b/tests/sip/test_protocol.py @@ -365,13 +365,13 @@ def _make_session(self, rtp, fake_transport=None): def test_extract_frames__empty_buffer(self, rtp): """Return an empty iterator when the receive buffer is empty.""" session = self._make_session(rtp) - assert [bytes(f) for f in session.extract_frames()] == [] + assert [bytes(f) for f in session._extract_frames()] == [] def test_extract_frames__complete_message(self, rtp): """Extract a single complete SIP message and clear the buffer.""" session = self._make_session(rtp) session.recv_buffer.extend(INVITE_BYTES) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert frames == [INVITE_BYTES] assert len(session.recv_buffer) == 0 @@ -379,7 +379,7 @@ def test_extract_frames__partial_headers(self, rtp): """Keep partial message bytes in the buffer until complete.""" session = self._make_session(rtp) session.recv_buffer.extend(INVITE_BYTES[:20]) - assert [bytes(f) for f in session.extract_frames()] == [] + assert [bytes(f) for f in session._extract_frames()] == [] assert len(session.recv_buffer) == 20 def test_extract_frames__two_coalesced_messages(self, rtp): @@ -395,7 +395,7 @@ def test_extract_frames__two_coalesced_messages(self, rtp): ) session = self._make_session(rtp) session.recv_buffer.extend(INVITE_BYTES + second) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert len(frames) == 2 assert frames[0] == INVITE_BYTES assert frames[1] == second @@ -418,7 +418,7 @@ def test_extract_frames__message_with_body(self, rtp): message = headers + body session = self._make_session(rtp) session.recv_buffer.extend(message) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert frames == [message] assert len(session.recv_buffer) == 0 @@ -437,14 +437,14 @@ def test_extract_frames__incomplete_body(self, rtp): ) session = self._make_session(rtp) session.recv_buffer.extend(headers + body[:5]) - assert [bytes(f) for f in session.extract_frames()] == [] + assert [bytes(f) for f in session._extract_frames()] == [] assert len(session.recv_buffer) == len(headers) + 5 def test_extract_frames__ping(self, rtp): """Extract an RFC 5626 PING (CRLF CRLF) keepalive frame.""" session = self._make_session(rtp) session.recv_buffer.extend(PING) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert frames == [PING] assert len(session.recv_buffer) == 0 @@ -452,7 +452,7 @@ def test_extract_frames__pong(self, rtp): """Extract an RFC 5626 PONG (CRLF) keepalive frame.""" session = self._make_session(rtp) session.recv_buffer.extend(PONG) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert frames == [PONG] assert len(session.recv_buffer) == 0 @@ -460,14 +460,14 @@ def test_extract_frames__partial_keepalive_wait(self, rtp): """Buffer a partial PING (CRLF CR) without dispatching until the 4th byte arrives.""" session = self._make_session(rtp) session.recv_buffer.extend(b"\r\n\r") - assert [bytes(f) for f in session.extract_frames()] == [] + assert [bytes(f) for f in session._extract_frames()] == [] assert len(session.recv_buffer) == 3 def test_extract_frames__ping_followed_by_message(self, rtp): """Extract a PING and a SIP message from the same buffer.""" session = self._make_session(rtp) session.recv_buffer.extend(PING + INVITE_BYTES) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert len(frames) == 2 assert frames[0] == PING assert frames[1] == INVITE_BYTES @@ -476,7 +476,7 @@ def test_extract_frames__pong_followed_by_message(self, rtp): """Extract a PONG and a SIP message coalesced in the same buffer.""" session = self._make_session(rtp) session.recv_buffer.extend(PONG + INVITE_BYTES) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert len(frames) == 2 assert frames[0] == PONG assert frames[1] == INVITE_BYTES @@ -495,14 +495,14 @@ def test_extract_frames__invalid_content_length(self, rtp): ) session = self._make_session(rtp) session.recv_buffer.extend(message) - frames = [bytes(f) for f in session.extract_frames()] + frames = [bytes(f) for f in session._extract_frames()] assert len(frames) == 1 def test_extract_frames__single_cr_waits(self, rtp): """Keep a lone CR in the buffer without dispatching (incomplete keepalive).""" session = self._make_session(rtp) session.recv_buffer.extend(b"\r") - assert [bytes(f) for f in session.extract_frames()] == [] + assert [bytes(f) for f in session._extract_frames()] == [] assert session.recv_buffer == bytearray(b"\r") # ------------------------------------------------------------------ @@ -515,21 +515,21 @@ def test_dispatch_frame__pong(self, rtp, fake_transport, caplog): session = self._make_session(rtp, fake_transport) with caplog.at_level(logging.INFO): - session.dispatch_frame(b"\r\n") + session._dispatch_frame(b"\r\n") assert "PONG" in caplog.text assert fake_transport.sent == [] def test_dispatch_frame__ping__sends_pong(self, rtp, fake_transport): """Reply with a CRLF PONG when a PING (CRLF CRLF) frame is dispatched.""" session = self._make_session(rtp, fake_transport) - session.dispatch_frame(b"\r\n\r\n") + session._dispatch_frame(b"\r\n\r\n") assert b"\r\n" in fake_transport.sent def test_dispatch_frame__sip_request(self, rtp, fake_transport): """Dispatch a SIP request frame to request_received.""" session = self._make_session(rtp, fake_transport) before = len(session.transactions) - session.dispatch_frame(INVITE_BYTES) + session._dispatch_frame(INVITE_BYTES) assert len(session.transactions) > before def test_dispatch_frame__sip_response(self, rtp, fake_transport): @@ -551,7 +551,7 @@ def test_dispatch_frame__sip_response(self, rtp, fake_transport): f"CSeq: 1 INVITE\r\n" f"\r\n" ).encode() - session.dispatch_frame(response_bytes) + session._dispatch_frame(response_bytes) # ------------------------------------------------------------------ # data_received – stream reassembly @@ -671,7 +671,9 @@ def test_allow_header__is_comma_separated_without_sip_fixture(self, rtp): assert "OPTIONS" in header assert "," in header - def test_method_not_allowed__sends_405_without_sip_fixture(self, rtp, fake_transport): + def test_method_not_allowed__sends_405_without_sip_fixture( + self, rtp, fake_transport + ): """method_not_allowed() sends a 405 Method Not Allowed response.""" session = self._make_session(rtp, fake_transport) request = Message.parse( diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index b4bb595..15320da 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -133,39 +133,11 @@ async def send_keepalive(self) -> None: self.transport.write(PING) def data_received(self, data: bytes) -> None: - """Buffer incoming bytes and dispatch each complete frame.""" self.recv_buffer.extend(data) - for frame in self.extract_frames(): - self.dispatch_frame(frame) - - def extract_frames(self) -> typing.Generator[memoryview | bytes, None, None]: - """Extract complete SIP messages and keepalive frames from [`recv_buffer`][voip.sip.protocol.SessionInitiationProtocol.recv_buffer]. - - TCP is a stream protocol; a single [`data_received`][voip.sip.protocol.SessionInitiationProtocol.data_received] - call may carry a partial message, an exact message, or several coalesced messages. - This method uses the `Content-Length` header for body framing per - [RFC 3261 §18.3](https://datatracker.ietf.org/doc/html/rfc3261#section-18.3) - and recognises [RFC 5626](https://datatracker.ietf.org/doc/html/rfc5626) - [`PING`][voip.sip.protocol.PING] and [`PONG`][voip.sip.protocol.PONG] keepalive sequences. - - Each call mutates [`recv_buffer`][voip.sip.protocol.SessionInitiationProtocol.recv_buffer] - in-place, consuming only the bytes that belong to complete frames. - - For SIP messages a [`memoryview`][] into the buffer is yielded — zero copy until - [`dispatch_frame`][voip.sip.protocol.SessionInitiationProtocol.dispatch_frame] converts - to [`bytes`][] for parsing. The view is explicitly released and the consumed bytes - removed from the buffer before the next frame is extracted. - - !!! warning - Each yielded [`memoryview`][] is only valid until the generator is advanced to - the next frame. Callers **must not** hold a reference to the view after the - current loop iteration (i.e. after [`dispatch_frame`][voip.sip.protocol.SessionInitiationProtocol.dispatch_frame] - returns). Use `bytes(frame)` to materialise the data if a longer-lived copy is needed. - - Yields: - A [`memoryview`][] for each complete SIP message, or the [`PING`][voip.sip.protocol.PING] / - [`PONG`][voip.sip.protocol.PONG] constant for keepalive frames. - """ + for frame in self._extract_frames(): + self._dispatch_frame(frame) + + 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. @@ -201,15 +173,7 @@ def extract_frames(self) -> typing.Generator[memoryview | bytes, None, None]: # Single CR or other incomplete sequence – wait for more data. break - def dispatch_frame(self, frame: memoryview | bytes) -> None: - """Dispatch a single complete frame (SIP message or keepalive). - - Args: - frame: A [`memoryview`][] (SIP message) or keepalive constant - ([`PING`][voip.sip.protocol.PING] / [`PONG`][voip.sip.protocol.PONG]) - as yielded by - [`extract_frames`][voip.sip.protocol.SessionInitiationProtocol.extract_frames]. - """ + def _dispatch_frame(self, frame: memoryview | bytes) -> None: peer = NetworkAddress(*self.transport.get_extra_info("peername")) if frame == PONG: logger.info("PONG", extra={"addr": peer})