fix(stream): drop stray datagrams instead of splicing them into the TS output

e2e/helpers/mock_rtsp.py

@@ -421,6 +421,11 @@ class MockRTSPServerZTE(_RTSPServerBase):
     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.
+
+    ``echo_probe_after`` makes the server bounce the 84-byte punch packet back
+    onto the media port after that many RTP packets, the way ZTE servers
+    acknowledge a punch mid-stream.  It reproduces the stray non-RTP datagram
+    that must never reach the client's MPEG-TS output.
     """
 
     def __init__(
@@ -430,12 +435,14 @@ def __init__(
         expected_ip: str | None = None,
         expected_control_port: int | None = None,
         check_source_port: bool = True,
+        echo_probe_after: int | None = None,
     ):
         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._echo_probe_after = echo_probe_after
         self._server_rtp_socket: socket.socket | None = None
         self._server_rtcp_socket: socket.socket | None = None
         self._receiver_thread: threading.Thread | None = None
@@ -535,13 +542,17 @@ def _after_play(self, conn: socket.socket, addr: tuple) -> None:
             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))
+        probe, destination = next(
+            (payload, 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):
+            for index in range(self._num_packets):
                 if self._stop.is_set():
                     break
+                if index == self._echo_probe_after:
+                    self._server_rtp_socket.sendto(probe, destination)
                 self._server_rtp_socket.sendto(make_rtp_packet(seq, ts), destination)
                 seq = (seq + 1) & 0xFFFF
                 ts = (ts + 3600) & 0xFFFFFFFF

e2e/test_rtsp_zte_nat.py

@@ -173,6 +173,43 @@ def test_ipv6_upstream_falls_back_to_ordinary_rtsp(self, r2h_binary):
             r2h.stop()
             rtsp.stop()
 
+    def test_probe_echo_never_reaches_the_client(self, r2h_binary):
+        """A punch ack bounced onto the media port must be dropped, not relayed.
+
+        The media socket is unconnected, so any stray datagram lands in the same
+        recv() as the media.  Splicing an 84-byte packet into the body shifts
+        every following TS packet off the 188-byte grid -- ffmpeg-based players
+        resync on the next sync byte, but strict demuxers stall for good.
+        """
+        echo_after = 20
+        rtsp = MockRTSPServerZTE(num_packets=500, echo_probe_after=echo_after)
+        rtsp.start()
+        r2h_port = find_free_port()
+        r2h = R2HProcess(r2h_binary, r2h_port, extra_args=["-v", "4"], capture_log=True)
+        r2h.start()
+        try:
+            status, _, body = stream_get(
+                "127.0.0.1",
+                r2h_port,
+                "/rtsp/127.0.0.1:%d/stream" % rtsp.port,
+                read_bytes=188 * (echo_after * 4),
+                timeout=20.0,
+            )
+            assert status == 200
+            assert rtsp.valid_probe_received
+            # Read well past the echo so a shifted grid cannot hide in the tail.
+            assert len(body) >= 188 * (echo_after * 2)
+
+            assert b"ZXV10STB" not in body
+            aligned_len = len(body) - len(body) % 188
+            misaligned = [offset for offset in range(0, aligned_len, 188) if body[offset] != 0x47]
+            assert not misaligned, "TS alignment lost at byte offset(s) %s" % misaligned[:5]
+
+            assert "Dropped 84-byte datagram that is neither RTP nor MPEG-TS" in r2h.read_log()
+        finally:
+            r2h.stop()
+            rtsp.stop()
+
     def test_redirect_recaptures_control_endpoint(self, r2h_binary):
         target = MockRTSPServerZTE(num_packets=300)
         target.start()

src/stream.c

@@ -303,7 +303,20 @@ int stream_process_rtp_payload(stream_context_t *ctx, buffer_ref_t *buf_ref, str
   stream_metadata_note_media(ctx, pkt_type, payload, payload_len, origin);
 
   if (pkt_type == 0) {
-    /* Non-RTP packet - pass through directly (no reordering needed) */
+    /* Non-RTP packet.  The only legitimate case is an upstream that sends bare
+     * MPEG-TS over UDP (RTSP servers negotiating plain MP2T, raw TS multicast
+     * streams), so anything that is not TS is a stray datagram: a ZTE
+     * ZXV10STB NAT punch reply landing on the media port, an RTCP report sent
+     * to the wrong port, or an unrelated sender - the media sockets are
+     * unconnected and accept from anyone.  Splicing such a datagram into the
+     * output breaks the 188-byte TS alignment for the rest of the stream,
+     * which strict demuxers (mpegts.js) never recover from. */
+    if (!stream_payload_is_mpegts(payload, payload_len)) {
+      logger(LOG_DEBUG, "Stream: Dropped %d-byte datagram that is neither RTP nor MPEG-TS", payload_len);
+      return 0;
+    }
+
+    /* Bare MPEG-TS - pass through directly (no reordering needed) */
     if (ctx->snapshot.initialized) {
       return snapshot_process_packet(&ctx->snapshot, buf_ref->data_size, data_ptr, ctx->conn);
     }