video: close remaining credential downgrade paths

Reject duplicate RTMP connect properties, preserve the first credential across RTMP setup commands, and continue validating credentials on every RTSP request. Bound RTSP framing, reject ambiguous content lengths, and cover the reported downgrade cases with integration tests.

Author: tridge

httpreq.cpp

@@ -110,6 +110,36 @@ std::string HttpRequest::header(const char *name) const
     return "";
 }
 
+size_t HttpRequest::header_count(const char *name) const
+{
+    const std::string want = lower(name);
+    size_t count = 0;
+    size_t pos = 0;
+    while (pos < headers_.size()) {
+        size_t eol = headers_.find('\n', pos);
+        if (eol == std::string::npos) {
+            eol = headers_.size();
+        }
+        std::string line = headers_.substr(pos, eol - pos);
+        if (!line.empty() && line.back() == '\r') {
+            line.pop_back();
+        }
+        const size_t colon = line.find(':');
+        if (colon != std::string::npos) {
+            std::string field = line.substr(0, colon);
+            const size_t end = field.find_last_not_of(" \t");
+            if (end != std::string::npos) {
+                field.resize(end + 1);
+            }
+            if (lower(field) == want) {
+                count++;
+            }
+        }
+        pos = eol + 1;
+    }
+    return count;
+}
+
 std::string HttpRequest::query(const char *name) const
 {
     std::string value;

httpreq.h

@@ -32,6 +32,10 @@ class HttpRequest {
     // Header lookup, case-insensitive. Empty string when absent.
     std::string header(const char *name) const;
 
+    // Number of occurrences of a header, case-insensitive. Security
+    // parsers use this to reject ambiguous framing fields.
+    size_t header_count(const char *name) const;
+
     // Query parameter from the request target. Empty when absent.
     std::string query(const char *name) const;
 

tests/rtmp_client.py

@@ -29,7 +29,8 @@ def _amf_null():
 
 def _amf_obj(d):
     out = b'\x03'
-    for k, v in d.items():
+    items = d.items() if hasattr(d, 'items') else d
+    for k, v in items:
         kb = k.encode()
         out += struct.pack('>H', len(kb)) + kb
         out += _amf_str(v) if isinstance(v, str) else _amf_num(v)
@@ -117,7 +118,7 @@ def handshake(self):
         self.s.sendall(got[1:1537])                 # C2 echoes S1
         return got
 
-    def connect(self, password=None):
+    def connect(self, password=None, properties=None):
         stream = self.stream
         if password:
             stream = '%s?pw=%s' % (stream, password)
@@ -127,9 +128,11 @@ def connect(self, password=None):
         # told otherwise.
         self.s.sendall(self._chunk(2, 1, 0, 0,
                                    self.out_chunk.to_bytes(4, 'big')))
+        if properties is None:
+            properties = {'app': self.app, 'tcUrl': tc,
+                          'flashVer': 'test'}
         self.s.sendall(self._command(
-            _amf_str('connect') + _amf_num(1) +
-            _amf_obj({'app': self.app, 'tcUrl': tc, 'flashVer': 'test'})))
+            _amf_str('connect') + _amf_num(1) + _amf_obj(properties)))
         self._drain()
         self.s.sendall(self._command(
             _amf_str('createStream') + _amf_num(2) + _amf_null()))
@@ -138,6 +141,12 @@ def connect(self, password=None):
             _amf_str('publish') + _amf_num(3) + _amf_null() +
             _amf_str(stream) + _amf_str('live')))
 
+    def fcpublish(self, stream):
+        self.s.sendall(self._command(
+            _amf_str('FCPublish') + _amf_num(0) + _amf_null() +
+            _amf_str(stream)))
+        self._drain()
+
     def publish(self, first_tags=(), pipeline=False):
         """Send publish. With pipeline, media rides the same write.
 

tests/test_video_rtsp.py

@@ -388,6 +388,9 @@ class TestPublishPassword:
     """
 
     def test_accepted_with_no_mavlink_session_at_all(self, session, clip):
