feat(rtsp): always apply ZTE NAT traversal

e2e/helpers/__init__.py

@@ -46,6 +46,7 @@
     MockRTSPServerNoTeardownResponse,
     MockRTSPServerSilent,
     MockRTSPServerUDP,
+    MockRTSPServerZTE,
 )
 from .mock_stun import MockSTUNServer
 from .ports import (
@@ -73,6 +74,7 @@
     "MockRTSPServerNoTeardownResponse",
     "MockRTSPServerSilent",
     "MockRTSPServerUDP",
+    "MockRTSPServerZTE",
     "MockSTUNServer",
     "MulticastSender",
     "R2HProcess",

e2e/helpers/mock_rtsp.py

@@ -78,6 +78,7 @@ def __init__(
         self._stop = threading.Event()
         self.requests_received: list[str] = []
         self.requests_detailed: list[dict] = []
+        self.control_peer: tuple | None = None
 
     # -- lifecycle -----------------------------------------------------------
 
@@ -126,16 +127,22 @@ def _accept(self) -> None:
 
     def _handle(self, conn: socket.socket, addr: tuple) -> None:
         conn.settimeout(10.0)
+        self.control_peer = addr
         transport_hdr = ""
         try:
+            pending = b""
             while True:
-                data = b""
-                while b"\r\n\r\n" not in data:
+                while b"\r\n\r\n" not in pending:
                     chunk = conn.recv(4096)
                     if not chunk:
                         return
-                    data += chunk
-                req = data.decode(errors="replace")
+                    pending += chunk
+                # Split off exactly one request; anything left is a later
+                # request that arrived in the same segment.  Keeping it buffered
+                # (rather than folding it into this one) is what makes an
+                # unexpectedly pipelined request visible to tests.
+                data, pending = pending.split(b"\r\n\r\n", 1)
+                req = data.decode(errors="replace") + "\r\n\r\n"
                 first_line = req.split("\r\n")[0].split()
                 method = first_line[0]
                 uri = first_line[1] if len(first_line) > 1 else ""
@@ -401,6 +408,148 @@ def _after_play(self, conn: socket.socket, addr: tuple) -> None:
             udp_sock.close()
 
 
+# ---------------------------------------------------------------------------
+# MockRTSPServerZTE  --  ZTE UDP NAT traversal mode
+# ---------------------------------------------------------------------------
+
+
+class MockRTSPServerZTE(_RTSPServerBase):
+    """RTSP server that starts UDP media only after a valid ZTE punch packet.
+
+    ``expected_ip`` / ``expected_control_port`` override what the punch packet is
+    validated against; leave them unset to expect the RTSP control connection's
+    own endpoint.  Set them when rtp2httpd advertises a STUN-discovered mapping
+    instead, in which case the UDP source port no longer matches the advertised
+    RTP port and ``check_source_port`` must be disabled.
+    """
+
+    def __init__(
+        self,
+        port: int = 0,
+        num_packets: int = 200,
+        expected_ip: str | None = None,
+        expected_control_port: int | None = None,
+        check_source_port: bool = True,
+    ):
+        super().__init__(port)
+        self._num_packets = num_packets
+        self._expected_ip = expected_ip
+        self._expected_control_port = expected_control_port
+        self._check_source_port = check_source_port
+        self._server_rtp_socket: socket.socket | None = None
+        self._server_rtcp_socket: socket.socket | None = None
+        self._receiver_thread: threading.Thread | None = None
+        self._play_started = threading.Event()
+        self._valid_probe = threading.Event()
+        self._client_rtp_port = 0
+        self._client_address = ""
+        self._server_rtp_port = 0
+        self._server_rtcp_port = 0
+        self.udp_datagrams: list[tuple[bytes, tuple]] = []
+
+    @property
+    def valid_probe_received(self) -> bool:
+        return self._valid_probe.is_set()
+
+    def stop(self) -> None:
+        self._play_started.set()
+        super().stop()
+        if self._server_rtp_socket:
+            self._server_rtp_socket.close()
+        if self._server_rtcp_socket:
+            self._server_rtcp_socket.close()
+        if self._receiver_thread:
+            self._receiver_thread.join(timeout=2)
+
+    def _setup_response(self, cseq: str, transport_hdr: str) -> str:
+        for part in transport_hdr.split(";"):
+            part = part.strip()
+            if part.startswith("client_port="):
+                self._client_rtp_port = int(part.split("=", 1)[1].split("-", 1)[0])
+            elif part.startswith("client_address="):
+                self._client_address = part.split("=", 1)[1]
+
+        while True:
+            rtp_port, rtcp_port = find_free_udp_port_pair()
+            rtp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            rtcp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            try:
+                rtp_socket.bind((self.host, rtp_port))
+                rtcp_socket.bind((self.host, rtcp_port))
+                break
+            except OSError:
+                rtp_socket.close()
+                rtcp_socket.close()
+
+        self._server_rtp_socket = rtp_socket
+        self._server_rtcp_socket = rtcp_socket
+        self._server_rtp_port = rtp_port
+        self._server_rtcp_port = rtcp_port
+        self._receiver_thread = threading.Thread(target=self._receive_probes, daemon=True)
+        self._receiver_thread.start()
+
+        return (
+            "RTSP/1.0 200 OK\r\nCSeq: %s\r\n"
+            "Transport: MP2T/RTP/UDP;unicast;client_port=%d-%d;server_port=%d-%d\r\n"
+            "Session: t1\r\n\r\n"
+            % (cseq, self._client_rtp_port, self._client_rtp_port + 1, self._server_rtp_port, self._server_rtcp_port)
+        )
+
+    def _receive_probes(self) -> None:
+        assert self._server_rtp_socket is not None
+        self._server_rtp_socket.settimeout(0.05)
+        while not self._stop.is_set():
+            try:
+                payload, source = self._server_rtp_socket.recvfrom(2048)
+                self.udp_datagrams.append((payload, source))
+                if self._probe_is_valid(payload, source):
+                    self._valid_probe.set()
+            except socket.timeout:
+                if self._play_started.is_set():
+                    return
+            except OSError:
+                return
+
+    def _probe_is_valid(self, payload: bytes, source: tuple) -> bool:
+        if not self.control_peer or len(payload) != 84:
+            return False
+        expected_ip = socket.inet_aton(self._expected_ip or self.control_peer[0])
+        expected_tcp_port = self._expected_control_port or self.control_peer[1]
+        if self._check_source_port and source[1] != self._client_rtp_port:
+            return False
+        return (
+            payload[:8] == b"ZXV10STB"
+            and payload[8:12] == b"\x7f\xff\xff\xff"
+            and payload[12:16] == expected_ip
+            and struct.unpack("!H", payload[16:18])[0] == self._client_rtp_port
+            and struct.unpack("!H", payload[18:20])[0] == expected_tcp_port
+            and payload[20:] == bytes(64)
+            and source[0] == self.control_peer[0]
+        )
+
+    def _after_play(self, conn: socket.socket, addr: tuple) -> None:
+        self._play_started.set()
+        if not self._valid_probe.wait(timeout=2.0) or not self.udp_datagrams:
+            return
+        if self._receiver_thread:
+            self._receiver_thread.join(timeout=0.2)
+
+        assert self._server_rtp_socket is not None
+        destination = next(source for payload, source in self.udp_datagrams if self._probe_is_valid(payload, source))
+        seq = 0
+        ts = 0
+        try:
+            for _ in range(self._num_packets):
+                if self._stop.is_set():
+                    break
+                self._server_rtp_socket.sendto(make_rtp_packet(seq, ts), destination)
+                seq = (seq + 1) & 0xFFFF
+                ts = (ts + 3600) & 0xFFFFFFFF
+                time.sleep(0.001)
+        except OSError:
+            pass
+
+
 # ---------------------------------------------------------------------------
 # MockRTSPServerSilent  --  accepts connection but never responds
 # ---------------------------------------------------------------------------