diff --git a/Makefile b/Makefile index 31a793a..32b832e 100644 --- a/Makefile +++ b/Makefile @@ -12,7 +12,7 @@ CXXFLAGS := $(CXXFLAGS) -DMAVLINK_SIGNING_TIMESTAMP_LIMIT=600 LIBS := -ltdb -lssl -lcrypto # Source files -SOURCES := supportproxy.cpp mavlink.cpp util.cpp keydb.cpp conntdb.cpp tlog.cpp session.cpp binlog.cpp cleanup.cpp websocket.cpp video.cpp videoauth.cpp videots.cpp videostream.cpp videorec.cpp videoview.cpp httpreq.cpp videortsp.cpp videortmp.cpp +SOURCES := supportproxy.cpp mavlink.cpp util.cpp keydb.cpp conntdb.cpp tlog.cpp session.cpp binlog.cpp cleanup.cpp websocket.cpp video.cpp videoauth.cpp videots.cpp videostream.cpp videorec.cpp videoview.cpp httpreq.cpp videortsp.cpp videortmp.cpp videortspview.cpp OBJECTS := $(SOURCES:.cpp=.o) TARGET := supportproxy @@ -107,6 +107,8 @@ videorec.o: videorec.cpp videorec.h session.h cleanup.h videoview.o: videoview.cpp videoview.h httpreq.h videostream.h videots.h videoauth.h keydb.h httpreq.o: httpreq.cpp httpreq.h videortsp.o: videortsp.cpp videortsp.h +video.o videoview.o videortspview.o: videortspview.h +videortspview.o: videortspview.cpp videoview.h videostream.h videomkv.h videots.h videoauth.h httpreq.h keydb.h videortmp.o: videortmp.cpp videortmp.h httpreq.h # Testing diff --git a/README.md b/README.md index 8f77a39..df305d2 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,39 @@ supplied and is wrong is still refused. Chrome and Firefox will not decode HEVC in Media Source Extensions on desktop Linux, and nothing here transcodes. +RTSP playback is also available on each existing video port: + +```bash +ffplay -rtsp_transport tcp 'rtsp://HOST:PORT/view/v1' +vlc --rtsp-tcp 'rtsp://HOST:PORT/view/v1' +``` + +For Mission Planner, right-click the HUD, select **Video → Set GStreamer +Source**, and paste this pipeline as one line: + +```text +rtspsrc location="rtsp://HOST:PORT/view/v1" protocols=tcp latency=100 ! application/x-rtp ! decodebin3 ! queue max-size-buffers=1 leaky=2 ! videoconvert ! video/x-raw,format=BGRA ! appsink name=outsink sync=false +``` + +The video page offers this pipeline with the entry's host, port and slot +already filled in. It follows Mission Planner's [live video setup](https://ardupilot.org/planner/docs/live-video.html) +with TCP transport selected for the proxy. + +Use `/view/v2` on the second slot's port, and so on. If a viewer password +is set, append `?pw=YOUR_VIEWER_PASSWORD` (URL-encode special characters), +or use the player's HTTP Basic authentication support. Existing viewer +tokens (`?t=...`) also work. These are viewer credentials, independent of +the publish password or MAVLink admission. + +Playback carries MPEG-TS in RTP (MP2T) over the RTSP TCP connection, so +it needs no extra firewall ports or transcoding and preserves the video +codec and any audio already present in the stream. The player must +support MP2T over RTSP and TCP transport; UDP playback is not offered. +Matroska/raw thermal streams use their existing HTTP viewer instead. +A viewer starts at a buffered keyframe, and reconnects after the +publisher restarts. The `/view/` RTSP namespace is reserved for viewing; +RTSP publishers should continue using a different path, such as `/cam`. + **Disk.** Video has its own budget, separate from telemetry, so a busy camera can never evict a user's tlogs. Set it per entry in the web UI; a free-space floor stops recording before the disk fills. diff --git a/tests/test_video_rtsp_view.py b/tests/test_video_rtsp_view.py new file mode 100644 index 0000000..471c742 --- /dev/null +++ b/tests/test_video_rtsp_view.py @@ -0,0 +1,532 @@ +"""RTSP viewers share the TS ring without acquiring a publisher slot.""" +import base64 +import os +import shutil +import socket +import struct +import subprocess +import time + +import pytest +import tsgen + +from test_video_child import VPORT, PORT_ENG +from test_video_view import session, http_get, read_for # noqa: F401 +from test_video_rtsp import clip, RtspSession, _workdir # noqa: F401 + + +class Client: + def __init__(self, suffix='', receive_buffer=None): + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.settimeout(5) + if receive_buffer is not None: + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, receive_buffer) + self.sock.connect(('127.0.0.1', VPORT)) + self.url = 'rtsp://127.0.0.1:%d/view/v1%s' % (VPORT, suffix) + self.cseq = 0 + self.pending = b'' + self.session = None + self.media = [] + + def close(self): + self.sock.close() + + def more(self): + data = self.sock.recv(65536) + if not data: + raise EOFError() + self.pending += data + + def frame(self): + while len(self.pending) < 4: + self.more() + channel, size = struct.unpack('!BH', self.pending[1:4]) + while len(self.pending) < size + 4: + self.more() + data, self.pending = self.pending[4:size+4], self.pending[size+4:] + return channel, data + + def request(self, method, extra='', url=None, fragmented=False): + self.cseq += 1 + if self.session: + extra += 'Session: %s\r\n' % self.session + data = ('%s %s RTSP/1.0\r\nCSeq: %d\r\n%s\r\n' % + (method, url or self.url, self.cseq, extra)).encode() + if fragmented: + # Split inside the request target, before classification. + split = data.index(b'/view/') + 8 + self.sock.sendall(data[:split]) + time.sleep(0.1) + self.sock.sendall(data[split:]) + else: + self.sock.sendall(data) + return self.response() + + def response(self): + while True: + if not self.pending: + self.more() + if self.pending[0] == ord('$'): + self.media.append(self.frame()) + continue + end = self.pending.find(b'\r\n\r\n') + if end < 0: + self.more() + continue + lines = self.pending[:end].decode().split('\r\n') + headers = dict(line.lower().split(': ', 1) for line in lines[1:]) + size = int(headers.get('content-length', 0)) + while len(self.pending) < end + 4 + size: + self.more() + body = self.pending[end+4:end+4+size] + self.pending = self.pending[end+4+size:] + return int(lines[0].split()[1]), headers, body + + def play(self, channel=0): + assert self.request('OPTIONS')[0] == 200 + code, _, sdp = self.request('DESCRIBE') + assert code == 200 + assert b'MP2T/90000' in sdp + code, headers, _ = self.request('SETUP', + 'Transport: RTP/AVP/TCP;unicast;interleaved=%d-%d\r\n' % + (channel, channel+1), url=self.url.split('?')[0]+'/trackID=0') + assert code == 200 + self.session = headers['session'].split(';')[0] + assert self.request('PLAY')[0] == 200 + + +class TaggedTSGen(tsgen.TSGen): + """Distinguish replayed packets even when continuity counters wrap.""" + def __init__(self): + super().__init__() + self.serial = 0 + + def video(self, **kwargs): + packet = bytearray(super().video(**kwargs)) + struct.pack_into('!Q', packet, len(packet)-8, self.serial) + self.serial += 1 + return bytes(packet) + + +def rtp_frame(client): + while True: + channel, packet = client.frame() + if channel == 0: + return packet + assert channel == 1 + + +def pause_after_buffered_data(client, final_packet): + # Establish the exact pause cursor instead of relying on one pump + # iteration having sent the complete initial GOP. + while True: + packet = rtp_frame(client) + client.media.append((0, packet)) + if packet.endswith(final_packet): + break + assert client.request('PAUSE')[0] == 200 + return [packet for channel, packet in client.media if channel == 0] + + +@pytest.mark.integration +def test_pause_resume_preserves_buffered_position(session): + s = session() + s.feed.gen = TaggedTSGen() + assert s.wait_ready(), s.proxy.log + c = Client() + try: + c.play() + before = pause_after_buffered_data(c, s.feed.snapshot()[-188:]) + assert before + + # No keyframe arrives while paused: reusing the join anchor + # would replay the end of the previous GOP in this session. + new_data = b''.join(s.feed.gen.video() for _ in range(21)) + for datagram in s.feed.gen.datagrams(new_data): + s.feed.pub.sock.sendto(datagram, ('127.0.0.1', VPORT)) + time.sleep(0.002) + assert c.request('PLAY')[0] == 200 + after, data = [], b'' + while len(data) < len(new_data): + packet = rtp_frame(c) + after.append(packet) + data += packet[12:] + assert data == new_data, 'resume replayed or skipped buffered TS packets' + assert after[0][8:12] == before[-1][8:12] # same SSRC + first = struct.unpack('!H', after[0][2:4])[0] + previous = struct.unpack('!H', before[-1][2:4])[0] + assert first == (previous+1) % 65536 + finally: + c.close() + + +@pytest.mark.integration +def test_pause_resume_rejoins_at_keyframe_after_lapping(session): + s = session(env={'SUPPORTPROXY_VIDEO_RING_BYTES': str(64 * 1024)}) + s.feed.gen = TaggedTSGen() + assert s.wait_ready(), s.proxy.log + c = Client() + try: + c.play() + before = pause_after_buffered_data(c, s.feed.snapshot()[-188:]) + assert before + s.feed.burst(2000) # overwrite the paused cursor + assert c.request('PLAY')[0] == 200 + packet = rtp_frame(c) + data = packet[12:] + assert data[0] == 0x47 and ((data[1] & 31) << 8) | data[2] == 0 + assert data in s.feed.snapshot() + assert data not in b''.join(p[12:] for p in before) + assert packet[8:12] == before[-1][8:12] + finally: + c.close() + + +@pytest.mark.integration +@pytest.mark.parametrize('half_close', [False, True]) +def test_blocked_teardown_does_not_spin_on_pending_input(session, half_close): + s = session() + assert s.wait_ready(), s.proxy.log + c = Client(receive_buffer=4096) + try: + c.play() + s.feed.burst(6000) # fill the TCP send buffer while the viewer stops reading + peer_port = c.sock.getsockname()[1] + with open('/proc/net/tcp') as f: + connections = [line.split() for line in f.readlines()[1:]] + queues = [int(row[4].split(':')[0], 16) for row in connections + if int(row[1].split(':')[1], 16) == VPORT and + int(row[2].split(':')[1], 16) == peer_port] + assert queues and queues[0] > 0, 'test did not backpressure the viewer' + + c.cseq += 1 + request = ('TEARDOWN %s RTSP/1.0\r\nCSeq: %d\r\nSession: %s\r\n\r\n' % + (c.url, c.cseq, c.session)).encode() + # More than a read chunk remains in the socket after TEARDOWN. + c.sock.sendall(request + b'x' * 32768) + if half_close: + c.sock.shutdown(socket.SHUT_WR) + assert s.proxy.wait_for('viewer disconnected', timeout=3), s.proxy.log + else: + vpid = s.proxy.video_pid() + assert vpid is not None + + def ticks(): + with open('/proc/%d/stat' % vpid) as f: + fields = f.read().split() + return int(fields[13]) + int(fields[14]) + + start = ticks() + time.sleep(2) + assert ticks()-start < os.sysconf('SC_CLK_TCK') // 2, \ + 'closing RTSP viewer busy-spun on unread input' + finally: + c.close() + + +@pytest.mark.integration +def test_rtsp_payload_is_verbatim_and_http_stays_live(session): + s = session() + assert s.wait_ready(), s.proxy.log + s.feed.start() + c = Client() + try: + c.play(channel=4) + payloads, seqs, reports = [], [], [] + for _ in range(35): + channel, packet = c.frame() + if channel == 5: + reports.append(packet) + continue + assert channel == 4 and packet[:2] == b'\x80\x21' + seqs.append(struct.unpack('!H', packet[2:4])[0]) + payloads.append(packet[12:]) + data = b''.join(payloads) + assert len(data) > 10000 + assert data in s.feed.snapshot() + assert all(b == (a+1) % 65536 for a, b in zip(seqs, seqs[1:])) + assert reports and reports[0][1] == 200 + assert data[0] == 0x47 + assert ((data[1] & 31) << 8) | data[2] == 0 # join starts at PAT + http, headers, body = http_get(VPORT) + try: + assert b'200 OK' in headers + assert len(read_for(http, 0.5, body)) > 1000 + finally: + http.close() + # Receiver reports share the TCP connection with RTSP controls. + c.sock.sendall(b'$\x05\x00\x08\x80\xc9\x00\x01\x00\x00\x00\x01') + assert c.request('GET_PARAMETER')[0] == 200 + assert c.request('PAUSE')[0] == 200 + assert c.request('PLAY')[0] == 200 + assert c.request('TEARDOWN')[0] == 200 + finally: + c.close() + + +@pytest.mark.integration +@pytest.mark.parametrize('credential', ['query', 'basic', 'token']) +def test_viewer_credentials_and_fragmented_request(session, credential): + s = session(viewer_pass='viewer-secret') + assert s.wait_ready(), s.proxy.log + c = Client() + try: + assert c.request('DESCRIBE', fragmented=True)[0] == 401 + assert c.request('DESCRIBE', url=c.url+'?pw=wrong')[0] == 401 + if credential == 'query': + extra, url = '', c.url+'?pw=viewer-secret' + elif credential == 'basic': + extra = 'Authorization: Basic %s\r\n' % base64.b64encode( + b'viewer:viewer-secret').decode() + url = c.url + else: + import keydb_lib + from webadmin.videotoken import mint + db = keydb_lib.open_db(str(s.workdir / 'keys.tdb')) + ke = keydb_lib.KeyEntry(PORT_ENG) + assert ke.fetch(db) + db.close() + url, extra = c.url+'?t='+mint(ke.secret_key, ke.port2, 0), '' + code, headers, body = c.request('DESCRIBE', extra, url=url) + assert code == 200 and b'MP2T' in body + assert '?' not in headers['content-base'] + # This connection retains authorisation for the relative track URI. + assert c.request('SETUP', 'Transport: RTP/AVP/TCP;unicast;interleaved=0-1\r\n')[0] == 200 + assert 'viewer-secret' not in s.proxy.log + assert 'RTSP publisher' not in s.proxy.log + finally: + c.close() + + +@pytest.mark.integration +def test_not_ready_paths_transport_and_session(session): + s = session() + c = Client() + try: + assert c.request('DESCRIBE')[0] == 503 + assert s.wait_ready(), s.proxy.log + assert c.request('DESCRIBE', url=c.url.replace('v1', 'v2'))[0] == 404 + assert c.request('SETUP', 'Transport: RTP/AVP;unicast;client_port=6000-6001\r\n')[0] == 461 + assert c.request('SETUP', 'Transport: RTP/AVP/TCP;interleaved=255-256\r\n')[0] == 461 + assert c.request('PLAY')[0] == 454 + c.play() + assert c.request('PLAY', 'Scale: 1.000\r\n')[0] == 200 + assert c.request('PLAY', 'Scale: 2.0\r\n')[0] == 457 + assert c.request('PLAY', 'Scale: nan\r\n')[0] == 457 + c.session = 'wrong' + assert c.request('TEARDOWN')[0] == 454 + assert c.request('GET_PARAMETER')[0] == 454 + finally: + c.close() + + +@pytest.mark.integration +def test_selects_tcp_from_transport_alternatives(session): + s = session() + assert s.wait_ready(), s.proxy.log + c = Client() + try: + code, headers, _ = c.request('SETUP', + 'Transport: RTP/AVP;unicast;client_port=6000-6001, ' + 'RTP/AVP/TCP;unicast;interleaved=6-7\r\n') + assert code == 200 + assert headers['transport'] == 'rtp/avp/tcp;unicast;interleaved=6-7' + c.session = headers['session'].split(';')[0] + assert c.request('PLAY')[0] == 200 + while True: + channel, _ = c.frame() + if channel == 6: + break + assert channel == 7 + finally: + c.close() + + +@pytest.mark.integration +@pytest.mark.slow +def test_unauthenticated_options_cannot_extend_admission_deadline(session): + s = session(viewer_pass='viewer-secret') + c = Client() + try: + start = time.monotonic() + disconnected = False + while time.monotonic()-start < 20: + try: + assert c.request('OPTIONS')[0] == 200 + except (EOFError, ConnectionError): + disconnected = True + break + time.sleep(1) + assert disconnected, s.proxy.log + assert s.proxy.wait_for('RTSP authentication timeout'), s.proxy.log + healthy = Client('?pw=viewer-secret') + try: + assert healthy.request('DESCRIBE')[0] == 503 + finally: + healthy.close() + finally: + c.close() + + +@pytest.mark.integration +def test_repeated_authentication_failures_close_connection(session): + s = session(viewer_pass='viewer-secret') + c = Client('?pw=wrong') + try: + for _ in range(3): + assert c.request('DESCRIBE')[0] == 401 + assert c.sock.recv(1024) == b'' + assert 'RTSP publisher' not in s.proxy.log + finally: + c.close() + + +@pytest.mark.integration +@pytest.mark.slow +def test_rtcp_receiver_reports_keep_session_alive(session): + s = session() + c = Client() + try: + # SETUP without PLAY keeps output quiet, isolating the session + # timer from write progress and the independent publisher timer. + code, headers, _ = c.request('SETUP', + 'Transport: RTP/AVP/TCP;interleaved=0-1\r\n') + assert code == 200 + c.session = headers['session'].split(';')[0] + start = time.monotonic() + while time.monotonic()-start < 70: + c.sock.sendall(b'$\x01\x00\x08\x80\xc9\x00\x01\x00\x00\x00\x01') + time.sleep(5) + assert c.request('GET_PARAMETER')[0] == 200 + finally: + c.close() + + +@pytest.mark.integration +@pytest.mark.parametrize('packet', [ + b'', # no RTCP header + b'\x40\xc9\x00\x01\x00\x00\x00\x01', # wrong version + b'\x81\xc9\x00\x01\x00\x00\x00\x01', # RR report block missing + b'\xa0\xc9\x00\x01\x00\x00\x00\x00', # zero padding count +]) +def test_invalid_rtcp_is_rejected(session, packet): + session() + c = Client() + try: + code, _, _ = c.request('SETUP', + 'Transport: RTP/AVP/TCP;interleaved=0-1\r\n') + assert code == 200 + c.sock.sendall(b'$\x01'+struct.pack('!H', len(packet))+packet) + assert c.sock.recv(1024) == b'' + finally: + c.close() + + +@pytest.mark.integration +@pytest.mark.parametrize('wire_request', [ + b'DESCRIBE {url} RTSP/1.0\r\nCSeq: 1\r\nContent-Length: 999999999\r\n\r\n', + b'DESCRIBE {url} RTSP/1.0\r\nCSeq: 1\r\nContent-Length: 0\r\nContent-Length: 1\r\n\r\n', + b'DESCRIBE {url} RTSP/1.0\r\nCSeq: 1\r\nTransfer-Encoding: chunked\r\n\r\n', + b'DESCRIBE {url} RTSP/1.0\r\nCSeq: bad\r\n\r\n', +]) +def test_malformed_control_cannot_disturb_publisher(session, wire_request): + s = session() + assert s.wait_ready(), s.proxy.log + c = Client() + try: + c.sock.sendall(wire_request.replace(b'{url}', c.url.encode())) + assert c.sock.recv(1024) == b'' + healthy = Client() + try: + assert healthy.request('DESCRIBE')[0] == 200 + finally: + healthy.close() + finally: + c.close() + + +@pytest.mark.integration +def test_slow_rtsp_viewer_is_dropped_without_stalling_http(session): + s = session(env={'SUPPORTPROXY_VIDEO_RING_BYTES': str(256 * 1024)}) + assert s.wait_ready(), s.proxy.log + c = Client() + try: + c.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096) + c.play() + s.feed.start() + assert s.proxy.wait_for(r'viewer disconnected .*\(fell too far behind', timeout=25), s.proxy.log + healthy, headers, body = http_get(VPORT) + try: + assert b'200 OK' in headers + data = read_for(healthy, 1, body) + assert len(data) > 10000 and data in s.feed.snapshot() + finally: + healthy.close() + finally: + c.close() + + +@pytest.mark.integration +@pytest.mark.skipif(shutil.which('ffmpeg') is None, reason='real media needs ffmpeg') +@pytest.mark.parametrize('player', ['ffmpeg', 'ffmpeg_basic', 'gstreamer', 'vlc']) +def test_real_rtsp_publisher_can_be_decoded_by_rtsp_viewer(tmp_path, clip, player): + if player == 'gstreamer' and shutil.which('gst-launch-1.0') is None: + pytest.skip('GStreamer is not installed') + if player == 'vlc' and shutil.which('cvlc') is None: + pytest.skip('VLC is not installed') + if player == 'vlc': + modules = subprocess.run(['cvlc', '--list'], capture_output=True, timeout=10) + if b'live555' not in modules.stdout: + pytest.skip('VLC build has no live555 RTSP plugin') + import keydb_lib + wd = _workdir(tmp_path) + db = keydb_lib.open_db(str(wd / 'keys.tdb')) + keydb_lib.set_video_viewer_pass(db, PORT_ENG, 'viewer-secret') + db.close() + s = RtspSession(wd) + try: + s.publish(clip) + assert s.proxy.wait_for(r'join=ready', timeout=40), s.proxy.log + url = 'rtsp://127.0.0.1:%d/view/v1?pw=viewer-secret' % VPORT + if player == 'ffmpeg_basic': + url = 'rtsp://viewer:viewer-secret@127.0.0.1:%d/view/v1' % VPORT + if player.startswith('ffmpeg'): + argv = [ + 'ffmpeg', '-hide_banner', '-loglevel', 'error', + '-rtsp_transport', 'tcp', '-i', url, + '-frames:v', '5', '-f', 'framemd5', '-', + ] + elif player == 'gstreamer': + argv = [ + # Mission Planner's documented pipeline. The identity + # ends it after decoded frames arrive; without an MP + # consumer, appsink must not wait for its queue at EOS. + 'gst-launch-1.0', '-q', 'rtspsrc', 'location='+url, + 'protocols=tcp', 'latency=100', '!', 'application/x-rtp', '!', + 'decodebin3', '!', 'queue', 'max-size-buffers=1', 'leaky=2', '!', + 'videoconvert', '!', 'video/x-raw,format=BGRA', '!', + 'identity', 'eos-after=6', '!', 'appsink', + 'name=outsink', 'wait-on-eos=false', 'sync=false', + ] + else: + argv = [ + 'cvlc', '--ignore-config', '--no-media-library', '-vv', + '--rtsp-tcp', '--vout=dummy', '--aout=dummy', '--no-audio', + '--avcodec-hw=none', '--run-time=2', url, 'vlc://quit', + ] + result = subprocess.run(argv, capture_output=True, timeout=25) + assert result.returncode == 0, result.stderr.decode() + if player.startswith('ffmpeg'): + frames = [line for line in result.stdout.splitlines() if line and not line.startswith(b'#')] + assert len(frames) == 5, result.stdout + elif player == 'vlc': + assert b'using video decoder module' in result.stderr, result.stderr.decode() + c = Client('?pw=viewer-secret') + try: + c.play() + s.stop_publisher() + assert s.proxy.wait_for(r'viewer disconnected .*\(publisher gone\)', timeout=30), s.proxy.log + finally: + c.close() + finally: + s.stop() diff --git a/tests/webadmin/test_video_page.py b/tests/webadmin/test_video_page.py index e445ca0..cd3827c 100644 --- a/tests/webadmin/test_video_page.py +++ b/tests/webadmin/test_video_page.py @@ -76,6 +76,9 @@ def test_fallback_command_is_offered(self, client, keydb_path): assert 'ffplay' in html assert 'http://' in html assert '/v1.ts' in html + assert 'rtsp://localhost:' in html + assert '/view/v1' in html and '-rtsp_transport tcp' in html + assert 'Mission Planner' in html and 'appsink name=outsink' in html def test_fallback_mentions_the_password_when_one_is_set(self, client, keydb_path): diff --git a/video.cpp b/video.cpp index d5e0dd6..d61dc87 100644 --- a/video.cpp +++ b/video.cpp @@ -898,9 +898,9 @@ void VideoChild::handle_rtsp(Slot &s, int idx, int fd, const std::string target = first.substr(sp1 + 1, sp2 - sp1 - 1); pw_present = http_query_value(target, "pw", pw); - // Publishers are authorised; viewers are not, and on this port an - // RTSP connection is a publisher (we do not parse enough to tell - // them apart -- see videortsp.h). + // The viewer classifier keeps /view/ RTSP playback connections. + // Everything reaching this splice is still a publisher and must + // pass publish admission before a backend is started. const video_admit_t r = auth_.admit(ke_, uint32_t(from.sin_addr.s_addr), pw_present ? &pw : nullptr, true, session_ok(idx), open_publish(idx), diff --git a/videortsp.h b/videortsp.h index cf78581..6c2d884 100644 --- a/videortsp.h +++ b/videortsp.h @@ -14,9 +14,9 @@ rejects an ANNOUNCE numbered 2. Splicing from byte zero sidesteps both: ffmpeg sees the whole exchange, starting at CSeq 1. - Consequence, and the reason RTSP *egress* is a separate phase: with no - RTSP parsing we cannot tell a publisher from a viewer, so an RTSP - connection to a video port is treated as a publisher. + Playback uses the reserved /view/ URL namespace and is handled by + RtspViewer. Other RTSP targets still enter this opaque publisher + splice from byte zero, preserving publishers' initial CSeq exchange. A native RTP depacketiser plus TS muxer was measured at ~2300-3000 lines against ~150 for this, would still not carry the PCM audio the diff --git a/videortspview.cpp b/videortspview.cpp new file mode 100644 index 0000000..b33b7a8 --- /dev/null +++ b/videortspview.cpp @@ -0,0 +1,420 @@ +#include "videortspview.h" +#include "videoview.h" +#include "videoauth.h" +#include "httpreq.h" + +#include +#include +#include +#include +#include +#include +#include + +namespace { +constexpr size_t REQUEST_LIMIT = 16384; +constexpr size_t RTP_PAYLOAD = 7 * TS_PACKET_SIZE; + +std::string path(const std::string &target) +{ + size_t start = 0; + if (target.compare(0, 7, "rtsp://") == 0) { + start = target.find('/', 7); + if (start == std::string::npos) return "/"; + } + return target.substr(start, target.find('?', start) - start); +} + +bool decimal(const std::string &s, unsigned &value, unsigned max) +{ + if (s.empty()) return false; + value = 0; + for (unsigned char c : s) { + if (c < '0' || c > '9' || value > max / 10) return false; + value = value * 10 + c - '0'; + if (value > max) return false; + } + return true; +} + +void put32(std::string &s, uint32_t v) +{ + for (int shift = 24; shift >= 0; shift -= 8) s += char(v >> shift); +} + +uint32_t timestamp() +{ + struct timespec t {}; + clock_gettime(CLOCK_MONOTONIC, &t); + return uint32_t(uint64_t(t.tv_sec) * 90000 + t.tv_nsec * 9 / 100000); +} + +std::string trim(const std::string &s) +{ + const size_t first = s.find_first_not_of(" \t"); + if (first == std::string::npos) return ""; + return s.substr(first, s.find_last_not_of(" \t") - first + 1); +} + +bool tcp_transport(const std::string &header, unsigned &channel) +{ + size_t offset = 0; + while (offset < header.size()) { + const size_t comma = header.find(',', offset); + const std::string entry = trim(header.substr(offset, comma-offset)); + const size_t semicolon = entry.find(';'); + bool valid = trim(entry.substr(0, semicolon)) == "RTP/AVP/TCP"; + bool have_channels = false; + unsigned first = 0, second; + size_t param = semicolon; + while (param != std::string::npos) { + const size_t next = entry.find(';', param + 1); + const std::string value = trim(entry.substr(param + 1, next-param-1)); + if (value == "multicast") valid = false; + if (value.compare(0, 12, "interleaved=") == 0) { + const std::string channels = value.substr(12); + const size_t dash = channels.find('-'); + valid = valid && !have_channels && dash != std::string::npos && + decimal(channels.substr(0, dash), first, 254) && + decimal(channels.substr(dash + 1), second, 255) && second == first + 1; + have_channels = true; + } + param = next; + } + if (valid) { + channel = first; + return true; + } + if (comma == std::string::npos) break; + offset = comma + 1; + } + return false; +} + +bool normal_scale(const std::string &scale) +{ + if (scale.empty()) return true; + char *end = nullptr; + errno = 0; + const double value = strtod(scale.c_str(), &end); + return errno == 0 && end != scale.c_str() && *end == '\0' && value == 1.0; +} + +bool valid_rtcp(const uint8_t *data, size_t size) +{ + if (size < 4) return false; + size_t offset = 0; + while (offset < size) { + const uint8_t *p = data + offset; + if (size-offset < 4 || (p[0] >> 6) != 2 || p[1] < 192 || p[1] > 223) + return false; + const size_t length = ((size_t(p[2]) << 8) | p[3]) * 4 + 4; + if (length > size-offset) return false; + if (p[1] == 201 && length < 8 + 24 * size_t(p[0] & 31)) return false; + if ((p[0] & 0x20) && (offset+length != size || p[length-1] == 0 || p[length-1] > length-4)) + return false; + offset += length; + } + return true; +} +} + +RtspViewer::RtspViewer(int fd, int port2, time_t now) + : fd_(fd), port2_(port2), connected_(now), progress_(now), control_(now) +{ + uint8_t random[14]; + if (RAND_bytes(random, sizeof(random)) != 1) { + closing_ = true; + reason_ = "RTSP random generation failed"; + return; + } + char id[17]; + for (size_t i = 0; i < 8; i++) snprintf(id + 2*i, 3, "%02x", random[i]); + session_ = id; + memcpy(&ssrc_, random + 8, sizeof(ssrc_)); + memcpy(&seq_, random + 12, sizeof(seq_)); +} + +bool RtspViewer::playback_target(const std::string &target) +{ + return path(target).compare(0, 6, "/view/") == 0; +} + +void RtspViewer::reply(const std::string &cseq, int code, const char *reason, + const std::string &headers, const std::string &body) +{ + out_ += "RTSP/1.0 " + std::to_string(code) + " " + reason + "\r\n" + "CSeq: " + cseq + "\r\n" + headers + + "Content-Length: " + std::to_string(body.size()) + "\r\n\r\n" + body; +} + +bool RtspViewer::request(const std::string &head, const KeyEntry &ke, int slot, + const VideoRing &ring, const VideoScanner &scanner, + time_t now) +{ + HttpRequest req; + if (req.feed(reinterpret_cast(head.data()), head.size()) != 1) + return false; + const size_t eol = head.find("\r\n"); + if (eol == std::string::npos || eol < 9 || head.substr(eol-9, 9) != " RTSP/1.0") + return false; + const std::string cseq = req.header("CSeq"); + for (unsigned char c : req.target()) { + if (c <= 0x20 || c == 0x7f) return false; + } + unsigned number; + if (req.header_count("CSeq") != 1 || !decimal(cseq, number, 2147483647)) + return false; + control_ = now; + const std::string &method = req.method(); + if (method == "OPTIONS") { + reply(cseq, 200, "OK", "Public: OPTIONS, DESCRIBE, SETUP, PLAY, PAUSE, TEARDOWN, GET_PARAMETER\r\n"); + return true; + } + const std::string base = "/view/v" + std::to_string(slot + 1); + const std::string target_path = path(req.target()); + if (target_path != base && target_path != base + "/" && target_path != base + "/trackID=0") { + reply(cseq, 404, "Not Found"); + return true; + } + + std::string pw, token; + const bool has_pw = http_query_value(req.target(), "pw", pw); + const bool has_token = http_query_value(req.target(), "t", token); + const bool has_basic = req.header_count("Authorization") != 0; + bool permitted = authorised_; + if (!authorised_ || has_pw || has_token || has_basic) { + permitted = (has_token && video_token_valid(ke, port2_, slot, token.c_str(), now)); + if (!permitted) { + if (!has_pw) pw = http_basic_password(req.header("Authorization")); + permitted = video_viewer_authorised(ke, pw); + } + } + if (!permitted) { + reply(cseq, 401, "Unauthorized", "WWW-Authenticate: Basic realm=\"SupportProxy video\"\r\n"); + if (++auth_failures_ >= 3) closing_ = true; + return true; + } + authorised_ = true; + if (scanner.matroska) { + reply(cseq, 415, "Unsupported Media Type"); + return true; + } + if (method == "DESCRIBE") { + uint64_t anchor; + if (!scanner.join_offset(anchor) || !ring.resident(anchor)) { + reply(cseq, 503, "Service Unavailable"); + return true; + } + // A relative track URL deliberately carries no password or token. + // Admission belongs to this TCP connection, including subsequent SETUP. + std::string location = req.target().substr(0, req.target().find('?')); + if (location.back() != '/') location += '/'; + const std::string sdp = + "v=0\r\no=- " + std::to_string(ssrc_) + " 1 IN IP4 0.0.0.0\r\n" + "s=SupportProxy video\r\nc=IN IP4 0.0.0.0\r\nt=0 0\r\n" + "a=control:*\r\nm=video 0 RTP/AVP 33\r\n" + "a=rtpmap:33 MP2T/90000\r\na=control:trackID=0\r\n"; + reply(cseq, 200, "OK", "Content-Type: application/sdp\r\nContent-Base: " + location + "\r\n", sdp); + } else if (method == "SETUP") { + if (playing_) { + reply(cseq, 455, "Method Not Valid in This State"); + return true; + } + // Only interleaved unicast TCP. Never connect or send UDP to a + // client-supplied destination (including Transport destination=). + unsigned first; + if (!tcp_transport(req.header("Transport"), first)) { + reply(cseq, 461, "Unsupported Transport"); + return true; + } + const std::string requested_session = req.header("Session"); + if (!requested_session.empty() && requested_session != session_) { + reply(cseq, 454, "Session Not Found"); + return true; + } + setup_ = true; + channel_ = uint8_t(first); + reply(cseq, 200, "OK", "Session: " + session_ + ";timeout=" + + std::to_string(VIDEO_RTSP_SESSION_TIMEOUT_S) + "\r\n" + "Transport: RTP/AVP/TCP;unicast;interleaved=" + std::to_string(first) + "-" + + std::to_string(first + 1) + "\r\n"); + } else if (method == "PLAY" || method == "PAUSE" || method == "TEARDOWN" || method == "GET_PARAMETER") { + if (!setup_ || req.header("Session") != session_) { + reply(cseq, 454, "Session Not Found"); + return true; + } + if (method == "PLAY") { + // Live-only. Seeking and Scale would need different ring policy. + const std::string range = req.header("Range"); + if ((!range.empty() && range != "npt=0.000-" && range != "npt=0-" && range != "npt=now-") || + !normal_scale(req.header("Scale"))) { + reply(cseq, 457, "Invalid Range"); + return true; + } + // A short pause must not rewind within the existing RTP + // session. Only the first PLAY or an overwritten cursor + // needs a new keyframe anchor. + if (!playing_ && (!played_ || !ring.resident(read_pos_)) && + (!scanner.join_offset(read_pos_) || !ring.resident(read_pos_))) { + reply(cseq, 503, "Service Unavailable"); + return true; + } + playing_ = true; + played_ = true; + progress_ = now; + } else if (method == "PAUSE") { + playing_ = false; + } else if (method == "TEARDOWN") { + playing_ = false; + closing_ = true; + } + reply(cseq, 200, "OK", "Session: " + session_ + "\r\n"); + } else { + reply(cseq, 405, "Method Not Allowed"); + } + return true; +} + +bool RtspViewer::on_readable(const KeyEntry &ke, int slot, const VideoRing &ring, + const VideoScanner &scanner, time_t now) +{ + if (closing_ && sent_ == out_.size()) return false; + if (!authorised_ && now-connected_ > VIDEO_RTSP_AUTH_TIMEOUT_S) { + reason_ = "RTSP authentication timeout"; + return false; + } + uint8_t buf[4096]; + const ssize_t n = recv(fd_, buf, sizeof(buf), 0); + if (n == 0) return false; + if (n < 0) return errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR; + // EPOLLIN remains level-triggered while queued output is flushed. + // Consume and discard further input so a blocked TEARDOWN reply + // cannot spin the shared video child on unread data or EOF. + if (closing_) return true; + input_.append(reinterpret_cast(buf), size_t(n)); + if (input_.size() > REQUEST_LIMIT) return false; + while (!input_.empty()) { + if (input_[0] == '$') { + if (input_.size() < 4) break; + const size_t len = (uint8_t(input_[2]) << 8) | uint8_t(input_[3]); + if (!setup_ || uint8_t(input_[1]) != channel_ + 1 || len > REQUEST_LIMIT - 4) + return false; + if (input_.size() < len + 4) break; + if (!valid_rtcp(reinterpret_cast(input_.data()+4), len)) + return false; + control_ = now; + input_.erase(0, len + 4); // bounded RTCP receiver reports + continue; + } + const size_t end = input_.find("\r\n\r\n"); + if (end == std::string::npos) break; + const size_t header_size = end + 4; + HttpRequest req; + if (req.feed(reinterpret_cast(input_.data()), header_size) != 1) + return false; + unsigned body_size = 0; + if (req.header_count("Content-Length") > 1 || req.header_count("Transfer-Encoding") || + (req.header_count("Content-Length") && + !decimal(req.header("Content-Length"), body_size, REQUEST_LIMIT))) return false; + if (header_size + body_size > REQUEST_LIMIT) return false; + if (input_.size() < header_size + body_size) break; + if (!request(input_.substr(0, header_size), ke, slot, ring, scanner, now)) return false; + input_.erase(0, header_size + body_size); + if (out_.size() > REQUEST_LIMIT) return false; + if (closing_) break; + } + return true; +} + +bool RtspViewer::flush(time_t now) +{ + while (sent_ < out_.size()) { + const ssize_t n = send(fd_, out_.data() + sent_, out_.size() - sent_, MSG_NOSIGNAL); + if (n < 0 && errno == EINTR) continue; + if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + blocked_ = true; + if (now - progress_ > VIDEO_VIEWER_STUCK_S) { + reason_ = "stalled"; + return false; + } + return true; + } + if (n <= 0) return false; + sent_ += size_t(n); + progress_ = now; + } + out_.clear(); sent_ = 0; blocked_ = false; + return true; +} + +void RtspViewer::sender_report(uint32_t rtp_time) +{ + // RFC 3550 compound SR + SDES, on the negotiated RTCP channel. + std::string report; + report += char(0x80); report += char(200); report += char(0); report += char(6); + put32(report, ssrc_); + struct timespec t {}; + clock_gettime(CLOCK_REALTIME, &t); + put32(report, uint32_t(t.tv_sec + 2208988800ULL)); + put32(report, uint32_t((uint64_t(t.tv_nsec) << 32) / 1000000000)); + put32(report, rtp_time); put32(report, packets_); put32(report, octets_); + const std::string cname = "SupportProxy-" + session_; + std::string sdes; + put32(sdes, ssrc_); sdes += char(1); sdes += char(cname.size()); sdes += cname; + sdes += char(0); + while (sdes.size() % 4) sdes += char(0); + report += char(0x81); report += char(202); + report += char((sdes.size()/4) >> 8); report += char(sdes.size()/4); + report += sdes; + out_ += '$'; out_ += char(channel_ + 1); + out_ += char(report.size() >> 8); out_ += char(report.size()); out_ += report; +} + +bool RtspViewer::on_writable(const VideoRing &ring, time_t now) +{ + if (!authorised_ && now-connected_ > VIDEO_RTSP_AUTH_TIMEOUT_S) { + reason_ = "RTSP authentication timeout"; + return false; + } + // Check even while a partial TCP frame is blocked. Waiting to flush + // it first would postpone the ring's drop policy until the socket + // becomes writable, which a wedged viewer might never do. + if (playing_ && !ring.resident(read_pos_)) { + reason_ = "fell too far behind (lapped)"; + return false; + } + if (!flush(now)) return false; + if (!out_.empty()) return true; + if (closing_) return false; + if (now - control_ > VIDEO_RTSP_SESSION_TIMEOUT_S + VIDEO_RTSP_TIMEOUT_GRACE_S) { + reason_ = "RTSP control timeout"; + return false; + } + if (!playing_) return true; + size_t budget = VIDEO_VIEWER_WRITE_CHUNK; + while (budget >= TS_PACKET_SIZE) { + uint8_t payload[RTP_PAYLOAD]; + const size_t want = std::min(budget, sizeof(payload)) / TS_PACKET_SIZE * TS_PACKET_SIZE; + const size_t got = ring.read_at(read_pos_, payload, want) / TS_PACKET_SIZE * TS_PACKET_SIZE; + if (!got) break; + const uint32_t rtp_time = timestamp(); + if (!report_ || now - report_ >= 5) { + sender_report(rtp_time); + report_ = now; + } + const size_t length = got + 12; + out_ += '$'; out_ += char(channel_); + out_ += char(length >> 8); out_ += char(length); + out_ += char(0x80); out_ += char(33); + out_ += char(seq_ >> 8); out_ += char(seq_); seq_++; + put32(out_, rtp_time); put32(out_, ssrc_); + out_.append(reinterpret_cast(payload), got); + read_pos_ += got; bytes_sent_ += got; + packets_++; octets_ += uint32_t(got); + budget -= got; + if (!flush(now)) return false; + if (!out_.empty()) break; + } + return true; +} diff --git a/videortspview.h b/videortspview.h new file mode 100644 index 0000000..859cd35 --- /dev/null +++ b/videortspview.h @@ -0,0 +1,51 @@ +/* RTSP playback of the existing MPEG-TS ring, over interleaved TCP. + * The socket remains owned by VideoViewer. No child or extra ports are needed. + */ +#pragma once + +#include +#include +#include + +#include "keydb.h" +#include "videomkv.h" +#include "videostream.h" + +#define VIDEO_RTSP_AUTH_TIMEOUT_S 15 +#define VIDEO_RTSP_SESSION_TIMEOUT_S 60 +#define VIDEO_RTSP_TIMEOUT_GRACE_S 5 + +class RtspViewer { +public: + RtspViewer(int fd, int port2, time_t now); + bool on_readable(const KeyEntry &ke, int slot, const VideoRing &ring, + const VideoScanner &scanner, time_t now); + bool on_writable(const VideoRing &ring, time_t now); + bool wants_write() const { return blocked_ || sent_ < out_.size(); } + uint64_t bytes_sent() const { return bytes_sent_; } + const char *drop_reason() const { return reason_; } + static bool playback_target(const std::string &target); + +private: + int fd_, port2_; + std::string input_, out_, session_; + size_t sent_ = 0; + bool authorised_ = false, setup_ = false, playing_ = false; + bool played_ = false; + bool closing_ = false, blocked_ = false; + uint8_t channel_ = 0; + uint16_t seq_ = 0; + uint32_t ssrc_ = 0; + uint64_t read_pos_ = 0, bytes_sent_ = 0; + uint32_t packets_ = 0, octets_ = 0; + time_t connected_, progress_, control_, report_ = 0; + unsigned auth_failures_ = 0; + const char *reason_ = "RTSP connection closed"; + + bool request(const std::string &head, const KeyEntry &ke, int slot, + const VideoRing &ring, const VideoScanner &scanner, time_t now); + void reply(const std::string &cseq, int code, const char *reason, + const std::string &headers = "", const std::string &body = ""); + bool flush(time_t now); + void sender_report(uint32_t timestamp); +}; diff --git a/videoview.cpp b/videoview.cpp index 054fe9c..2a8131f 100644 --- a/videoview.cpp +++ b/videoview.cpp @@ -61,6 +61,8 @@ void VideoViewer::start(int fd, int port2, uint32_t peer_ip_be, void VideoViewer::close(void) { + delete rtsp_view_; + rtsp_view_ = nullptr; if (ws_ != nullptr) { delete ws_; // frees SSL objects; the fd is ours to close ws_ = nullptr; @@ -137,6 +139,7 @@ bool VideoViewer::on_readable(const struct KeyEntry &ke, int slot, const VideoRing &ring, const VideoScanner &scanner, time_t now) { + if (rtsp_view_) return rtsp_view_->on_readable(ke, slot, ring, scanner, now); if (state_ != VV_DETECT) { if (kind_ == VVK_WS) { // Let the WebSocket consume control frames (ping/close). @@ -197,7 +200,8 @@ bool VideoViewer::on_readable(const struct KeyEntry &ke, int slot, } /* - An RTSP request line means this is a publisher, not a viewer. + RTSP playback uses the reserved /view/ namespace. Other targets + remain untouched for the publisher splice, including OPTIONS. It has to be recognised here rather than at accept(): a publisher connects and only then sends OPTIONS, so at accept there is usually nothing to look at, and classifying on an empty peek @@ -206,7 +210,7 @@ bool VideoViewer::on_readable(const struct KeyEntry &ke, int slot, */ static const char *rtsp_methods[] = { "OPTIONS ", "ANNOUNCE ", "DESCRIBE ", "SETUP ", "PLAY ", - "RECORD ", "TEARDOWN ", "GET_PARAMETER ", "SET_PARAMETER ", + "RECORD ", "PAUSE ", "TEARDOWN ", "GET_PARAMETER ", "SET_PARAMETER ", }; for (const char *m : rtsp_methods) { const size_t len = strlen(m); @@ -224,6 +228,15 @@ bool VideoViewer::on_readable(const struct KeyEntry &ke, int slot, holding_bytes_ = true; return true; } + const std::string line(reinterpret_cast(buf), size_t(n)); + const size_t sp = line.find(' ', len); + if (sp != std::string::npos && + RtspViewer::playback_target(line.substr(len, sp-len))) { + kind_ = VVK_RTSP_VIEW; + state_ = VV_STREAMING; + rtsp_view_ = new RtspViewer(fd_, port2_, now); + return rtsp_view_->on_readable(ke, slot, ring, scanner, now); + } kind_ = VVK_RTSP; return true; // the child takes the socket from here } @@ -365,6 +378,7 @@ bool VideoViewer::flush(time_t now) bool VideoViewer::on_writable(const VideoRing &ring, time_t now) { + if (rtsp_view_) return rtsp_view_->on_writable(ring, now); if (!flush(now)) { return false; } @@ -469,6 +483,7 @@ bool VideoViewer::on_writable(const VideoRing &ring, time_t now) bool VideoViewer::wants_write(void) const { + if (rtsp_view_) return rtsp_view_->wants_write(); return blocked_ || out_sent_ < out_.size(); } diff --git a/videoview.h b/videoview.h index efadb24..b89aadd 100644 --- a/videoview.h +++ b/videoview.h @@ -25,6 +25,7 @@ #include "keydb.h" #include "videostream.h" #include "videomkv.h" +#include "videortspview.h" // Per-slot viewer cap. The limit that bites first is egress bandwidth, // not CPU: 32 viewers of an 8 Mbit/s stream is 256 Mbit/s. @@ -57,6 +58,7 @@ enum viewer_kind { VVK_RTSP, // RTSP: handed to the ingest splice, not served VVK_MKV, // HTTP chunked Matroska publisher VVK_RTMP, // RTMP publish: same, a different backend + VVK_RTSP_VIEW, // RTSP playback on /view/vN }; class VideoViewer { @@ -124,8 +126,10 @@ class VideoViewer { uint32_t peer_ip_be(void) const { return peer_ip_be_; } uint16_t peer_port_be(void) const { return peer_port_be_; } time_t connected_at(void) const { return connected_at_; } - uint64_t bytes_sent(void) const { return bytes_sent_; } - const char *drop_reason(void) const { return drop_reason_; } + uint64_t bytes_sent(void) const { return rtsp_view_ ? rtsp_view_->bytes_sent() : bytes_sent_; } + const char *drop_reason(void) const { + return drop_reason_[0] || !rtsp_view_ ? drop_reason_ : rtsp_view_->drop_reason(); + } // Only ever a string literal: the field is a borrowed pointer and // outlives nothing. @@ -160,6 +164,7 @@ class VideoViewer { only a plain-HTTP viewer's request is actually read off. */ WebSocket *ws_ = nullptr; + RtspViewer *rtsp_view_ = nullptr; bool ws_ready_ = false; const char *drop_reason_ = ""; diff --git a/webadmin/templates/video.html b/webadmin/templates/video.html index c8b157b..a917ea9 100644 --- a/webadmin/templates/video.html +++ b/webadmin/templates/video.html @@ -52,6 +52,14 @@