+        # ffmpeg later resolves the SDP control URI as
+        # ?pw=pubsecret/streamid=0. Reaching join=ready therefore also
+        # covers the guard's exact-password-plus-control-path handling.
         s = session(with_mav=False, publish_pass='pubsecret')
         s.pub = _publish_with('?pw=pubsecret', clip)
         assert s.proxy.wait_for(r'RTSP publisher', timeout=25), s.proxy.log
@@ -528,6 +531,43 @@ def test_fragmented_request_line_waits_for_the_credential(self, session):
         finally:
             sock.close()
 
+    def test_wrong_password_on_later_rtsp_request_is_refused(self, session):
+        """Session fallback on OPTIONS must not hide a credential later."""
+        s = session(with_mav=True, publish_pass='pubsecret', session_ok=True)
+        sock = socket.create_connection(('127.0.0.1', VPORT), 5)
+        sock.settimeout(10)
+        try:
+            sock.sendall(
+                ('OPTIONS rtsp://127.0.0.1:%d/cam RTSP/1.0\r\n'
+                 'CSeq: 1\r\n\r\n' % VPORT).encode())
+            assert b'RTSP/1.0 200' in sock.recv(4096)
+            sock.sendall(
+                ('ANNOUNCE rtsp://127.0.0.1:%d/cam?pw=wrong RTSP/1.0\r\n'
+                 'CSeq: 2\r\nContent-Length: 0\r\n\r\n' % VPORT).encode())
+            assert s.proxy.wait_for(r'wrong publish password', timeout=10), \
+                s.proxy.log
+        finally:
+            sock.close()
+
+    def test_ambiguous_rtsp_body_length_is_refused(self, session):
+        """A framing disagreement must not hide a later credential."""
+        s = session(with_mav=True, publish_pass='pubsecret', session_ok=True)
+        sock = socket.create_connection(('127.0.0.1', VPORT), 5)
+        sock.settimeout(10)
+        try:
+            sock.sendall(
+                ('OPTIONS rtsp://127.0.0.1:%d/cam RTSP/1.0\r\n'
+                 'CSeq: 1\r\n\r\n' % VPORT).encode())
+            assert b'RTSP/1.0 200' in sock.recv(4096)
+            sock.sendall(
+                b'ANNOUNCE rtsp://127.0.0.1/cam RTSP/1.0\r\n'
+                b'CSeq: 2\r\nContent-Length: 64\r\n'
+                b'Content-Length: 0\r\n\r\n')
+            assert s.proxy.wait_for(r'RTSP publisher gone', timeout=10), \
+                s.proxy.log
+        finally:
+            sock.close()
+
     def test_offering_none_falls_back_on_a_flagged_slot(self, session, clip):
         s = session(with_mav=True, publish_pass='pubsecret', session_ok=True)
         s.pub = _publish_with('', clip)
