rtsp_runtime/
client.rs

1//! Client-side RTSP session engine — RFC 2326 Appendix A.1.
2//!
3//! [`ClientSession`] is a sans-IO driver: request-builder methods return the
4//! outbound bytes to send, and [`ClientSession::handle_data`] consumes inbound
5//! bytes (responses and interleaved `$` frames) and returns typed
6//! [`ClientEvent`]s. It holds the session state, the next `CSeq`, the negotiated
7//! `Session` id and timeout, optional credentials, and the digest
8//! [`Authenticator`].
9//!
10//! Behaviour implemented here (see [`docs/state-machines.md`](../docs/state-machines.md),
11//! [`docs/methods-and-status.md`](../docs/methods-and-status.md),
12//! [`docs/auth.md`](../docs/auth.md)):
13//!
14//! - Request builders reject any method not valid in the current state
15//!   ([`Error::MethodNotValidInState`]) before emitting bytes.
16//! - Every request carries an incrementing `CSeq`, the `Session` id once known,
17//!   and (once authenticated) a freshly-computed `Authorization` header.
18//! - A `2xx` response advances the state per the §A.1 table; a `3xx` resets it
19//!   to `Init`.
20//! - A `401` with configured credentials transparently re-sends the request
21//!   with `Authorization` (a new `CSeq`), including on `stale=true`.
22//! - The `Session` id and timeout are captured from the SETUP response.
23//! - Interleaved frames are surfaced as [`ClientEvent::MediaData`].
24
25use std::collections::HashMap;
26
27use rtsp_types::{headers, Message, Method, Request, StatusCode, Version};
28
29use crate::auth::{Authenticator, Credentials};
30use crate::error::{Error, Result};
31use crate::interleaved::{self, MAGIC};
32use crate::state::{client_next_state, SessionState};
33use crate::transport::Transport;
34
35/// A message body type: owned bytes.
36type Body = Vec<u8>;
37
38/// Record of a request the client has sent and is awaiting a response for.
39#[derive(Debug, Clone)]
40struct Pending {
41    method: Method,
42    uri: String,
43    /// The full request, retained so it can be re-signed and re-sent on a 401.
44    request: Request<Body>,
45    /// Whether an auth retry has already been attempted for this logical request.
46    auth_retried: bool,
47}
48
49/// An event produced by [`ClientSession::handle_data`].
50#[non_exhaustive]
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub enum ClientEvent {
53    /// A response was correlated to a request and the state machine updated.
54    Response {
55        /// The `CSeq` of the correlated request.
56        cseq: u32,
57        /// The method that was responded to.
58        method: Method,
59        /// The response status code.
60        status: StatusCode,
61        /// The response body (e.g. the SDP for a DESCRIBE), possibly empty.
62        body: Vec<u8>,
63    },
64    /// The engine transparently re-sent a request with an `Authorization`
65    /// header after a `401`. The caller MUST write `request` to the socket.
66    AuthRetry {
67        /// The method being retried.
68        method: Method,
69        /// The `CSeq` assigned to the retried request.
70        cseq: u32,
71        /// The serialized retried request bytes to send.
72        request: Vec<u8>,
73    },
74    /// Interleaved binary media data (RFC 2326 §10.12).
75    MediaData {
76        /// The interleaved channel id.
77        channel: u8,
78        /// The payload bytes (one upper-layer PDU).
79        data: Vec<u8>,
80    },
81}
82
83/// A driveable RTSP client session (RFC 2326 §A.1).
84#[derive(Debug)]
85pub struct ClientSession {
86    state: SessionState,
87    next_cseq: u32,
88    session_id: Option<String>,
89    session_timeout: Option<u64>,
90    credentials: Option<Credentials>,
91    authenticator: Option<Authenticator>,
92    negotiated_transport: Option<Transport>,
93    pending: HashMap<u32, Pending>,
94    /// Accumulates inbound bytes across `handle_data` calls (partial frames /
95    /// partial messages).
96    inbound: Vec<u8>,
97    user_agent: String,
98}
99
100impl Default for ClientSession {
101    fn default() -> Self {
102        Self::new()
103    }
104}
105
106impl ClientSession {
107    /// Creates a fresh client session in the `Init` state with `CSeq` starting
108    /// at 1.
109    pub fn new() -> Self {
110        ClientSession {
111            state: SessionState::Init,
112            next_cseq: 1,
113            session_id: None,
114            session_timeout: None,
115            credentials: None,
116            authenticator: None,
117            negotiated_transport: None,
118            pending: HashMap::new(),
119            inbound: Vec::new(),
120            user_agent: "rtsp-runtime".to_string(),
121        }
122    }
123
124    /// Attaches credentials so the engine can answer `401` challenges (§14).
125    pub fn with_credentials(mut self, credentials: Credentials) -> Self {
126        self.credentials = Some(credentials);
127        self
128    }
129
130    /// Overrides the `User-Agent` header value sent on requests.
131    pub fn with_user_agent(mut self, ua: impl Into<String>) -> Self {
132        self.user_agent = ua.into();
133        self
134    }
135
136    /// The current session state.
137    pub fn state(&self) -> SessionState {
138        self.state
139    }
140
141    /// The negotiated session id, once a SETUP response has been processed.
142    pub fn session_id(&self) -> Option<&str> {
143        self.session_id.as_deref()
144    }
145
146    /// The session timeout in seconds, if the SETUP response declared one.
147    pub fn session_timeout(&self) -> Option<u64> {
148        self.session_timeout
149    }
150
151    /// The transport negotiated in the SETUP response, if any.
152    pub fn negotiated_transport(&self) -> Option<&Transport> {
153        self.negotiated_transport.as_ref()
154    }
155
156    // --- Request builders -------------------------------------------------
157
158    /// Builds an `OPTIONS` request (state-neutral).
159    pub fn options(&mut self, uri: &str) -> Result<Vec<u8>> {
160        self.build_request(Method::Options, uri, None, &[])
161    }
162
163    /// Builds a `DESCRIBE` request with `Accept: application/sdp` (state-neutral).
164    pub fn describe(&mut self, uri: &str) -> Result<Vec<u8>> {
165        self.build_request(
166            Method::Describe,
167            uri,
168            None,
169            &[(headers::ACCEPT, "application/sdp".to_string())],
170        )
171    }
172
173    /// Builds a `SETUP` request carrying the given `Transport` (Init/Ready/…).
174    pub fn setup(&mut self, uri: &str, transport: &Transport) -> Result<Vec<u8>> {
175        self.build_request(
176            Method::Setup,
177            uri,
178            None,
179            &[(headers::TRANSPORT, transport.to_header_value())],
180        )
181    }
182
183    /// Builds a `PLAY` request (valid in Ready/Playing).
184    pub fn play(&mut self, uri: &str) -> Result<Vec<u8>> {
185        self.build_request(Method::Play, uri, None, &[])
186    }
187
188    /// Builds a `PAUSE` request (valid in Playing/Recording).
189    pub fn pause(&mut self, uri: &str) -> Result<Vec<u8>> {
190        self.build_request(Method::Pause, uri, None, &[])
191    }
192
193    /// Builds a `TEARDOWN` request (valid in any non-Init state, and Init).
194    pub fn teardown(&mut self, uri: &str) -> Result<Vec<u8>> {
195        self.build_request(Method::Teardown, uri, None, &[])
196    }
197
198    /// Builds a `GET_PARAMETER` request, optionally with a body (state-neutral;
199    /// an empty body is the liveness ping).
200    pub fn get_parameter(&mut self, uri: &str, body: &[u8]) -> Result<Vec<u8>> {
201        self.build_request_with_body(Method::GetParameter, uri, body, &[])
202    }
203
204    fn build_request(
205        &mut self,
206        method: Method,
207        uri: &str,
208        _range: Option<&str>,
209        extra: &[(headers::HeaderName, String)],
210    ) -> Result<Vec<u8>> {
211        self.build_request_with_body(method, uri, &[], extra)
212    }
213
214    fn build_request_with_body(
215        &mut self,
216        method: Method,
217        uri: &str,
218        body: &[u8],
219        extra: &[(headers::HeaderName, String)],
220    ) -> Result<Vec<u8>> {
221        // Reject methods not valid in the current state (state-neutral pass).
222        client_next_state(self.state, &method)?;
223
224        let cseq = self.next_cseq;
225        let request = self.assemble(method.clone(), uri, cseq, body, extra)?;
226        let bytes = serialize(&Message::from(request.clone()))?;
227        self.next_cseq += 1;
228        self.pending.insert(
229            cseq,
230            Pending {
231                method,
232                uri: uri.to_string(),
233                request,
234                auth_retried: false,
235            },
236        );
237        Ok(bytes)
238    }
239
240    /// Assembles a `Request` with CSeq, User-Agent, Session (if known),
241    /// Authorization (if authenticated), any extra headers, and the body.
242    fn assemble(
243        &mut self,
244        method: Method,
245        uri: &str,
246        cseq: u32,
247        body: &[u8],
248        extra: &[(headers::HeaderName, String)],
249    ) -> Result<Request<Body>> {
250        let url = rtsp_types::Url::parse(uri)
251            .map_err(|e| Error::TransportParse(format!("invalid request URI {uri:?}: {e}")))?;
252        let mut builder = Request::builder(method.clone(), Version::V1_0)
253            .request_uri(url)
254            .header(headers::CSEQ, cseq.to_string())
255            .header(headers::USER_AGENT, self.user_agent.clone());
256        if let Some(sid) = &self.session_id {
257            builder = builder.header(headers::SESSION, sid.clone());
258        }
259        for (name, value) in extra {
260            builder = builder.header(name.clone(), value.clone());
261        }
262        if let Some(auth) = &mut self.authenticator {
263            let value = auth.authorization(<&str>::from(&method), uri)?;
264            builder = builder.header(headers::AUTHORIZATION, value);
265        }
266        let request = if body.is_empty() {
267            builder.build(Vec::new())
268        } else {
269            builder.build(body.to_vec())
270        };
271        Ok(request)
272    }
273
274    // --- Inbound handling -------------------------------------------------
275
276    /// Feeds inbound bytes and returns the events produced. Retains any partial
277    /// trailing message or frame internally for the next call.
278    pub fn handle_data(&mut self, data: &[u8]) -> Result<Vec<ClientEvent>> {
279        self.inbound.extend_from_slice(data);
280        let mut events = Vec::new();
281
282        loop {
283            if self.inbound.is_empty() {
284                break;
285            }
286            if self.inbound[0] == MAGIC {
287                // Interleaved frame path.
288                match interleaved::InterleavedFrame::parse(&self.inbound)? {
289                    Some((frame, consumed)) => {
290                        events.push(ClientEvent::MediaData {
291                            channel: frame.channel,
292                            data: frame.payload,
293                        });
294                        self.inbound.drain(..consumed);
295                    }
296                    None => break, // need more bytes
297                }
298                continue;
299            }
300
301            // RTSP message path.
302            match Message::<Body>::parse(&self.inbound) {
303                Ok((message, consumed)) => {
304                    self.inbound.drain(..consumed);
305                    self.process_message(message, &mut events)?;
306                }
307                Err(rtsp_types::ParseError::Incomplete(_)) => break,
308                Err(rtsp_types::ParseError::Error) => {
309                    return Err(Error::MessageParse("malformed RTSP message".into()));
310                }
311            }
312        }
313        Ok(events)
314    }
315
316    fn process_message(
317        &mut self,
318        message: Message<Body>,
319        events: &mut Vec<ClientEvent>,
320    ) -> Result<()> {
321        match message {
322            Message::Response(response) => {
323                let cseq = header_value(response.header(&headers::CSEQ))
324                    .and_then(|s| s.trim().parse::<u32>().ok())
325                    .ok_or(Error::MissingCSeq)?;
326                let status = response.status();
327
328                // 401: attempt a transparent auth retry.
329                if status == StatusCode::Unauthorized {
330                    if let Some(retry) = self.try_auth_retry(cseq, &response)? {
331                        events.push(retry);
332                        return Ok(());
333                    }
334                }
335
336                let pending = self.pending.remove(&cseq).ok_or(Error::UnknownCSeq(cseq))?;
337
338                // Capture Session id + timeout (typically from SETUP).
339                if let Some(session_hdr) = header_value(response.header(&headers::SESSION)) {
340                    let (id, timeout) = parse_session(session_hdr);
341                    self.session_id = Some(id);
342                    if timeout.is_some() {
343                        self.session_timeout = timeout;
344                    }
345                }
346                // Capture negotiated transport from SETUP response.
347                if pending.method == Method::Setup {
348                    if let Some(t) = header_value(response.header(&headers::TRANSPORT)) {
349                        self.negotiated_transport = Some(Transport::parse(t)?);
350                    }
351                }
352
353                // State transition.
354                if status.is_success() {
355                    self.state = client_next_state(self.state, &pending.method)?;
356                    // TEARDOWN invalidates the session.
357                    if pending.method == Method::Teardown {
358                        self.session_id = None;
359                        self.session_timeout = None;
360                        self.authenticator = None;
361                    }
362                } else if status.is_redirection() {
363                    self.state = SessionState::Init;
364                }
365                // 4xx (other than the handled 401) / 5xx: no state change.
366
367                events.push(ClientEvent::Response {
368                    cseq,
369                    method: pending.method,
370                    status,
371                    body: response.into_body(),
372                });
373                Ok(())
374            }
375            Message::Data(data) => {
376                events.push(ClientEvent::MediaData {
377                    channel: data.channel_id(),
378                    data: data.into_body(),
379                });
380                Ok(())
381            }
382            Message::Request(_) => {
383                // Server-initiated requests (e.g. S->C OPTIONS, REDIRECT,
384                // ANNOUNCE) are out of scope for this round; ignore.
385                Ok(())
386            }
387        }
388    }
389
390    /// On a 401, build/refresh the authenticator from `WWW-Authenticate` and
391    /// re-send the pending request with an `Authorization` header, unless a
392    /// retry was already attempted (wrong credentials) or none are configured.
393    fn try_auth_retry(
394        &mut self,
395        cseq: u32,
396        response: &rtsp_types::Response<Body>,
397    ) -> Result<Option<ClientEvent>> {
398        let creds = match &self.credentials {
399            Some(c) => c.clone(),
400            None => return Ok(None),
401        };
402        // Only retry if the original request is still pending and hasn't retried.
403        let (method, uri, already) = match self.pending.get(&cseq) {
404            Some(p) => (p.method.clone(), p.uri.clone(), p.auth_retried),
405            None => return Ok(None),
406        };
407
408        let challenge = header_value(response.header(&headers::WWW_AUTHENTICATE))
409            .ok_or_else(|| Error::Auth("401 without WWW-Authenticate".into()))?;
410        let stale = challenge.to_ascii_lowercase().contains("stale=true");
411
412        // Fresh challenge => (re)build the authenticator. On stale=true this
413        // picks up the new nonce; on first 401 it establishes the client.
414        if self.authenticator.is_none() || already || stale {
415            self.authenticator = Some(Authenticator::from_challenge(challenge, creds)?);
416        }
417        // Guard: if we already retried and it isn't a stale refresh, give up so
418        // the caller sees the 401 (wrong credentials).
419        if already && !stale {
420            return Ok(None);
421        }
422
423        // Preserve method-specific headers (Accept/Transport/Range) before we
424        // drop the old pending entry, then issue a new request with a fresh CSeq.
425        let extra = self.replay_extra(&method, cseq);
426        self.pending.remove(&cseq);
427        let new_cseq = self.next_cseq;
428        let request = self.assemble(method.clone(), &uri, new_cseq, &[], &extra)?;
429        let bytes = serialize(&Message::from(request.clone()))?;
430        self.next_cseq += 1;
431        self.pending.insert(
432            new_cseq,
433            Pending {
434                method: method.clone(),
435                uri,
436                request,
437                auth_retried: true,
438            },
439        );
440        Ok(Some(ClientEvent::AuthRetry {
441            method,
442            cseq: new_cseq,
443            request: bytes,
444        }))
445    }
446
447    /// Re-derive method-specific headers (e.g. Accept/Transport) for an auth
448    /// replay from the previously-sent request.
449    fn replay_extra(&self, _method: &Method, old_cseq: u32) -> Vec<(headers::HeaderName, String)> {
450        let mut extra = Vec::new();
451        if let Some(p) = self.pending.get(&old_cseq) {
452            for name in [headers::ACCEPT, headers::TRANSPORT, headers::RANGE] {
453                if let Some(v) = header_value(p.request.header(&name)) {
454                    extra.push((name, v.to_string()));
455                }
456            }
457        }
458        extra
459    }
460}
461
462/// Extracts the string value of an optional header.
463fn header_value(h: Option<&headers::HeaderValue>) -> Option<&str> {
464    h.map(|v| v.as_str())
465}
466
467/// Parses a `Session` header value into (id, optional timeout seconds).
468fn parse_session(value: &str) -> (String, Option<u64>) {
469    let mut parts = value.split(';').map(str::trim);
470    let id = parts.next().unwrap_or("").to_string();
471    let timeout = value
472        .split(';')
473        .filter_map(|s| s.trim().strip_prefix("timeout="))
474        .find_map(|s| s.trim().parse::<u64>().ok());
475    (id, timeout)
476}
477
478/// Serializes an RTSP message to bytes.
479fn serialize(message: &Message<Body>) -> Result<Vec<u8>> {
480    let mut out = Vec::new();
481    message
482        .write(&mut out)
483        .map_err(|e| Error::MessageWrite(e.to_string()))?;
484    Ok(out)
485}
486
487#[cfg(test)]
488mod tests {
489    use super::*;
490
491    #[test]
492    fn play_in_init_bites() {
493        let mut c = ClientSession::new();
494        assert!(c.play("rtsp://h/s").is_err());
495    }
496
497    #[test]
498    fn setup_allowed_in_init() {
499        let mut c = ClientSession::new();
500        let t = Transport::single(crate::transport::TransportSpec::rtp_avp_tcp_interleaved(
501            0, 1,
502        ));
503        assert!(c.setup("rtsp://h/s", &t).is_ok());
504    }
505
506    #[test]
507    fn cseq_increments() {
508        let mut c = ClientSession::new();
509        let a = c.options("rtsp://h/s").unwrap();
510        let b = c.describe("rtsp://h/s").unwrap();
511        assert!(String::from_utf8_lossy(&a).contains("CSeq: 1"));
512        assert!(String::from_utf8_lossy(&b).contains("CSeq: 2"));
513    }
514}
rtsp_runtime/client.rs

rtsp_runtime/
client.rs
