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, MessageKind, 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 {
158                what: MessageKind::Request,
159                source: Box::new(e),
160            })?;
161        let request = match message {
162            Message::Request(r) => r,
163            _ => {
164                return Err(Error::MessageParse {
165                    what: MessageKind::Request,
166                    source: "expected an RTSP request".into(),
167                });
168            }
169        };
170        self.handle_parsed(request)
171    }
172
173    fn handle_parsed(&mut self, request: Request<Body>) -> Result<(Vec<u8>, Vec<ServerEvent>)> {
174        let method = request.method().clone();
175        let cseq = header_value(request.header(&headers::CSEQ))
176            .and_then(|s| s.trim().parse::<u32>().ok())
177            .ok_or(Error::MissingCSeq)?;
178
179        if !self.session_header_matches(&method, header_value(request.header(&headers::SESSION))) {
180            let resp = self.build_response(StatusCode::SessionNotFound, cseq, |b| b);
181            return Ok((serialize(&Message::from(resp))?, Vec::new()));
182        }
183
184        // Validate the method against the current state (§A.2).
185        let next_state = match server_next_state(self.state, &method) {
186            Ok(s) => s,
187            Err(_) => {
188                let resp = self.build_response(StatusCode::MethodNotValidInThisState, cseq, |b| b);
189                let bytes = serialize(&Message::from(resp))?;
190                return Ok((
191                    bytes,
192                    vec![ServerEvent::MethodNotValid {
193                        method,
194                        state: self.state,
195                    }],
196                ));
197            }
198        };
199
200        let mut events = Vec::new();
201
202        // SETUP: allocate session + negotiate transport.
203        if method == Method::Setup {
204            let chosen = header_value(request.header(&headers::TRANSPORT))
205                .and_then(|t| Transport::parse(t).ok())
206                .and_then(|t| t.first().cloned());
207            let Some(chosen) = chosen else {
208                let resp = self.build_response(StatusCode::UnsupportedTransport, cseq, |b| b);
209                return Ok((serialize(&Message::from(resp))?, events));
210            };
211            // §12.39: the reply carries the one spec the server selected.
212            let transport = Transport::single(chosen);
213            let session_id = self
214                .session_id
215                .clone()
216                .unwrap_or_else(|| self.allocate_session());
217            self.session_id = Some(session_id.clone());
218            self.negotiated_transport = Some(transport.clone());
219
220            let sid = session_id.clone();
221            let mut session_hdr = crate::session_header::SessionHeader::new(sid.clone());
222            session_hdr.timeout = self.session_timeout.map(std::time::Duration::from_secs);
223            let session_hdr = session_hdr.to_header_value()?;
224            let transport_hdr = transport.to_header_value()?;
225            let resp = self.build_response(StatusCode::Ok, cseq, |b| {
226                b.header(headers::SESSION, session_hdr)
227                    .header(headers::TRANSPORT, transport_hdr)
228            });
229            self.state = next_state;
230            events.push(ServerEvent::SessionSetup {
231                session_id,
232                transport,
233            });
234            events.push(ServerEvent::RequestAccepted {
235                method,
236                cseq,
237                state: self.state,
238            });
239            return Ok((serialize(&Message::from(resp))?, events));
240        }
241
242        // Other methods: echo Session if we have one, send 200, transition.
243        let session_hdr = self.session_id.clone();
244        let resp = self.build_response(StatusCode::Ok, cseq, |mut b| {
245            if let Some(sid) = &session_hdr {
246                b = b.header(headers::SESSION, sid.clone());
247            }
248            b
249        });
250        self.state = next_state;
251        if method == Method::Teardown {
252            self.session_id = None;
253            self.negotiated_transport = None;
254        }
255        events.push(ServerEvent::RequestAccepted {
256            method,
257            cseq,
258            state: self.state,
259        });
260        Ok((serialize(&Message::from(resp))?, events))
261    }
262
263    fn allocate_session(&mut self) -> String {
264        let id = (self.session_ids)();
265        format!("{id:016x}")
266    }
267
268    /// RFC 2326 §12.37: a `Session` header must name this session's id; a
269    /// session-scoped method must carry one once a session exists.
270    fn session_header_matches(&self, method: &Method, header: Option<&str>) -> bool {
271        match (header, &self.session_id) {
272            (Some(value), Some(ours)) => {
273                let theirs = crate::session_header::SessionHeader::parse(value)
274                    .map(|h| h.id)
275                    .unwrap_or_default();
276                broadcast_auth::constant_time_eq(theirs.as_bytes(), ours.as_bytes())
277            }
278            (Some(_), None) => false,
279            (None, Some(_)) => !SESSION_SCOPED_METHODS.contains(method),
280            (None, None) => true,
281        }
282    }
283
284    /// Builds a response with Version 1.0, CSeq, and Server headers, plus any
285    /// headers added by `f`.
286    fn build_response<F>(&self, status: StatusCode, cseq: u32, f: F) -> Response<Body>
287    where
288        F: FnOnce(rtsp_types::ResponseBuilder) -> rtsp_types::ResponseBuilder,
289    {
290        let builder = Response::builder(Version::V1_0, status)
291            .header(headers::CSEQ, cseq.to_string())
292            .header(headers::SERVER, self.server_header.clone());
293        f(builder).build(Vec::new())
294    }
295}
296
297fn header_value(h: Option<&headers::HeaderValue>) -> Option<&str> {
298    h.map(|v| v.as_str())
299}
300
301fn serialize(message: &Message<Body>) -> Result<Vec<u8>> {
302    let mut out = Vec::new();
303    message.write(&mut out).map_err(Error::MessageWrite)?;
304    Ok(out)
305}
306
307#[cfg(test)]
308mod tests {
309    use super::*;
310
311    /// A session drawing ids from a counter unique per call site in this test
312    /// binary, standing in for a random source.
313    fn test_session() -> ServerSession {
314        use std::sync::atomic::{AtomicU64, Ordering};
315        static NEXT: AtomicU64 = AtomicU64::new(0x9e37_79b9_7f4a_7c15);
316        ServerSession::new(|| NEXT.fetch_add(0x2545_f491_4f6c_dd1d, Ordering::Relaxed))
317    }
318
319    fn req(bytes: &str) -> Vec<u8> {
320        bytes.replace('\n', "\r\n").into_bytes()
321    }
322
323    #[test]
324    fn a_request_without_cseq_is_rejected() {
325        let mut s = test_session();
326        let err = s
327            .handle_request(&req("OPTIONS rtsp://h/s RTSP/1.0\n\n"))
328            .unwrap_err();
329        assert!(matches!(err, Error::MissingCSeq), "{err:?}");
330    }
331
332    #[test]
333    fn setup_transitions_init_to_ready() {
334        let mut s = test_session();
335        let (resp, events) = s
336            .handle_request(&req(
337                "SETUP rtsp://h/s RTSP/1.0\nCSeq: 1\nTransport: RTP/AVP/TCP;interleaved=0-1\n\n",
338            ))
339            .unwrap();
340        assert_eq!(s.state(), SessionState::Ready);
341        let text = String::from_utf8_lossy(&resp);
342        assert!(text.contains("200"));
343        assert!(text.contains("Session:"));
344        assert!(text.contains("Transport:"));
345        assert!(
346            events
347                .iter()
348                .any(|e| matches!(e, ServerEvent::SessionSetup { .. }))
349        );
350    }
351
352    #[test]
353    fn play_in_init_returns_455() {
354        let mut s = test_session();
355        let (resp, events) = s
356            .handle_request(&req("PLAY rtsp://h/s RTSP/1.0\nCSeq: 1\n\n"))
357            .unwrap();
358        assert_eq!(s.state(), SessionState::Init);
359        assert!(String::from_utf8_lossy(&resp).contains("455"));
360        assert!(
361            events
362                .iter()
363                .any(|e| matches!(e, ServerEvent::MethodNotValid { .. }))
364        );
365    }
366
367    fn setup(s: &mut ServerSession, transport: &str) -> String {
368        let (resp, _) = s
369            .handle_request(&req(&format!(
370                "SETUP rtsp://h/s RTSP/1.0\nCSeq: 1\nTransport: {transport}\n\n"
371            )))
372            .unwrap();
373        String::from_utf8(resp).unwrap()
374    }
375
376    #[test]
377    fn session_ids_differ_across_sessions_and_carry_64_bits() {
378        let mut a = test_session();
379        let mut b = test_session();
380        setup(&mut a, "RTP/AVP/TCP;interleaved=0-1");
381        setup(&mut b, "RTP/AVP/TCP;interleaved=0-1");
382        let (a, b) = (a.session_id().unwrap(), b.session_id().unwrap());
383        assert_ne!(a, b, "two sessions must not share an id");
384        assert!(a.len() >= 16, "id {a} is shorter than 64 bits of hex");
385    }
386
387    #[test]
388    fn request_with_wrong_session_header_gets_454() {
389        let mut s = test_session();
390        setup(&mut s, "RTP/AVP/TCP;interleaved=0-1");
391        let (resp, _) = s
392            .handle_request(&req(
393                "PLAY rtsp://h/s RTSP/1.0\nCSeq: 2\nSession: not-this-one\n\n",
394            ))
395            .unwrap();
396        assert!(
397            String::from_utf8_lossy(&resp).starts_with("RTSP/1.0 454"),
398            "got: {}",
399            String::from_utf8_lossy(&resp)
400        );
401        assert_eq!(s.state(), SessionState::Ready);
402        // Missing Session on a session-scoped method is rejected the same way.
403        let (resp, _) = s
404            .handle_request(&req("TEARDOWN rtsp://h/s RTSP/1.0\nCSeq: 3\n\n"))
405            .unwrap();
406        assert!(String::from_utf8_lossy(&resp).starts_with("RTSP/1.0 454"));
407        assert_eq!(s.state(), SessionState::Ready);
408        // The right id (with a timeout parameter) is accepted.
409        let sid = s.session_id().unwrap().to_string();
410        let (resp, _) = s
411            .handle_request(&req(&format!(
412                "PLAY rtsp://h/s RTSP/1.0\nCSeq: 4\nSession: {sid};timeout=60\n\n"
413            )))
414            .unwrap();
415        assert!(String::from_utf8_lossy(&resp).starts_with("RTSP/1.0 200"));
416        assert_eq!(s.state(), SessionState::Playing);
417    }
418
419    #[test]
420    fn setup_answers_exactly_one_transport() {
421        let mut s = test_session();
422        let text = setup(
423            &mut s,
424            "RTP/AVP;unicast;client_port=8000-8001,RTP/AVP/TCP;unicast;interleaved=0-1",
425        );
426        let line = text.lines().find(|l| l.starts_with("Transport:")).unwrap();
427        let answered = Transport::parse(line["Transport:".len()..].trim()).unwrap();
428        assert_eq!(answered.specs.len(), 1, "got: {line}");
429        assert_eq!(answered.first().unwrap().client_port, Some((8000, 8001)));
430        assert_eq!(s.negotiated_transport().unwrap().specs.len(), 1);
431    }
432
433    #[test]
434    fn unparseable_transport_gets_461() {
435        let mut s = test_session();
436        let text = setup(&mut s, "not a transport;;");
437        assert!(text.starts_with("RTSP/1.0 461"), "got: {text}");
438        assert_eq!(s.state(), SessionState::Init);
439    }
440}
rtsp_runtime/server.rs

rtsp_runtime/
server.rs
