diff --git a/Cargo.lock b/Cargo.lock index 183e91e..89179e5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2866,7 +2866,7 @@ dependencies = [ [[package]] name = "server" -version = "0.2.0" +version = "0.3.0" dependencies = [ "argon2", "async-broadcast", @@ -2887,6 +2887,7 @@ dependencies = [ "serde_json", "str0m", "symphonia", + "thiserror 2.0.18", "tokio", "tower-http", "tracing", diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 23e2328..6e6bcc7 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -36,3 +36,4 @@ opus = "0.3.1" rubato = "3.0.0" chrono = "0.4.45" regex = "1" +thiserror = "2.0.18" diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index b423697..4491284 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -46,6 +46,7 @@ use tracing::{debug, info, warn}; use crate::{ AppState, hash::{hash_password, verify_password}, + http_error::HttpError, webrtc_ingest::handle_whip_injest, }; @@ -85,11 +86,6 @@ impl HttpServer { .allow_credentials(true) .allow_origin(origins); - // TODO: Add an error type that impls IntoResponse, enum HttpError {}; impl IntoResponse for - // HttpError - // https://docs.rs/thiserror/latest/thiserror/ - // - // Its rather shitty let app = Router::new() .route("/api/catalog", get(catalog_handler)) .route("/api/user", post(create_user_handler)) @@ -136,42 +132,38 @@ async fn edit_stream_key( State(state): State>, auth: AuthUser, Json(payload): Json, -) -> impl IntoResponse { +) -> Result { // Trim surrounding whitespace so labels aren't stored with leading/trailing spaces. let new_label = payload.new.trim(); // Length (1..=67) and allowed charset ([A-Za-z0-9 _-], space included). if !KEY_RE.is_match(new_label) { - return (StatusCode::BAD_REQUEST, "invalid label").into_response(); + return Err(HttpError::BadRequest("invalid label".into())); } // Replaces the JS lookahead: label must contain at least one letter. if !new_label.chars().any(|c| c.is_ascii_alphabetic()) { - return (StatusCode::BAD_REQUEST, "label must contain a letter").into_response(); + return Err(HttpError::BadRequest("label must contain a letter".into())); } - let stream_key_id_result = stream_key::Entity::find_by_id(payload.id) + let stream_key = stream_key::Entity::find_by_id(payload.id) .one(&state.db) - .await; - if stream_key_id_result.is_err() { - return StatusCode::INTERNAL_SERVER_ERROR.into_response(); - } - let stream_key = stream_key_id_result.unwrap().unwrap(); + .await? + .ok_or(HttpError::NotFound)?; // Do they own the stream key if stream_key.user_id != auth.0.id { - return StatusCode::FORBIDDEN.into_response(); + return Err(HttpError::Forbidden); } stream_key .into_active_model() .change_label_value(&state.db, new_label.to_string()) - .await - .unwrap(); + .await?; - StatusCode::OK.into_response() + Ok(StatusCode::OK) } -async fn catalog_handler(State(state): State>) -> impl IntoResponse { - let streams = stream_session::Model::get_all_active_sessions(&state.db) - .await - .unwrap(); +async fn catalog_handler( + State(state): State>, +) -> Result { + let streams = stream_session::Model::get_all_active_sessions(&state.db).await?; debug!("{:#?}", streams); let catalog: Vec = futures::future::join_all(streams.iter().map(|listing| { let db = state.db.clone(); @@ -180,14 +172,12 @@ async fn catalog_handler(State(state): State>) -> impl IntoRespo async move { let key_info = stream_key::Entity::find_by_id(stream_key_id) .one(&db) - .await - .unwrap() - .unwrap(); + .await? + .ok_or(HttpError::NotFound)?; let user = users::Entity::find_by_id(key_info.user_id) .one(&db) - .await - .unwrap() - .unwrap(); + .await? + .ok_or(HttpError::NotFound)?; let meow = state2 .appstate .clone() @@ -195,18 +185,20 @@ async fn catalog_handler(State(state): State>) -> impl IntoRespo .await .stream_sessions .get(&key_info.id) - .unwrap() + .ok_or(HttpError::NotFound)? .started_at; - StreamListing { + Ok::<_, HttpError>(StreamListing { id: stream_key_id, label: key_info.label, user: user.username, started_at: meow, - } + }) } })) - .await; - Json(catalog) + .await + .into_iter() + .collect::, HttpError>>()?; + Ok(Json(catalog)) } #[derive(Deserialize)] @@ -233,18 +225,17 @@ fn extract_session_token(headers: &HeaderMap) -> Option { struct AuthUser(entity::users::Model); impl FromRequestParts> for AuthUser { - type Rejection = StatusCode; + type Rejection = HttpError; async fn from_request_parts( parts: &mut Parts, state: &Arc, - ) -> Result { - let token = extract_session_token(&parts.headers).ok_or(StatusCode::UNAUTHORIZED)?; + ) -> Result { + let token = extract_session_token(&parts.headers).ok_or(HttpError::Unauthorized)?; let user = users::Entity::find_by_auth_session(&state.db, token) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? - .ok_or(StatusCode::UNAUTHORIZED)?; + .await? + .ok_or(HttpError::Unauthorized)?; Ok(AuthUser(user)) } @@ -254,12 +245,11 @@ async fn create_stream_key_handler( State(state): State>, auth: AuthUser, Json(payload): Json, -) -> impl IntoResponse { +) -> Result { let uuid = uuid::Uuid::new_v4(); let value = format!("stream-key-{uuid}"); let key_amount = stream_key::Entity::find_by_user(&state.db, auth.0.id) - .await - .unwrap() + .await? .len(); if payload.label.is_empty() || payload.label.len() > MAX_LABEL_LEN { warn!( @@ -268,7 +258,7 @@ async fn create_stream_key_handler( max = MAX_LABEL_LEN, "stream key creation rejected: label length invalid" ); - return StatusCode::UNPROCESSABLE_ENTITY; + return Err(HttpError::Unprocessable("label length invalid".into())); } if key_amount >= auth.0.stream_key_limit.try_into().unwrap() { warn!( @@ -276,16 +266,12 @@ async fn create_stream_key_handler( limit = auth.0.stream_key_limit, "stream key limit reached" ); - return StatusCode::NOT_ACCEPTABLE; + return Err(HttpError::NotAcceptable("stream key limit reached".into())); } - let key = stream_key::Entity::create(&state.db, auth.0.id, value, payload.label, false).await; + let key = stream_key::Entity::create(&state.db, auth.0.id, value, payload.label, false).await?; - if let Ok(ref k) = key { - info!(user_id = auth.0.id, stream_key_id = k.id, label = %k.label, "stream key created"); - StatusCode::CREATED - } else { - StatusCode::INTERNAL_SERVER_ERROR - } + info!(user_id = auth.0.id, stream_key_id = key.id, label = %key.label, "stream key created"); + Ok(StatusCode::CREATED) } struct StreamKeys { @@ -301,15 +287,9 @@ struct StreamKey { async fn get_all_stream_keys( State(state): State>, auth: AuthUser, -) -> impl IntoResponse { - // let user_keys = stream_key::Entity::find_by_user(&state.db, auth.0.id) - // .await - // .unwrap(); - - match stream_key::Entity::find_by_user(&state.db, auth.0.id).await { - Ok(keys) => Json(keys).into_response(), - Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), - } +) -> Result { + let keys = stream_key::Entity::find_by_user(&state.db, auth.0.id).await?; + Ok(Json(keys)) } #[derive(Deserialize)] @@ -326,41 +306,33 @@ struct LoginResponse { async fn login_handler( State(state): State>, Json(payload): Json, -) -> impl IntoResponse { - if let Ok(x) = users::Entity::find_by_username(&state.db, payload.username.clone()).await { - if let Some(x) = x { - let pass = verify_password(&payload.password, &x.hashed_password); - if !pass { - warn!(username = %payload.username, "login failed: wrong password"); - let mut meow = Response::new("".to_string()); - *meow.status_mut() = StatusCode::UNAUTHORIZED; - return meow; - }; - info!(user_id = x.id, username = %x.username, "login successful"); - let auth = auth_session::Entity::create(&state.db, x.id).await.unwrap(); - let token = auth.value; - let mut meow = Response::new("".to_string()); - // TODO: Replace with axum cookie jar https://docs.rs/axum-extra/latest/axum_extra/extract/cookie/struct.CookieJar.html - meow.headers_mut().insert( - SET_COOKIE, - format!("session={token}; SameSite=Strict; Path=/; Max-Age=2592000") - .parse() - .unwrap(), - ); - *meow.status_mut() = StatusCode::OK; - return meow; - } else { +) -> Result { + let user = users::Entity::find_by_username(&state.db, payload.username.clone()) + .await? + .ok_or_else(|| { warn!(username = %payload.username, "login failed: user not found"); - let mut meow = Response::new("".to_string()); - *meow.status_mut() = StatusCode::UNAUTHORIZED; - return meow; - } - } else { - warn!(username = %payload.username, "login failed: DB error"); - let mut meow = Response::new("".to_string()); - *meow.status_mut() = StatusCode::UNAUTHORIZED; - return meow; + HttpError::Unauthorized + })?; + + if !verify_password(&payload.password, &user.hashed_password) { + warn!(username = %payload.username, "login failed: wrong password"); + return Err(HttpError::Unauthorized); } + + info!(user_id = user.id, username = %user.username, "login successful"); + let auth = auth_session::Entity::create(&state.db, user.id).await?; + let token = auth.value; + + let mut meow = Response::new("".to_string()); + // TODO: Replace with axum cookie jar https://docs.rs/axum-extra/latest/axum_extra/extract/cookie/struct.CookieJar.html + meow.headers_mut().insert( + SET_COOKIE, + format!("session={token}; SameSite=Strict; Path=/; Max-Age=2592000") + .parse() + .unwrap(), + ); + *meow.status_mut() = StatusCode::OK; + Ok(meow) } #[derive(Deserialize)] @@ -373,34 +345,33 @@ struct CreateUserForm { async fn create_user_handler( State(state): State>, Json(payload): Json, -) -> (HeaderMap, StatusCode) { +) -> Result<(HeaderMap, StatusCode), HttpError> { if state.config.signup_code.is_empty() || payload.ref_token != state.config.signup_code { warn!(username = %payload.username, "signup rejected: invalid signup code"); - return (HeaderMap::new(), StatusCode::UNAUTHORIZED); + return Err(HttpError::Unauthorized); } if payload.username.is_empty() || payload.username.len() > MAX_USERNAME_LEN { warn!(username = %payload.username, max = MAX_USERNAME_LEN, "signup rejected: username length invalid"); - return (HeaderMap::new(), StatusCode::UNPROCESSABLE_ENTITY); + return Err(HttpError::Unprocessable("username length invalid".into())); } - let meow = users::Entity::create( + let user = users::Entity::create( &state.db, payload.username.clone(), - hash_password(&payload.password).unwrap(), + hash_password(&payload.password).map_err(|_| HttpError::Hash)?, ) - .await; + .await + .map_err(|e| { + // A create failure here is almost always a username unique-constraint + // violation, so surface it as a conflict rather than a 500. + warn!(username = %payload.username, error = %e, "user creation failed (likely username conflict)"); + HttpError::Conflict + })?; - // Create session - let session: auth_session::Model = if let Ok(ref user) = meow { - info!(user_id = user.id, username = %user.username, "user created"); - auth_session::Entity::create(&state.db, user.id) - .await - .unwrap() - } else { - warn!(username = %payload.username, "user creation failed (likely username conflict)"); - return (HeaderMap::new(), StatusCode::CONFLICT); - }; + info!(user_id = user.id, username = %user.username, "user created"); + let session = auth_session::Entity::create(&state.db, user.id).await?; let token = session.value; + let mut headers = HeaderMap::new(); // TODO: Replace with axum cookie jar https://docs.rs/axum-extra/latest/axum_extra/extract/cookie/struct.CookieJar.html headers.insert( @@ -409,14 +380,14 @@ async fn create_user_handler( .parse() .unwrap(), ); - (headers, StatusCode::OK) + Ok((headers, StatusCode::OK)) } async fn stream_handler( State(state): State>, Path(slug): Path, body: String, -) -> impl IntoResponse { +) -> Result { let request_id_clone = state.request_count.fetch_add(1, Ordering::Relaxed); let stream_key_id = { @@ -427,47 +398,42 @@ async fn stream_handler( .map(|e| *e.key()) }; - let stream_key_id = if let Some(id) = stream_key_id { - id - } else { + let stream_key_id = stream_key_id.ok_or_else(|| { warn!(slug = %slug, "WHEP request for unknown or inactive stream"); - return Response::builder() - .status(StatusCode::NOT_FOUND) - .header("content-type", "application/text") - .body("".to_string()) - .unwrap(); - }; + HttpError::NotFound + })?; info!(request_id = request_id_clone, slug = %slug, stream_key_id, "WHEP offer received"); let mut accept_rx = state.accept_rx.activate_cloned(); - let _ = state + // The webrtc worker owning the receiver died if this fails. + state .offer_tx .send((request_id_clone, stream_key_id, body)) .await - .unwrap(); + .map_err(|_| HttpError::Internal)?; debug!( request_id = request_id_clone, "offer sent, waiting for answer" ); - let mut reply_body = String::new(); + let reply_body = String::new(); while let Ok(answer) = accept_rx.recv().await { debug!(request_id = request_id_clone, "received answer candidate"); if let Some(reply) = answer.1 { if answer.0 == request_id_clone { - return Response::builder() + return Ok(Response::builder() .status(StatusCode::CREATED) .header("content-type", "application/sdp") .body(reply) - .unwrap(); + .unwrap()); } else { continue; } } else if answer.0 == request_id_clone { - return Response::builder() + return Ok(Response::builder() .status(StatusCode::UNSUPPORTED_MEDIA_TYPE) .body(String::new()) - .unwrap(); + .unwrap()); } } info!( @@ -475,11 +441,11 @@ async fn stream_handler( "answer channel closed without a match" ); - Response::builder() + Ok(Response::builder() .status(StatusCode::CREATED) .header("content-type", "application/sdp") .body(reply_body) - .unwrap() + .unwrap()) } async fn meow_handler() -> &'static str { diff --git a/crates/server/src/http_error.rs b/crates/server/src/http_error.rs new file mode 100644 index 0000000..3f22566 --- /dev/null +++ b/crates/server/src/http_error.rs @@ -0,0 +1,59 @@ +use axum::{ + body::Body, + http::{Response, StatusCode}, + response::IntoResponse, +}; +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum HttpError { + #[error("db shit itself: {0}")] + DbErr(#[from] sea_orm::DbErr), + #[error("hashing failed")] + Hash, + #[error("resource not found")] + NotFound, + #[error("unauthorized")] + Unauthorized, + #[error("forbidden")] + Forbidden, + #[error("conflict")] + Conflict, + #[error("bad request: {0}")] + BadRequest(String), + #[error("unprocessable entity: {0}")] + Unprocessable(String), + #[error("not acceptable: {0}")] + NotAcceptable(String), + #[error("internal error")] + Internal, +} + +impl HttpError { + fn status(&self) -> StatusCode { + match self { + Self::DbErr(_) | Self::Hash | Self::Internal => StatusCode::INTERNAL_SERVER_ERROR, + Self::NotFound => StatusCode::NOT_FOUND, + Self::Unauthorized => StatusCode::UNAUTHORIZED, + Self::Forbidden => StatusCode::FORBIDDEN, + Self::Conflict => StatusCode::CONFLICT, + Self::BadRequest(_) => StatusCode::BAD_REQUEST, + Self::Unprocessable(_) => StatusCode::UNPROCESSABLE_ENTITY, + Self::NotAcceptable(_) => StatusCode::NOT_ACCEPTABLE, + } + } +} + +impl IntoResponse for HttpError { + fn into_response(self) -> Response { + let status = self.status(); + // Only surface a message body for client (4xx) errors; keep an empty + // body for 5xx so internal details aren't leaked. + let body = if status.is_client_error() { + self.to_string() + } else { + String::new() + }; + (status, body).into_response() + } +} diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index a366adc..27335b3 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -35,6 +35,7 @@ mod audio; mod codec; mod hash; mod http; +mod http_error; mod rtmp; mod webrtc; mod webrtc_ingest;