@@ -1090,6 +1130,36 @@ def test_supplied_rtmp_password_cannot_become_absent(self, tmp_path,
                 pub.close()
             s.stop()
 
+    @pytest.mark.parametrize('source', ['duplicate_app', 'fcpublish'])
+    def test_earlier_rtmp_password_cannot_be_erased(self, tmp_path, source):
+        """Every pre-publish credential source preserves explicit presence."""
+        wd = _workdir(tmp_path, publish_pass='secret', session_ok=True)
+        s = RtspSession(wd, with_mav=True)
+        pub = None
+        try:
+            pub = rtmp_client.RtmpPublisher(
+                '127.0.0.1', VPORT, app='PhoenixFPV', stream='FPV')
+            pub.handshake()
+            if source == 'duplicate_app':
+                pub.connect(properties=[
+                    ('app', 'PhoenixFPV?pw=wrong'),
+                    ('app', 'PhoenixFPV'),
+                    ('tcUrl', 'rtmp://127.0.0.1/PhoenixFPV'),
+                ])
+                assert s.proxy.wait_for(r'duplicate connect property',
+                                        timeout=15), s.proxy.log
+            else:
+                pub.connect()
+                pub.fcpublish('FPV?pw=wrong')
+                pub.publish()
+                assert s.proxy.wait_for(r'wrong publish password',
+                                        timeout=15), s.proxy.log
+            assert 'RTMP publishing' not in s.proxy.log
+        finally:
+            if pub:
+                pub.close()
+            s.stop()
+
     def test_no_orphan_backend_after_the_rtmp_publisher_leaves(
             self, tmp_path, clip):
         s = RtspSession(_workdir(tmp_path, rtmp_path='PhoenixFPV/FPV'))

video.cpp

@@ -9,6 +9,7 @@
  */
 #include "video.h"
 
+#include <algorithm>
 #include <initializer_list>
 #include <memory>
 #include <vector>
@@ -116,6 +117,19 @@ struct SpliceQueue {
 // instead of this process buffering without limit.
 #define SPLICE_QUEUE_MAX (1u * 1024 * 1024)
 
+/*
+  RTSP is relayed byte-for-byte, but admission credentials can occur on
+  any request URI in the session. Hold only the current request header
+  long enough to inspect its request line. Bodies and interleaved RTP are
+  counted and streamed without interpretation.
+ */
+struct RtspRequestGuard {
+    std::vector<uint8_t> buffered;
+    size_t opaque_left = 0;
+
+    void clear(void) { buffered.clear(); opaque_left = 0; }
+};
+
 /*
   One RTMP handshake that has not published yet.
 
@@ -155,6 +169,7 @@ struct Slot {
     int rtsp_client_fd = -1;
     SpliceQueue to_backend;   // bytes read from the client, owed to ffmpeg
     SpliceQueue to_client;    // and the other way
+    RtspRequestGuard rtsp_guard;
     /*
       The RTMP session that owns the slot, once one has published and
       been admitted. Null until then.
@@ -227,6 +242,8 @@ class VideoChild {
                      splice_proto_t proto);
     void close_rtsp(Slot &s, int idx, const char *why);
     bool pump_rtsp(Slot &s, int idx, int fd, time_t now);
+    bool guard_rtsp_requests(Slot &s, int idx, const uint8_t *buf, size_t n,
+                             time_t now);
     bool pump_rtmp(Slot &s, int idx, int fd, time_t now);
     bool rtmp_start_backend(Slot &s, int idx);
     bool rtmp_drain_owner(Slot &s, int idx, bool alive);
@@ -691,6 +708,7 @@ void VideoChild::handle_rtsp(Slot &s, int idx, int fd,
     }
     fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
     s.rtsp_client_fd = fd;
+    s.rtsp_guard.clear();
     s.pub_ip_be = uint32_t(from.sin_addr.s_addr);
     s.pub_port_be = from.sin_port;
     latch_publisher(s, idx, now);
@@ -734,6 +752,7 @@ void VideoChild::close_rtsp(Slot &s, int idx, const char *why)
     s.rtmp.reset();
     s.to_backend.clear();
     s.to_client.clear();
+    s.rtsp_guard.clear();
     s.rtsp.stop();
     s.rec.close_segment();
     s.recording = false;
@@ -778,6 +797,119 @@ bool VideoChild::splice_flush(int to_fd, SpliceQueue &q)
     return true;
 }
 
+/*
+  Filter the client-to-backend half of an RTSP splice. The first request
+  was already checked before the backend started, but a client can put a
+  credential on a later ANNOUNCE. Without continued inspection an absent
+  OPTIONS could take session fallback and a later wrong password would be
+  silently ignored.
+ */
+bool VideoChild::guard_rtsp_requests(Slot &s, int idx, const uint8_t *buf,
+                                     size_t n, time_t now)
+{
+    RtspRequestGuard &g = s.rtsp_guard;
+    g.buffered.insert(g.buffered.end(), buf, buf + n);
+
+    while (!g.buffered.empty()) {
+        if (g.opaque_left > 0) {
+            const size_t take = std::min(g.opaque_left, g.buffered.size());
+            s.to_backend.buf.insert(s.to_backend.buf.end(),
+                                    g.buffered.begin(),
+                                    g.buffered.begin() + long(take));
+            g.buffered.erase(g.buffered.begin(),
+                             g.buffered.begin() + long(take));
+            g.opaque_left -= take;
+            continue;
+        }
+
+        // Interleaved RTP/RTCP: '$', channel, 16-bit big-endian length.
+        if (g.buffered[0] == '$') {
+            if (g.buffered.size() < 4) {
+                return true;
+            }
+            g.opaque_left = (size_t(g.buffered[2]) << 8) | g.buffered[3];
+            s.to_backend.buf.insert(s.to_backend.buf.end(),
+                                    g.buffered.begin(),
+                                    g.buffered.begin() + 4);
+            g.buffered.erase(g.buffered.begin(), g.buffered.begin() + 4);
+            continue;
+        }
+
+        size_t header_len = 0;
+        for (size_t i = 0; i + 1 < g.buffered.size(); i++) {
+            if (g.buffered[i] == '\n' && g.buffered[i + 1] == '\n') {
+                header_len = i + 2;
+                break;
+            }
+            if (i + 3 < g.buffered.size()
+                && g.buffered[i] == '\r' && g.buffered[i + 1] == '\n'
+                && g.buffered[i + 2] == '\r'
+                && g.buffered[i + 3] == '\n') {
+                header_len = i + 4;
+                break;
+            }
+        }
+        if (header_len == 0) {
+            return g.buffered.size() <= HTTP_MAX_REQUEST;
+        }
+
+        HttpRequest req;
+        if (req.feed(g.buffered.data(), header_len) != 1) {
+            return false;
+        }
+        std::string pw;
+        if (http_query_value(req.target(), "pw", pw)) {
+            bool have_pw = false;
+            for (uint8_t b : ke_.video_publish_key) {
+                have_pw |= b != 0;
+            }
+            bool matches = video_password_matches(ke_.video_publish_key, pw);
+            /*
+              ffmpeg resolves an SDP control path after the whole source
+              URI, producing e.g. ?pw=secret/streamid=0. URI syntax makes
+              that suffix part of the query value. Accept it only when a
+              slash-delimited prefix is itself the exact password; the
+              peer still has to know the configured credential.
+             */
+            const size_t slash = pw.rfind('/');
+            if (!matches && slash != std::string::npos) {
+                matches = video_password_matches(
+                    ke_.video_publish_key, pw.substr(0, slash));
+            }
+            if (have_pw && !matches) {
+                log_reject(s, idx, s.pub_ip_be, VIDEO_ADMIT_BAD_PASSWORD,
+                           now);
+                return false;
+            }
+        }
+
+        const size_t content_length_count =
+            req.header_count("Content-Length");
+        const std::string content_length = req.header("Content-Length");
+        if (content_length_count > 1
+            || (content_length_count == 1 && content_length.empty())) {
+            return false;
+        }
+        if (!content_length.empty()) {
+            char *end = nullptr;
+            errno = 0;
+            const unsigned long long body =
+                strtoull(content_length.c_str(), &end, 10);
+            if (errno != 0 || end == content_length.c_str() || *end != '\0'
+                || body > 16u * 1024u * 1024u) {
+                return false;
+            }
+            g.opaque_left = size_t(body);
+        }
+        s.to_backend.buf.insert(s.to_backend.buf.end(),
+                                g.buffered.begin(),
+                                g.buffered.begin() + long(header_len));
+        g.buffered.erase(g.buffered.begin(),
+                         g.buffered.begin() + long(header_len));
+    }
+    return true;
+}
+
 // EPOLLOUT only while something is queued. Armed unconditionally it
 // would make epoll_wait return immediately for ever on an idle splice,
 // which is the same idle-viewer CPU burn measured earlier.
@@ -1156,7 +1288,13 @@ bool VideoChild::pump_rtsp(Slot &s, int idx, int fd, time_t now)
                 return false;
             }
             if (n > 0) {
-                out.buf.insert(out.buf.end(), buf, buf + n);
+                if (fd == s.rtsp_client_fd) {
+                    if (!guard_rtsp_requests(s, idx, buf, size_t(n), now)) {
+                        return false;
+                    }
+                } else {
+                    out.buf.insert(out.buf.end(), buf, buf + n);
+                }
                 if (!splice_flush(to_fd, out)) {
                     return false;
                 }