Slot {{ s.number }} — port {{ s.port }}

http://{{ host }}:{{ s.port }}/v{{ s.number }}.ts{% if s.needs_password %}?pw=YOUR_VIEWER_PASSWORD{% endif %} 
vlc --network-caching=200 --live-caching=200 \
     http://{{ host }}:{{ s.port }}/v{{ s.number }}.ts{% if s.needs_password %}?pw=YOUR_VIEWER_PASSWORD{% endif %}
+

RTSP playback (select TCP transport in your player):

+ 
rtsp://{{ host }}:{{ s.port }}/view/v{{ s.number }}{% if s.needs_password %}?pw=YOUR_VIEWER_PASSWORD{% endif %}
+ 
ffplay -rtsp_transport tcp -fflags nobuffer -flags low_delay \
+       -probesize 500000 -analyzeduration 1000000 \
+       'rtsp://{{ host }}:{{ s.port }}/view/v{{ s.number }}{% if s.needs_password %}?pw=YOUR_VIEWER_PASSWORD{% endif %}'
+

Mission Planner: right-click the HUD, select Video → + Set GStreamer Source, and paste:

+ 
rtspsrc location="rtsp://{{ host }}:{{ s.port }}/view/v{{ s.number }}{% if s.needs_password %}?pw=YOUR_VIEWER_PASSWORD{% endif %}" protocols=tcp latency=100 ! application/x-rtp ! decodebin3 ! queue max-size-buffers=1 leaky=2 ! videoconvert ! video/x-raw,format=BGRA ! appsink name=outsink sync=false
{% endif %}