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::{Message, Method, Request, StatusCode, Version, headers};
28
29use crate::auth::{Authenticator, Credentials, RequestContext};
30use crate::error::{Error, Result};
31use crate::interleaved::{self, MAGIC};
32use crate::state::{SessionState, client_next_state};
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 an `ANNOUNCE` request carrying an SDP body (RFC 2326 §10.3).
199    pub fn announce(&mut self, uri: &str, sdp: &str) -> Result<Vec<u8>> {
200        self.build_request_with_body(
201            Method::Announce,
202            uri,
203            sdp.as_bytes(),
204            &[(headers::CONTENT_TYPE, "application/sdp".to_string())],
205        )
206    }
207
208    /// Builds a `RECORD` request (RFC 2326 §10.11, valid in Ready).
209    pub fn record(&mut self, uri: &str) -> Result<Vec<u8>> {
210        self.build_request(Method::Record, uri, None, &[])
211    }
212
213    /// Builds a `GET_PARAMETER` request, optionally with a body (state-neutral;
214    /// an empty body is the liveness ping).
215    pub fn get_parameter(&mut self, uri: &str, body: &[u8]) -> Result<Vec<u8>> {
216        self.build_request_with_body(Method::GetParameter, uri, body, &[])
217    }
218
219    fn build_request(
220        &mut self,
221        method: Method,
222        uri: &str,
223        _range: Option<&str>,
224        extra: &[(headers::HeaderName, String)],
225    ) -> Result<Vec<u8>> {
226        self.build_request_with_body(method, uri, &[], extra)
227    }
228
229    fn build_request_with_body(
230        &mut self,
231        method: Method,
232        uri: &str,
233        body: &[u8],
234        extra: &[(headers::HeaderName, String)],
235    ) -> Result<Vec<u8>> {
236        // Reject methods not valid in the current state (state-neutral pass).
237        client_next_state(self.state, &method)?;
238
239        let cseq = self.next_cseq;
240        let request = self.assemble(method.clone(), uri, cseq, body, extra)?;
241        let bytes = serialize(&Message::from(request.clone()))?;
242        self.next_cseq += 1;
243        self.pending.insert(
244            cseq,
245            Pending {
246                method,
247                uri: uri.to_string(),
248                request,
249                auth_retried: false,
250            },
251        );
252        Ok(bytes)
253    }
254
255    /// Assembles a `Request` with CSeq, User-Agent, Session (if known),
256    /// Authorization (if authenticated), any extra headers, and the body.
257    fn assemble(
258        &mut self,
259        method: Method,
260        uri: &str,
261        cseq: u32,
262        body: &[u8],
263        extra: &[(headers::HeaderName, String)],
264    ) -> Result<Request<Body>> {
265        let url = rtsp_types::Url::parse(uri)
266            .map_err(|e| Error::TransportParse(format!("invalid request URI {uri:?}: {e}")))?;
267        let mut builder = Request::builder(method.clone(), Version::V1_0)
268            .request_uri(url)
269            .header(headers::CSEQ, cseq.to_string())
270            .header(headers::USER_AGENT, self.user_agent.clone());
271        if let Some(sid) = &self.session_id {
272            builder = builder.header(headers::SESSION, sid.clone());
273        }
274        for (name, value) in extra {
275            builder = builder.header(name.clone(), value.clone());
276        }
277        if let Some(auth) = &mut self.authenticator {
278            let ctx = RequestContext::new(<&str>::from(&method), uri);
279            let value = auth.authorization(&ctx)?;
280            builder = builder.header(headers::AUTHORIZATION, value);
281        }
282        let request = if body.is_empty() {
283            builder.build(Vec::new())
284        } else {
285            builder.build(body.to_vec())
286        };
287        Ok(request)
288    }
289
290    // --- Inbound handling -------------------------------------------------
291
292    /// Feeds inbound bytes and returns the events produced. Retains any partial
293    /// trailing message or frame internally for the next call.
294    pub fn handle_data(&mut self, data: &[u8]) -> Result<Vec<ClientEvent>> {
295        self.inbound.extend_from_slice(data);
296        let mut events = Vec::new();
297
298        loop {
299            if self.inbound.is_empty() {
300                break;
301            }
302            if self.inbound[0] == MAGIC {
303                // Interleaved frame path.
304                match interleaved::InterleavedFrame::parse(&self.inbound)? {
305                    Some((frame, consumed)) => {
306                        events.push(ClientEvent::MediaData {
307                            channel: frame.channel,
308                            data: frame.payload,
309                        });
310                        self.inbound.drain(..consumed);
311                    }
312                    None => break, // need more bytes
313                }
314                continue;
315            }
316
317            // RTSP message path.
318            match Message::<Body>::parse(&self.inbound) {
319                Ok((message, consumed)) => {
320                    self.inbound.drain(..consumed);
321                    self.process_message(message, &mut events)?;
322                }
323                Err(rtsp_types::ParseError::Incomplete(_)) => break,
324                Err(rtsp_types::ParseError::Error) => {
325                    return Err(Error::MessageParse("malformed RTSP message".into()));
326                }
327            }
328        }
329        Ok(events)
330    }
331
332    fn process_message(
333        &mut self,
334        message: Message<Body>,
335        events: &mut Vec<ClientEvent>,
336    ) -> Result<()> {
337        match message {
338            Message::Response(response) => {
339                let cseq = header_value(response.header(&headers::CSEQ))
340                    .and_then(|s| s.trim().parse::<u32>().ok())
341                    .ok_or(Error::MissingCSeq)?;
342                let status = response.status();
343
344                // 401: attempt a transparent auth retry.
345                if status == StatusCode::Unauthorized {
346                    if let Some(retry) = self.try_auth_retry(cseq, &response)? {
347                        events.push(retry);
348                        return Ok(());
349                    }
350                }
351
352                let pending = self.pending.remove(&cseq).ok_or(Error::UnknownCSeq(cseq))?;
353
354                // Capture Session id + timeout (typically from SETUP).
355                if let Some(session_hdr) = header_value(response.header(&headers::SESSION)) {
356                    let (id, timeout) = parse_session(session_hdr);
357                    self.session_id = Some(id);
358                    if timeout.is_some() {
359                        self.session_timeout = timeout;
360                    }
361                }
362                // Capture negotiated transport from SETUP response.
363                if pending.method == Method::Setup {
364                    if let Some(t) = header_value(response.header(&headers::TRANSPORT)) {
365                        self.negotiated_transport = Some(Transport::parse(t)?);
366                    }
367                }
368
369                // State transition.
370                if status.is_success() {
371                    self.state = client_next_state(self.state, &pending.method)?;
372                    // TEARDOWN invalidates the session.
373                    if pending.method == Method::Teardown {
374                        self.session_id = None;
375                        self.session_timeout = None;
376                        self.authenticator = None;
377                    }
378                } else if status.is_redirection() {
379                    self.state = SessionState::Init;
380                }
381                // 4xx (other than the handled 401) / 5xx: no state change.
382
383                events.push(ClientEvent::Response {
384                    cseq,
385                    method: pending.method,
386                    status,
387                    body: response.into_body(),
388                });
389                Ok(())
390            }
391            Message::Data(data) => {
392                events.push(ClientEvent::MediaData {
393                    channel: data.channel_id(),
394                    data: data.into_body(),
395                });
396                Ok(())
397            }
398            Message::Request(_) => {
399                // Server-initiated requests (e.g. S->C OPTIONS, REDIRECT,
400                // ANNOUNCE) are out of scope for this round; ignore.
401                Ok(())
402            }
403        }
404    }
405
406    /// On a 401, build/refresh the authenticator from `WWW-Authenticate` and
407    /// re-send the pending request with an `Authorization` header, unless a
408    /// retry was already attempted (wrong credentials) or none are configured.
409    fn try_auth_retry(
410        &mut self,
411        cseq: u32,
412        response: &rtsp_types::Response<Body>,
413    ) -> Result<Option<ClientEvent>> {
414        let creds = match &self.credentials {
415            Some(c) => c.clone(),
416            None => return Ok(None),
417        };
418        // Only retry if the original request is still pending and hasn't retried.
419        let (method, uri, already) = match self.pending.get(&cseq) {
420            Some(p) => (p.method.clone(), p.uri.clone(), p.auth_retried),
421            None => return Ok(None),
422        };
423
424        let challenge = header_value(response.header(&headers::WWW_AUTHENTICATE))
425            .ok_or_else(|| Error::Auth("401 without WWW-Authenticate".into()))?;
426        let stale = challenge.to_ascii_lowercase().contains("stale=true");
427
428        // Fresh challenge => (re)build the authenticator. On stale=true this
429        // picks up the new nonce; on first 401 it establishes the client.
430        if self.authenticator.is_none() || already || stale {
431            self.authenticator = Some(Authenticator::from_challenge(challenge, creds)?);
432        }
433        // Guard: if we already retried and it isn't a stale refresh, give up so
434        // the caller sees the 401 (wrong credentials).
435        if already && !stale {
436            return Ok(None);
437        }
438
439        // Preserve method-specific headers (Accept/Transport/Range) before we
440        // drop the old pending entry, then issue a new request with a fresh CSeq.
441        let extra = self.replay_extra(&method, cseq);
442        self.pending.remove(&cseq);
443        let new_cseq = self.next_cseq;
444        let request = self.assemble(method.clone(), &uri, new_cseq, &[], &extra)?;
445        let bytes = serialize(&Message::from(request.clone()))?;
446        self.next_cseq += 1;
447        self.pending.insert(
448            new_cseq,
449            Pending {
450                method: method.clone(),
451                uri,
452                request,
453                auth_retried: true,
454            },
455        );
456        Ok(Some(ClientEvent::AuthRetry {
457            method,
458            cseq: new_cseq,
459            request: bytes,
460        }))
461    }
462
463    /// Re-derive method-specific headers (e.g. Accept/Transport) for an auth
464    /// replay from the previously-sent request.
465    fn replay_extra(&self, _method: &Method, old_cseq: u32) -> Vec<(headers::HeaderName, String)> {
466        let mut extra = Vec::new();
467        if let Some(p) = self.pending.get(&old_cseq) {
468            for name in [headers::ACCEPT, headers::TRANSPORT, headers::RANGE] {
469                if let Some(v) = header_value(p.request.header(&name)) {
470                    extra.push((name, v.to_string()));
471                }
472            }
473        }
474        extra
475    }
476}
477
478/// Extracts the string value of an optional header.
479fn header_value(h: Option<&headers::HeaderValue>) -> Option<&str> {
480    h.map(|v| v.as_str())
481}
482
483/// Parses a `Session` header value into (id, optional timeout seconds).
484fn parse_session(value: &str) -> (String, Option<u64>) {
485    let mut parts = value.split(';').map(str::trim);
486    let id = parts.next().unwrap_or("").to_string();
487    let timeout = value
488        .split(';')
489        .filter_map(|s| s.trim().strip_prefix("timeout="))
490        .find_map(|s| s.trim().parse::<u64>().ok());
491    (id, timeout)
492}
493
494/// Serializes an RTSP message to bytes.
495fn serialize(message: &Message<Body>) -> Result<Vec<u8>> {
496    let mut out = Vec::new();
497    message
498        .write(&mut out)
499        .map_err(|e| Error::MessageWrite(e.to_string()))?;
500    Ok(out)
501}
502
503#[cfg(test)]
504mod tests {
505    use super::*;
506
507    #[test]
508    fn play_in_init_bites() {
509        let mut c = ClientSession::new();
510        assert!(c.play("rtsp://h/s").is_err());
511    }
512
513    #[test]
514    fn setup_allowed_in_init() {
515        let mut c = ClientSession::new();
516        let t = Transport::single(crate::transport::TransportSpec::rtp_avp_tcp_interleaved(
517            0, 1,
518        ));
519        assert!(c.setup("rtsp://h/s", &t).is_ok());
520    }
521
522    #[test]
523    fn cseq_increments() {
524        let mut c = ClientSession::new();
525        let a = c.options("rtsp://h/s").unwrap();
526        let b = c.describe("rtsp://h/s").unwrap();
527        assert!(String::from_utf8_lossy(&a).contains("CSeq: 1"));
528        assert!(String::from_utf8_lossy(&b).contains("CSeq: 2"));
529    }
530
531    #[test]
532    fn announce_emits_request_with_sdp_content_type() {
533        let mut c = ClientSession::new();
534        let sdp = "v=0\r\no=- 0 0 IN IP4 0.0.0.0\r\ns=Test\r\n";
535        let bytes = c.announce("rtsp://h/s", sdp).unwrap();
536        let s = String::from_utf8_lossy(&bytes);
537        assert!(s.contains("ANNOUNCE rtsp://h/s"));
538        assert!(s.contains("Content-Type: application/sdp"));
539        assert!(s.contains(sdp));
540    }
541
542    #[test]
543    fn record_fails_from_init_state() {
544        let mut c = ClientSession::new();
545        assert!(c.record("rtsp://h/s").is_err());
546    }
547
548    #[test]
549    fn record_transitions_to_recording_state() {
550        // RTSP lines are CRLF-terminated (RFC 2326 §1); convert "\n" -> CRLF.
551        fn wire(s: &str) -> Vec<u8> {
552            s.replace('\n', "\r\n").into_bytes()
553        }
554        let mut c = ClientSession::new();
555        // Drive Init -> Ready via a successful SETUP round-trip.
556        let t = Transport::single(crate::transport::TransportSpec::rtp_avp_tcp_interleaved(
557            0, 1,
558        ));
559        c.setup("rtsp://h/s", &t).unwrap();
560        c.handle_data(&wire(
561            "RTSP/1.0 200 OK\nCSeq: 1\nSession: 42\nTransport: RTP/AVP/TCP;interleaved=0-1\n\n",
562        ))
563        .unwrap();
564        assert_eq!(c.state(), SessionState::Ready);
565
566        // RECORD is valid from Ready; the request alone does not move state.
567        c.record("rtsp://h/s").unwrap();
568        assert_eq!(c.state(), SessionState::Ready);
569        // A 2xx response advances Ready -> Recording.
570        c.handle_data(&wire("RTSP/1.0 200 OK\nCSeq: 2\nSession: 42\n\n"))
571            .unwrap();
572        assert_eq!(c.state(), SessionState::Recording);
573    }
574
575    // Security-blocker regression (pre-release audit): `ClientSession`
576    // derives `Debug` and embeds `Option<Credentials>` directly — it must
577    // inherit `Credentials`'s redacting `Debug`, never the raw secret.
578    #[test]
579    fn client_session_debug_does_not_leak_embedded_credentials_secret() {
580        let c = ClientSession::new()
581            .with_credentials(Credentials::new("admin", "extremely-secret-password"));
582        let debug = format!("{c:?}");
583        assert!(
584            !debug.contains("extremely-secret-password"),
585            "leaked via ClientSession Debug: {debug}"
586        );
587        assert!(debug.contains("***"), "expected redaction marker: {debug}");
588    }
589}
rtsp_runtime/client.rs

rtsp_runtime/
client.rs
