rtsp_runtime/
server.rs

1//! Server-side RTSP session engine — RFC 2326 Appendix A.2.
2//!
3//! [`ServerSession`] is sans-IO: [`ServerSession::handle_request`] parses an
4//! inbound request, validates it against the §A.2 server state table (see
5//! `private/specs-verified/rtp/rtsp-runtime/state-machines.md`), and returns the
6//! serialized response bytes plus typed [`ServerEvent`]s. A method not valid in
7//! the current state yields a `455 Method Not Valid In This State`
8//! (see `private/specs-verified/rtp/rtsp-runtime/methods-and-status.md`); state
9//! advances only when a `2xx` is actually sent.
10//!
11//! On `SETUP` the server allocates a `Session` id (if none yet) from the
12//! caller-supplied random source (RFC 2326 §3.4: session ids should be chosen
13//! randomly) and answers with the single transport-spec it chose (§12.39: the
14//! first one offered), or `461 Unsupported Transport` when none parses. Once a
15//! session exists, a request whose `Session` header names a different id — or
16//! a session-scoped request (`PLAY`/`PAUSE`/`RECORD`/`TEARDOWN`) with none —
17//! is answered `454 Session Not Found` (§11.3.2, §12.37).
18
19use rtsp_types::{Message, Method, Request, Response, StatusCode, Version, headers};
20
21use crate::error::{Error, Result};
22use crate::state::{SessionState, server_next_state};
23use crate::transport::Transport;
24
25/// A message body type: owned bytes.
26type Body = Vec<u8>;
27
28/// An event produced by [`ServerSession::handle_request`].
29#[non_exhaustive]
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub enum ServerEvent {
32    /// A request was accepted and the state machine advanced (2xx sent).
33    RequestAccepted {
34        /// The request method.
35        method: Method,
36        /// The `CSeq` echoed on the response.
37        cseq: u32,
38        /// The state after handling.
39        state: SessionState,
40    },
41    /// A request was rejected in the current state; a `455` was returned.
42    MethodNotValid {
43        /// The rejected method.
44        method: Method,
45        /// The state the request was rejected in.
46        state: SessionState,
47    },
48    /// A `SETUP` completed and a session id was allocated / reused.
49    SessionSetup {
50        /// The session id.
51        session_id: String,
52        /// The negotiated transport.
53        transport: Transport,
54    },
55    /// Interleaved binary media data arrived on the same TCP connection
56    /// (RFC 2326 §10.12) — an RTCP receiver report during PLAY, or media
57    /// during RECORD.
58    MediaData {
59        /// The interleaved channel id.
60        channel: u8,
61        /// The payload bytes (one upper-layer PDU).
62        data: Vec<u8>,
63    },
64}
65
66/// The source of `Session` id values a [`ServerSession`] draws from.
67type SessionIdSource = Box<dyn FnMut() -> u64 + Send>;
68
69/// A driveable RTSP server session (RFC 2326 §A.2).
70pub struct ServerSession {
71    state: SessionState,
72    session_id: Option<String>,
73    session_timeout: Option<u64>,
74    session_ids: SessionIdSource,
75    negotiated_transport: Option<Transport>,
76    server_header: String,
77}
78
79impl core::fmt::Debug for ServerSession {
80    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
81        f.debug_struct("ServerSession")
82            .field("state", &self.state)
83            .field("session_id", &self.session_id)
84            .field("session_timeout", &self.session_timeout)
85            .field("negotiated_transport", &self.negotiated_transport)
86            .field("server_header", &self.server_header)
87            .finish_non_exhaustive()
88    }
89}
90
91/// Methods that act on an established session and so must name it
92/// (RFC 2326 §12.37).
93const SESSION_SCOPED_METHODS: [Method; 4] = [
94    Method::Play,
95    Method::Pause,
96    Method::Record,
97    Method::Teardown,
98];
99
100impl ServerSession {
101    /// Creates a fresh server session in the `Init` state.
102    ///
103    /// `session_ids` is called once per allocated `Session` id and must return
104    /// values from a cryptographically secure random source (RFC 2326 §3.4) —
105    /// the id is the only thing tying later requests to this session. The
106    /// `tokio` adapter's `io::AsyncRtspServer::accept` supplies the OS RNG.
107    pub fn new(session_ids: impl FnMut() -> u64 + Send + 'static) -> Self {
108        ServerSession {
109            state: SessionState::Init,
110            session_id: None,
111            session_timeout: None,
112            session_ids: Box::new(session_ids),
113            negotiated_transport: None,
114            server_header: "rtsp-runtime".to_string(),
115        }
116    }
117
118    /// Sets the timeout (seconds) advertised in the `Session` header on SETUP.
119    pub fn with_session_timeout(mut self, seconds: u64) -> Self {
120        self.session_timeout = Some(seconds);
121        self
122    }
123
124    /// Replaces the id source with a counter starting at `seed`. For
125    /// deterministic tests only: the ids it yields are predictable, so it
126    /// must never be used for a server that accepts real connections.
127    #[doc(hidden)]
128    pub fn with_session_seed(mut self, seed: u64) -> Self {
129        let mut next = seed;
130        self.session_ids = Box::new(move || {
131            let id = next;
132            next = next.wrapping_add(1);
133            id
134        });
135        self
136    }
137
138    /// The current session state.
139    pub fn state(&self) -> SessionState {
140        self.state
141    }
142
143    /// The allocated session id, once a SETUP has been handled.
144    pub fn session_id(&self) -> Option<&str> {
145        self.session_id.as_deref()
146    }
147
148    /// The negotiated transport, once a SETUP has been handled.
149    pub fn negotiated_transport(&self) -> Option<&Transport> {
150        self.negotiated_transport.as_ref()
151    }
152
153    /// Parses an inbound request and returns the serialized response bytes plus
154    /// the events produced.
155    pub fn handle_request(&mut self, data: &[u8]) -> Result<(Vec<u8>, Vec<ServerEvent>)> {
156        let (message, _consumed) =
157            Message::<Body>::parse(data).map_err(|e| Error::MessageParse(format!("{e:?}")))?;
158        let request = match message {
159            Message::Request(r) => r,
160            _ => return Err(Error::MessageParse("expected an RTSP request".into())),
161        };
162        self.handle_parsed(request)
163    }
164
165    fn handle_parsed(&mut self, request: Request<Body>) -> Result<(Vec<u8>, Vec<ServerEvent>)> {
166        let method = request.method().clone();
167        let cseq = header_value(request.header(&headers::CSEQ))
168            .and_then(|s| s.trim().parse::<u32>().ok())
169            .ok_or(Error::MissingCSeq)?;
170
171        if !self.session_header_matches(&method, header_value(request.header(&headers::SESSION))) {
172            let resp = self.build_response(StatusCode::SessionNotFound, cseq, |b| b);
173            return Ok((serialize(&Message::from(resp))?, Vec::new()));
174        }
175
176        // Validate the method against the current state (§A.2).
177        let next_state = match server_next_state(self.state, &method) {
178            Ok(s) => s,
179            Err(_) => {
180                let resp = self.build_response(StatusCode::MethodNotValidInThisState, cseq, |b| b);
181                let bytes = serialize(&Message::from(resp))?;
182                return Ok((
183                    bytes,
184                    vec![ServerEvent::MethodNotValid {
185                        method,
186                        state: self.state,
187                    }],
188                ));
189            }
190        };
191
192        let mut events = Vec::new();
193
194        // SETUP: allocate session + negotiate transport.
195        if method == Method::Setup {
196            let chosen = header_value(request.header(&headers::TRANSPORT))
197                .and_then(|t| Transport::parse(t).ok())
198                .and_then(|t| t.first().cloned());
199            let Some(chosen) = chosen else {
200                let resp = self.build_response(StatusCode::UnsupportedTransport, cseq, |b| b);
201                return Ok((serialize(&Message::from(resp))?, events));
202            };
203            // §12.39: the reply carries the one spec the server selected.
204            let transport = Transport::single(chosen);
205            let session_id = self
206                .session_id
207                .clone()
208                .unwrap_or_else(|| self.allocate_session());
209            self.session_id = Some(session_id.clone());
210            self.negotiated_transport = Some(transport.clone());
211
212            let sid = session_id.clone();
213            let mut session_hdr = crate::session_header::SessionHeader::new(sid.clone());
214            session_hdr.timeout = self.session_timeout.map(std::time::Duration::from_secs);
215            let session_hdr = session_hdr.to_header_value()?;
216            let transport_hdr = transport.to_header_value()?;
217            let resp = self.build_response(StatusCode::Ok, cseq, |b| {
218                b.header(headers::SESSION, session_hdr)
219                    .header(headers::TRANSPORT, transport_hdr)
220            });
221            self.state = next_state;
222            events.push(ServerEvent::SessionSetup {
223                session_id,
224                transport,
225            });
226            events.push(ServerEvent::RequestAccepted {
227                method,
228                cseq,
229                state: self.state,
230            });
231            return Ok((serialize(&Message::from(resp))?, events));
232        }
233
234        // Other methods: echo Session if we have one, send 200, transition.
235        let session_hdr = self.session_id.clone();
236        let resp = self.build_response(StatusCode::Ok, cseq, |mut b| {
237            if let Some(sid) = &session_hdr {
238                b = b.header(headers::SESSION, sid.clone());
239            }
240            b
241        });
242        self.state = next_state;
243        if method == Method::Teardown {
244            self.session_id = None;
245            self.negotiated_transport = None;
246        }
247        events.push(ServerEvent::RequestAccepted {
248            method,
249            cseq,
250            state: self.state,
251        });
252        Ok((serialize(&Message::from(resp))?, events))
253    }
254
255    fn allocate_session(&mut self) -> String {
256        let id = (self.session_ids)();
257        format!("{id:016x}")
258    }
259
260    /// RFC 2326 §12.37: a `Session` header must name this session's id; a
261    /// session-scoped method must carry one once a session exists.
262    fn session_header_matches(&self, method: &Method, header: Option<&str>) -> bool {
263        match (header, &self.session_id) {
264            (Some(value), Some(ours)) => {
265                let theirs = crate::session_header::SessionHeader::parse(value)
266                    .map(|h| h.id)
267                    .unwrap_or_default();
268                broadcast_auth::constant_time_eq(theirs.as_bytes(), ours.as_bytes())
269            }
270            (Some(_), None) => false,
271            (None, Some(_)) => !SESSION_SCOPED_METHODS.contains(method),
272            (None, None) => true,
273        }
274    }
275
276    /// Builds a response with Version 1.0, CSeq, and Server headers, plus any
277    /// headers added by `f`.
278    fn build_response<F>(&self, status: StatusCode, cseq: u32, f: F) -> Response<Body>
279    where
280        F: FnOnce(rtsp_types::ResponseBuilder) -> rtsp_types::ResponseBuilder,
281    {
282        let builder = Response::builder(Version::V1_0, status)
283            .header(headers::CSEQ, cseq.to_string())
284            .header(headers::SERVER, self.server_header.clone());
285        f(builder).build(Vec::new())
286    }
287}
288
289fn header_value(h: Option<&headers::HeaderValue>) -> Option<&str> {
290    h.map(|v| v.as_str())
291}
292
293fn serialize(message: &Message<Body>) -> Result<Vec<u8>> {
294    let mut out = Vec::new();
295    message
296        .write(&mut out)
297        .map_err(|e| Error::MessageWrite(e.to_string()))?;
298    Ok(out)
299}
300
301#[cfg(test)]
302mod tests {
303    use super::*;
304
305    /// A session drawing ids from a counter unique per call site in this test
306    /// binary, standing in for a random source.
307    fn test_session() -> ServerSession {
308        use std::sync::atomic::{AtomicU64, Ordering};
309        static NEXT: AtomicU64 = AtomicU64::new(0x9e37_79b9_7f4a_7c15);
310        ServerSession::new(|| NEXT.fetch_add(0x2545_f491_4f6c_dd1d, Ordering::Relaxed))
311    }
312
313    fn req(bytes: &str) -> Vec<u8> {
314        bytes.replace('\n', "\r\n").into_bytes()
315    }
316
317    #[test]
318    fn a_request_without_cseq_is_rejected() {
319        let mut s = test_session();
320        assert_eq!(
321            s.handle_request(&req("OPTIONS rtsp://h/s RTSP/1.0\n\n"))
322                .unwrap_err(),
323            Error::MissingCSeq
324        );
325    }
326
327    #[test]
328    fn setup_transitions_init_to_ready() {
329        let mut s = test_session();
330        let (resp, events) = s
331            .handle_request(&req(
332                "SETUP rtsp://h/s RTSP/1.0\nCSeq: 1\nTransport: RTP/AVP/TCP;interleaved=0-1\n\n",
333            ))
334            .unwrap();
335        assert_eq!(s.state(), SessionState::Ready);
336        let text = String::from_utf8_lossy(&resp);
337        assert!(text.contains("200"));
338        assert!(text.contains("Session:"));
339        assert!(text.contains("Transport:"));
340        assert!(
341            events
342                .iter()
343                .any(|e| matches!(e, ServerEvent::SessionSetup { .. }))
344        );
345    }
346
347    #[test]
348    fn play_in_init_returns_455() {
349        let mut s = test_session();
350        let (resp, events) = s
351            .handle_request(&req("PLAY rtsp://h/s RTSP/1.0\nCSeq: 1\n\n"))
352            .unwrap();
353        assert_eq!(s.state(), SessionState::Init);
354        assert!(String::from_utf8_lossy(&resp).contains("455"));
355        assert!(
356            events
357                .iter()
358                .any(|e| matches!(e, ServerEvent::MethodNotValid { .. }))
359        );
360    }
361
362    fn setup(s: &mut ServerSession, transport: &str) -> String {
363        let (resp, _) = s
364            .handle_request(&req(&format!(
365                "SETUP rtsp://h/s RTSP/1.0\nCSeq: 1\nTransport: {transport}\n\n"
366            )))
367            .unwrap();
368        String::from_utf8(resp).unwrap()
369    }
370
371    #[test]
372    fn session_ids_differ_across_sessions_and_carry_64_bits() {
373        let mut a = test_session();
374        let mut b = test_session();
375        setup(&mut a, "RTP/AVP/TCP;interleaved=0-1");
376        setup(&mut b, "RTP/AVP/TCP;interleaved=0-1");
377        let (a, b) = (a.session_id().unwrap(), b.session_id().unwrap());
378        assert_ne!(a, b, "two sessions must not share an id");
379        assert!(a.len() >= 16, "id {a} is shorter than 64 bits of hex");
380    }
381
382    #[test]
383    fn request_with_wrong_session_header_gets_454() {
384        let mut s = test_session();
385        setup(&mut s, "RTP/AVP/TCP;interleaved=0-1");
386        let (resp, _) = s
387            .handle_request(&req(
388                "PLAY rtsp://h/s RTSP/1.0\nCSeq: 2\nSession: not-this-one\n\n",
389            ))
390            .unwrap();
391        assert!(
392            String::from_utf8_lossy(&resp).starts_with("RTSP/1.0 454"),
393            "got: {}",
394            String::from_utf8_lossy(&resp)
395        );
396        assert_eq!(s.state(), SessionState::Ready);
397        // Missing Session on a session-scoped method is rejected the same way.
398        let (resp, _) = s
399            .handle_request(&req("TEARDOWN rtsp://h/s RTSP/1.0\nCSeq: 3\n\n"))
400            .unwrap();
401        assert!(String::from_utf8_lossy(&resp).starts_with("RTSP/1.0 454"));
402        assert_eq!(s.state(), SessionState::Ready);
403        // The right id (with a timeout parameter) is accepted.
404        let sid = s.session_id().unwrap().to_string();
405        let (resp, _) = s
406            .handle_request(&req(&format!(
407                "PLAY rtsp://h/s RTSP/1.0\nCSeq: 4\nSession: {sid};timeout=60\n\n"
408            )))
409            .unwrap();
410        assert!(String::from_utf8_lossy(&resp).starts_with("RTSP/1.0 200"));
411        assert_eq!(s.state(), SessionState::Playing);
412    }
413
414    #[test]
415    fn setup_answers_exactly_one_transport() {
416        let mut s = test_session();
417        let text = setup(
418            &mut s,
419            "RTP/AVP;unicast;client_port=8000-8001,RTP/AVP/TCP;unicast;interleaved=0-1",
420        );
421        let line = text.lines().find(|l| l.starts_with("Transport:")).unwrap();
422        let answered = Transport::parse(line["Transport:".len()..].trim()).unwrap();
423        assert_eq!(answered.specs.len(), 1, "got: {line}");
424        assert_eq!(answered.first().unwrap().client_port, Some((8000, 8001)));
425        assert_eq!(s.negotiated_transport().unwrap().specs.len(), 1);
426    }
427
428    #[test]
429    fn unparseable_transport_gets_461() {
430        let mut s = test_session();
431        let text = setup(&mut s, "not a transport;;");
432        assert!(text.starts_with("RTSP/1.0 461"), "got: {text}");
433        assert_eq!(s.state(), SessionState::Init);
434    }
435}
rtsp_runtime/server.rs

rtsp_runtime/
server.rs
