o_sfu/runtime/http_server/
extractors.rs

1use std::{net::SocketAddr, str, sync::Arc};
2
3use axum::{
4    body::Bytes,
5    extract::{FromRef, FromRequest, FromRequestParts, Query, Request},
6    http::{HeaderMap, StatusCode, header, request::Parts},
7    response::{IntoResponse, Response},
8};
9pub use o_sfu_rfc::jwt::RegisteredJwtClaims;
10
11use crate::runtime::{
12    MediaTransport, RuntimeMetrics, RuntimeState,
13    auth::{self, HttpDisconnectClaims, HttpRoomClaims, derive_key_from_seed},
14    http_server::contract::CreateRoomQuery,
15    request_origin::RequestOrigin,
16    room::{RoomConfig, RoomManager},
17};
18
19#[derive(Debug, Clone)]
20pub(super) struct RoomServices {
21    pub(super) room_manager: Arc<RoomManager>,
22    pub(super) media_transport: MediaTransport,
23    pub(super) metrics: Arc<RuntimeMetrics>,
24}
25
26#[derive(Debug, Clone)]
27pub(super) struct DiagnosticsServices {
28    pub(super) room_manager: Arc<RoomManager>,
29    pub(super) media_transport: MediaTransport,
30}
31
32#[derive(Debug, Clone)]
33pub(super) struct MetricsServices {
34    pub(super) room_manager: Arc<RoomManager>,
35    pub(super) metrics: Arc<RuntimeMetrics>,
36}
37
38#[derive(Debug, Clone)]
39pub(super) struct VerifiedRoomRequest {
40    pub(super) issuer: String,
41    pub(super) room_key: String,
42    pub(super) config: RoomConfig,
43    pub(super) origin: RequestOrigin,
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub(super) struct VerifiedDisconnectClaims(pub(super) HttpDisconnectClaims);
48
49/// Operator authorization bound to the listener that serves the router.
50#[derive(Clone)]
51pub(super) struct OperatorAccessPolicy {
52    auth_token: Option<Arc<str>>,
53    listener_is_loopback: bool,
54}
55
56impl OperatorAccessPolicy {
57    pub(super) fn new(auth_token: Option<&str>, listener_address: SocketAddr) -> Self {
58        Self {
59            auth_token: auth_token.map(Arc::from),
60            listener_is_loopback: listener_address.ip().is_loopback(),
61        }
62    }
63}
64
65/// A configured token disables the loopback fallback.
66pub(super) struct OperatorAccess;
67
68impl FromRef<RuntimeState> for RoomServices {
69    fn from_ref(state: &RuntimeState) -> Self {
70        Self {
71            room_manager: Arc::clone(&state.room_manager),
72            media_transport: state.media_transport.clone(),
73            metrics: Arc::clone(&state.metrics),
74        }
75    }
76}
77
78impl FromRef<RuntimeState> for DiagnosticsServices {
79    fn from_ref(state: &RuntimeState) -> Self {
80        Self {
81            room_manager: Arc::clone(&state.room_manager),
82            media_transport: state.media_transport.clone(),
83        }
84    }
85}
86
87impl FromRef<RuntimeState> for MetricsServices {
88    fn from_ref(state: &RuntimeState) -> Self {
89        Self {
90            room_manager: Arc::clone(&state.room_manager),
91            metrics: Arc::clone(&state.metrics),
92        }
93    }
94}
95
96impl FromRequestParts<RuntimeState> for VerifiedRoomRequest {
97    type Rejection = StatusCode;
98
99    async fn from_request_parts(
100        parts: &mut Parts,
101        state: &RuntimeState,
102    ) -> Result<Self, Self::Rejection> {
103        let origin = match RequestOrigin::from_request_parts(parts, state).await {
104            Ok(origin) => origin,
105            Err(error) => match error {},
106        };
107        let Query(query) = Query::<CreateRoomQuery>::from_request_parts(parts, state)
108            .await
109            .map_err(|_error| StatusCode::BAD_REQUEST)?;
110        let Some(token) = room_authorization_token(&parts.headers) else {
111            return Err(record_room_rejection(state, StatusCode::UNAUTHORIZED));
112        };
113        let claims = auth::verify::<HttpRoomClaims>(token, &state.config.auth.key)
114            .map_err(|_error| record_room_rejection(state, StatusCode::UNAUTHORIZED))?;
115        match claims {
116            HttpRoomClaims {
117                registered: RegisteredJwtClaims { iss: None, .. },
118                ..
119            } => Err(record_room_rejection(state, StatusCode::FORBIDDEN)),
120            HttpRoomClaims {
121                registered:
122                    RegisteredJwtClaims {
123                        iss: Some(issuer), ..
124                    },
125                key,
126                key_seed,
127            } => {
128                let room_key = match (key, key_seed) {
129                    (None, None) => {
130                        return Err(record_room_rejection(state, StatusCode::BAD_REQUEST));
131                    }
132                    (Some(key), None) => key,
133                    (_, Some(seed)) if seed.is_empty() => {
134                        return Err(record_room_rejection(state, StatusCode::BAD_REQUEST));
135                    }
136                    (_, Some(seed)) => derive_key_from_seed(&state.config.auth.key, seed.as_ref())
137                        .map_err(|_error| record_room_rejection(state, StatusCode::BAD_REQUEST))?,
138                };
139                Ok(Self {
140                    issuer,
141                    room_key,
142                    config: RoomConfig {
143                        web_rtc_enabled: query.web_rtc_enabled(),
144                        recording_address: query.recording_address,
145                    },
146                    origin,
147                })
148            }
149        }
150    }
151}
152
153impl FromRequest<RuntimeState> for VerifiedDisconnectClaims {
154    type Rejection = Response;
155
156    async fn from_request(req: Request, state: &RuntimeState) -> Result<Self, Self::Rejection> {
157        let body = Bytes::from_request(req, state)
158            .await
159            .map_err(IntoResponse::into_response)?;
160        let token = str::from_utf8(&body)
161            .map_err(|_error| record_disconnect_rejection(state, StatusCode::BAD_REQUEST))?;
162        let mut claims = auth::verify::<HttpDisconnectClaims>(token, &state.config.auth.key)
163            .map_err(|_error| {
164                record_disconnect_rejection(state, StatusCode::UNPROCESSABLE_ENTITY)
165            })?;
166        claims.normalize_runtime_user_ids();
167        Ok(Self(claims))
168    }
169}
170
171impl FromRequestParts<OperatorAccessPolicy> for OperatorAccess {
172    type Rejection = StatusCode;
173
174    async fn from_request_parts(
175        parts: &mut Parts,
176        policy: &OperatorAccessPolicy,
177    ) -> Result<Self, Self::Rejection> {
178        if let Some(expected_token) = policy.auth_token.as_deref() {
179            return match bearer_authorization_token(&parts.headers) {
180                Some(actual_token) if tokens_match(actual_token, expected_token) => Ok(Self),
181                _ => Err(StatusCode::UNAUTHORIZED),
182            };
183        }
184        if policy.listener_is_loopback {
185            Ok(Self)
186        } else {
187            Err(StatusCode::FORBIDDEN)
188        }
189    }
190}
191
192fn record_room_rejection(state: &RuntimeState, status: StatusCode) -> StatusCode {
193    match status {
194        StatusCode::UNAUTHORIZED => state.metrics.record_http_room_unauthorized(),
195        StatusCode::FORBIDDEN => state.metrics.record_http_room_forbidden(),
196        StatusCode::BAD_REQUEST => state.metrics.record_http_room_bad_request(),
197        _ => {}
198    }
199    status
200}
201
202fn record_disconnect_rejection(state: &RuntimeState, status: StatusCode) -> Response {
203    match status {
204        StatusCode::BAD_REQUEST => state.metrics.record_http_disconnect_bad_request(),
205        StatusCode::UNPROCESSABLE_ENTITY => {
206            state.metrics.record_http_disconnect_unprocessable_entity();
207        }
208        _ => {}
209    }
210    status.into_response()
211}
212
213fn room_authorization_token(headers: &HeaderMap) -> Option<&str> {
214    authorization_token(headers, &["Bearer", "jwt"])
215}
216
217fn bearer_authorization_token(headers: &HeaderMap) -> Option<&str> {
218    authorization_token(headers, &["Bearer"])
219}
220
221fn authorization_token<'headers>(
222    headers: &'headers HeaderMap,
223    accepted_schemes: &[&str],
224) -> Option<&'headers str> {
225    let value = headers
226        .get(header::AUTHORIZATION)
227        .and_then(|value| value.to_str().ok())?;
228    let (scheme, token) = value.split_once(' ')?;
229    if !accepted_schemes
230        .iter()
231        .any(|accepted_scheme| scheme.eq_ignore_ascii_case(accepted_scheme))
232    {
233        return None;
234    }
235    let token = token.trim_start();
236    if token.is_empty() {
237        return None;
238    }
239    Some(token)
240}
241
242fn tokens_match(actual: &str, expected: &str) -> bool {
243    let mut diff = actual.len() ^ expected.len();
244    for (actual, expected) in actual.bytes().zip(expected.bytes()) {
245        diff |= usize::from(actual ^ expected);
246    }
247    diff == 0
248}
o_sfu/runtime/http_server/extractors.rs

o_sfu/runtime/http_server/
extractors.rs
