arcly_stream/auth.rs
1//! Stream-level authentication and authorization.
2//!
3//! Like [`Observer`](crate::Observer), authorization is an **injected trait with
4//! a permit-all default** — the engine never bakes in an identity scheme. A host
5//! supplies a [`StreamAuthenticator`] (validating stream keys, signed tokens,
6//! IP allow-lists, an external auth service, …) and the engine enforces it on
7//! the publish and play paths.
8//!
9//! ```no_run
10//! use arcly_stream::auth::{Credentials, StreamAuthenticator};
11//! use arcly_stream::prelude::*;
12//! use std::sync::Arc;
13//!
14//! struct KeyAuth { secret: String }
15//!
16//! #[async_trait]
17//! impl StreamAuthenticator for KeyAuth {
18//! async fn authorize_publish(&self, _key: &StreamKey, creds: &Credentials) -> Result<()> {
19//! match creds.token.as_deref() {
20//! Some(t) if t == self.secret => Ok(()),
21//! _ => Err(StreamError::Unauthorized("bad publish key".into())),
22//! }
23//! }
24//! }
25//!
26//! let engine = Engine::builder()
27//! .application(AppSpec::new("live"))
28//! .authenticator(KeyAuth { secret: "s3cr3t".into() })
29//! .build();
30//! # let _ = engine;
31//! ```
32
33use crate::{Result, StreamKey};
34use async_trait::async_trait;
35use std::net::SocketAddr;
36use std::sync::Arc;
37
38/// Decides whether a **playback** (egress/pull) request for a stream is allowed.
39///
40/// Server-side egress handlers (RTSP `PLAY`, SRT `m=request`, RTMP `play`) consult
41/// this before serving a player. The closure receives the stream key, the
42/// presented play token (`None` = none presented), and the player's peer address
43/// (`None` when the transport exposes none) — so the host can check the per-app
44/// egress toggle, verify the token, and bind the token to the client IP. Returns
45/// a future resolving to `true` to allow. Async so the host can load per-app
46/// settings from a database. When no gate is installed, egress is open
47/// (permit-all), matching the authenticator's default-permit philosophy.
48pub type EgressGate = Arc<
49 dyn Fn(
50 StreamKey,
51 Option<String>,
52 Option<SocketAddr>,
53 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>>
54 + Send
55 + Sync,
56>;
57
58/// Extract a `token=…` value from a query string, ignoring an empty value.
59///
60/// Accepts the full URI or stream name — anything before a `?` is dropped, then
61/// the remainder is parsed as `&`-separated `key=value` pairs. Returns the first
62/// non-empty `token` parameter. Shared by the protocol egress/ingest handlers
63/// (RTSP URI, RTMP stream name) that carry their credential in the URL query.
64pub fn token_from_query(s: &str) -> Option<String> {
65 let query = s.split_once('?').map(|(_, q)| q).unwrap_or(s);
66 query.split('&').find_map(|kv| {
67 kv.strip_prefix("token=")
68 .filter(|t| !t.is_empty())
69 .map(|t| t.to_string())
70 })
71}
72
73/// Credentials presented by a connecting publisher or player.
74///
75/// Protocol handlers populate whichever fields their transport carries (an RTMP
76/// stream key in `token`, a WHIP bearer in `token`, the peer address in `addr`,
77/// query parameters in `params`).
78#[derive(Debug, Default, Clone)]
79pub struct Credentials {
80 /// Opaque secret: a stream key, signed URL token, or bearer credential.
81 pub token: Option<String>,
82 /// Remote peer address, when the transport exposes one.
83 pub addr: Option<SocketAddr>,
84 /// Arbitrary protocol-supplied key/value parameters (e.g. query string).
85 pub params: Vec<(String, String)>,
86}
87
88impl Credentials {
89 /// Credentials carrying only a token (the common stream-key case).
90 pub fn token(token: impl Into<String>) -> Self {
91 Self {
92 token: Some(token.into()),
93 ..Self::default()
94 }
95 }
96
97 /// Look up a protocol parameter by key.
98 pub fn param(&self, key: &str) -> Option<&str> {
99 self.params
100 .iter()
101 .find(|(k, _)| k == key)
102 .map(|(_, v)| v.as_str())
103 }
104}
105
106/// Authorizes publish and play attempts. Both methods **default to permit**, so
107/// an implementor overrides only the side it gates.
108#[async_trait]
109pub trait StreamAuthenticator: Send + Sync + 'static {
110 /// Decide whether `creds` may publish to `key`. Return
111 /// [`StreamError::Unauthorized`](crate::StreamError::Unauthorized) to reject.
112 async fn authorize_publish(&self, _key: &StreamKey, _creds: &Credentials) -> Result<()> {
113 Ok(())
114 }
115
116 /// Decide whether `creds` may subscribe to `key`.
117 async fn authorize_play(&self, _key: &StreamKey, _creds: &Credentials) -> Result<()> {
118 Ok(())
119 }
120}
121
122/// The default authenticator: permits everything. Selected when the builder is
123/// given none, preserving the engine's zero-policy default.
124#[derive(Debug, Default, Clone, Copy)]
125pub struct AllowAll;
126
127impl StreamAuthenticator for AllowAll {}
128
129/// A production-ready, time-limited **signed-token** authenticator
130/// (`feature = "auth-token"`).
131///
132/// A token binds a [`StreamKey`] to an expiry under an HMAC-SHA-256 signature,
133/// so it cannot be forged without the shared secret nor replayed past its
134/// deadline. The wire form is:
135///
136/// ```text
137/// <expiry_unix_seconds>:<hex(hmac_sha256(secret, "app/stream:expiry"))>
138/// ```
139///
140/// Mint tokens out-of-band (e.g. in your sign-in / "get publish URL" endpoint)
141/// with [`sign`](Self::sign); the engine verifies them on the publish path —
142/// and, when [`gate_playback`](Self::gate_playback) is set, the play path too.
143/// Verification is constant-time and pulls in **no crypto dependency** (the
144/// HMAC is a small, test-vector-checked in-crate implementation).
145///
146/// ```
147/// use arcly_stream::auth::TokenAuthenticator;
148/// use arcly_stream::StreamKey;
149///
150/// let auth = TokenAuthenticator::new("super-secret");
151/// let key = StreamKey::new("live", "cam-1");
152/// // Mint a token valid until some absolute Unix time:
153/// let token = auth.sign(&key, 9_999_999_999);
154/// assert!(auth.verify(&key, &token).is_ok());
155/// // A token for a different stream is rejected:
156/// assert!(auth.verify(&StreamKey::new("live", "other"), &token).is_err());
157/// ```
158#[cfg(feature = "auth-token")]
159#[cfg_attr(docsrs, doc(cfg(feature = "auth-token")))]
160pub struct TokenAuthenticator {
161 secret: Vec<u8>,
162 gate_play: bool,
163}
164
165#[cfg(feature = "auth-token")]
166impl TokenAuthenticator {
167 /// New authenticator keyed by `secret`. Gates **publish** only by default;
168 /// call [`gate_playback`](Self::gate_playback) to gate play as well.
169 pub fn new(secret: impl Into<Vec<u8>>) -> Self {
170 Self {
171 secret: secret.into(),
172 gate_play: false,
173 }
174 }
175
176 /// Also require a valid token to *play* (subscribe), not just publish.
177 pub fn gate_playback(mut self, gate: bool) -> Self {
178 self.gate_play = gate;
179 self
180 }
181
182 /// The signed message a token covers: `"app/stream:expiry"`.
183 fn message(key: &StreamKey, expiry: u64) -> String {
184 format!("{}/{}:{}", key.app.as_str(), key.stream_id.as_str(), expiry)
185 }
186
187 /// Mint a token authorizing `key` until `expires_at` (Unix seconds).
188 pub fn sign(&self, key: &StreamKey, expires_at: u64) -> String {
189 let mac =
190 crate::crypto::hmac_sha256(&self.secret, Self::message(key, expires_at).as_bytes());
191 format!("{}:{}", expires_at, crate::crypto::to_hex(&mac))
192 }
193
194 /// Verify `token` against `key` and the current wall clock. Returns
195 /// [`StreamError::Unauthorized`](crate::StreamError::Unauthorized) on a
196 /// malformed, expired, or mis-signed token.
197 pub fn verify(&self, key: &StreamKey, token: &str) -> Result<()> {
198 let (exp_str, sig) = token
199 .split_once(':')
200 .ok_or_else(|| crate::StreamError::Unauthorized("malformed token".into()))?;
201 let expiry: u64 = exp_str
202 .parse()
203 .map_err(|_| crate::StreamError::Unauthorized("malformed token expiry".into()))?;
204 if now_unix_secs() > expiry {
205 return Err(crate::StreamError::Unauthorized("token expired".into()));
206 }
207 let expected = crate::crypto::to_hex(&crate::crypto::hmac_sha256(
208 &self.secret,
209 Self::message(key, expiry).as_bytes(),
210 ));
211 if crate::crypto::constant_time_eq(expected.as_bytes(), sig.as_bytes()) {
212 Ok(())
213 } else {
214 Err(crate::StreamError::Unauthorized(
215 "invalid token signature".into(),
216 ))
217 }
218 }
219
220 fn token_for(creds: &Credentials) -> Result<&str> {
221 creds
222 .token
223 .as_deref()
224 .ok_or_else(|| crate::StreamError::Unauthorized("missing token".into()))
225 }
226}
227
228/// Seconds since the Unix epoch, saturating to 0 before 1970.
229#[cfg(feature = "auth-token")]
230fn now_unix_secs() -> u64 {
231 std::time::SystemTime::now()
232 .duration_since(std::time::UNIX_EPOCH)
233 .map(|d| d.as_secs())
234 .unwrap_or(0)
235}
236
237#[cfg(feature = "auth-token")]
238#[async_trait]
239impl StreamAuthenticator for TokenAuthenticator {
240 async fn authorize_publish(&self, key: &StreamKey, creds: &Credentials) -> Result<()> {
241 self.verify(key, Self::token_for(creds)?)
242 }
243
244 async fn authorize_play(&self, key: &StreamKey, creds: &Credentials) -> Result<()> {
245 if self.gate_play {
246 self.verify(key, Self::token_for(creds)?)
247 } else {
248 Ok(())
249 }
250 }
251}
252
253#[cfg(test)]
254mod query_tests {
255 use super::token_from_query;
256
257 #[test]
258 fn extracts_token_among_params() {
259 assert_eq!(
260 token_from_query("rtsp://h/live/cam?a=1&token=secret&b=2"),
261 Some("secret".to_string())
262 );
263 }
264
265 #[test]
266 fn token_only_without_path() {
267 assert_eq!(token_from_query("token=xyz"), Some("xyz".to_string()));
268 }
269
270 #[test]
271 fn absent_token_is_none() {
272 assert_eq!(token_from_query("rtmp://h/live/cam"), None);
273 assert_eq!(token_from_query("a=1&b=2"), None);
274 }
275
276 #[test]
277 fn empty_token_is_none() {
278 assert_eq!(token_from_query("live/cam?token="), None);
279 assert_eq!(token_from_query("live/cam?token=&x=1"), None);
280 }
281}
282
283#[cfg(all(test, feature = "auth-token"))]
284mod token_tests {
285 use super::*;
286
287 #[tokio::test]
288 async fn signed_token_authorizes_its_stream() {
289 let auth = TokenAuthenticator::new("s3cr3t");
290 let key = StreamKey::new("live", "cam");
291 let token = auth.sign(&key, now_unix_secs() + 3600);
292 let creds = Credentials::token(token);
293 assert!(auth.authorize_publish(&key, &creds).await.is_ok());
294 // Play is open unless gated.
295 assert!(auth
296 .authorize_play(&key, &Credentials::default())
297 .await
298 .is_ok());
299 }
300
301 #[tokio::test]
302 async fn rejects_wrong_stream_secret_and_expiry() {
303 let auth = TokenAuthenticator::new("s3cr3t");
304 let key = StreamKey::new("live", "cam");
305 let token = auth.sign(&key, now_unix_secs() + 3600);
306
307 // Wrong stream.
308 assert!(auth
309 .verify(&StreamKey::new("live", "other"), &token)
310 .is_err());
311 // Wrong secret.
312 let other = TokenAuthenticator::new("different");
313 assert!(other.verify(&key, &token).is_err());
314 // Expired.
315 let stale = auth.sign(&key, now_unix_secs().saturating_sub(1));
316 assert!(auth.verify(&key, &stale).is_err());
317 // Malformed.
318 assert!(auth.verify(&key, "not-a-token").is_err());
319 }
320
321 #[tokio::test]
322 async fn gated_playback_requires_a_token() {
323 let auth = TokenAuthenticator::new("s3cr3t").gate_playback(true);
324 let key = StreamKey::new("live", "cam");
325 assert!(auth
326 .authorize_play(&key, &Credentials::default())
327 .await
328 .is_err());
329 let token = auth.sign(&key, now_unix_secs() + 60);
330 assert!(auth
331 .authorize_play(&key, &Credentials::token(token))
332 .await
333 .is_ok());
334 }
335}