601 lines
18 KiB
Rust
601 lines
18 KiB
Rust
use std::{
|
|
net::SocketAddr,
|
|
sync::{
|
|
Arc,
|
|
atomic::{AtomicI32, Ordering},
|
|
},
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
use axum_extra::extract::{CookieJar, cookie::Cookie};
|
|
|
|
/// Allowed charset for labels, passwords and custom IDs: letters, numbers,
|
|
/// dashes and apostrophes — no spaces. Mirrors the frontend
|
|
/// `/^[A-Za-z0-9'-]{0,67}$/` used by the keys-page popups.
|
|
fn valid_charset(s: &str) -> bool {
|
|
s.len() <= 67
|
|
&& s.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '\'')
|
|
}
|
|
|
|
use axum::{
|
|
Json, Router,
|
|
extract::{FromRequestParts, Path, State},
|
|
http::{
|
|
HeaderMap, HeaderName, Method, StatusCode,
|
|
header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE},
|
|
request::Parts,
|
|
},
|
|
response::{IntoResponse, Response},
|
|
routing::{delete, get, post},
|
|
};
|
|
use chrono::{DateTime, Utc};
|
|
use entity::{auth_session, stream_key, stream_session, users};
|
|
use sea_orm::{ActiveModelTrait, DatabaseConnection, EntityTrait, IntoActiveModel, Set};
|
|
use serde::{Deserialize, Serialize};
|
|
use sysinfo::System;
|
|
use tokio::{
|
|
net::TcpListener,
|
|
sync::{Mutex, mpsc::Sender},
|
|
};
|
|
use tower_http::cors::CorsLayer;
|
|
|
|
use tracing::{debug, info, warn};
|
|
|
|
use crate::{
|
|
AppState,
|
|
hash::{hash_password, verify_password},
|
|
http_error::HttpError,
|
|
webrtc_ingest::{handle_whip_injest, handle_whip_injest_delete, handle_whip_injest_patch},
|
|
};
|
|
|
|
const MAX_USERNAME_LEN: usize = 32;
|
|
const MAX_LABEL_LEN: usize = 64;
|
|
|
|
pub struct HttpServer {
|
|
pub offer_tx: Sender<(i32, i32, String)>,
|
|
pub accept_rx: async_broadcast::InactiveReceiver<(i32, Result<String, String>)>,
|
|
pub appstate: Arc<Mutex<AppState>>,
|
|
pub request_count: AtomicI32,
|
|
pub db: DatabaseConnection,
|
|
pub signup_code: String,
|
|
pub version: &'static str,
|
|
pub start_time: Instant,
|
|
}
|
|
|
|
impl HttpServer {
|
|
pub async fn run(self) {
|
|
let state = Arc::new(self);
|
|
|
|
let origins = [
|
|
"http://localhost:5173".parse().unwrap(),
|
|
"https://stream.h.doloro.co.uk".parse().unwrap(),
|
|
];
|
|
|
|
let cors = CorsLayer::new()
|
|
.allow_methods([
|
|
Method::GET,
|
|
Method::POST,
|
|
Method::PATCH,
|
|
Method::DELETE,
|
|
Method::OPTIONS,
|
|
])
|
|
.allow_headers([
|
|
AUTHORIZATION,
|
|
ACCEPT,
|
|
CONTENT_TYPE,
|
|
HeaderName::from_static("session"),
|
|
])
|
|
.allow_credentials(true)
|
|
.allow_origin(origins);
|
|
|
|
let app = Router::new()
|
|
.route("/api/catalog", get(catalog_handler))
|
|
.route("/api/user", post(create_user_handler))
|
|
.route(
|
|
"/api/stream-key",
|
|
post(create_stream_key_handler)
|
|
.get(get_all_stream_keys)
|
|
.patch(edit_stream_key),
|
|
)
|
|
.route("/api/login", post(login_handler))
|
|
.route("/api/stream/{slug}", post(stream_handler))
|
|
.route("/api/whip", post(handle_whip_injest))
|
|
.route(
|
|
"/api/whip/{slug}",
|
|
delete(handle_whip_injest_delete).patch(handle_whip_injest_patch),
|
|
)
|
|
.route("/api/meow", get(meow_handler))
|
|
.route("/api/health", get(health_handler))
|
|
.route("/api/stats", get(stats_handler))
|
|
.layer(cors)
|
|
.with_state(state);
|
|
|
|
let addr = SocketAddr::from(([0, 0, 0, 0], 3000));
|
|
let listener = TcpListener::bind(addr).await.unwrap();
|
|
axum::serve(
|
|
listener,
|
|
app.into_make_service_with_connect_info::<SocketAddr>(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
}
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct StreamListing {
|
|
label: String,
|
|
custom_url_label: String,
|
|
id: i32,
|
|
user: String,
|
|
started_at: DateTime<Utc>,
|
|
is_password_protected: bool,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct EditStreamKeyRequest {
|
|
id: i32,
|
|
#[serde(default)]
|
|
new: Option<String>,
|
|
#[serde(default)]
|
|
password: Option<String>,
|
|
#[serde(default)]
|
|
unlisted: Option<bool>,
|
|
#[serde(default)]
|
|
custom_id: Option<String>,
|
|
}
|
|
|
|
async fn edit_stream_key(
|
|
State(state): State<Arc<HttpServer>>,
|
|
auth: AuthUser,
|
|
Json(payload): Json<EditStreamKeyRequest>,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
// Trim surrounding whitespace so labels aren't stored with leading/trailing spaces.
|
|
let new_label = payload.new.as_deref().map(str::trim);
|
|
// Length (1..=67), allowed charset, and at least one letter.
|
|
if let Some(label) = new_label {
|
|
if label.is_empty() || !valid_charset(label) {
|
|
return Err(HttpError::BadRequest("invalid label".into()));
|
|
}
|
|
if !label.chars().any(|c| c.is_ascii_alphabetic()) {
|
|
return Err(HttpError::BadRequest("label must contain a letter".into()));
|
|
}
|
|
}
|
|
// Empty values are allowed (they clear the field).
|
|
if let Some(custom_id) = payload.custom_id.as_deref() {
|
|
if !custom_id.is_empty() && !valid_charset(custom_id) {
|
|
return Err(HttpError::BadRequest("invalid custom id".into()));
|
|
}
|
|
}
|
|
if let Some(pwd) = payload.password.as_deref() {
|
|
if !pwd.is_empty() && !valid_charset(pwd) {
|
|
return Err(HttpError::BadRequest("invalid password".into()));
|
|
}
|
|
}
|
|
|
|
let stream_key = stream_key::Entity::find_by_id(payload.id)
|
|
.one(&state.db)
|
|
.await?
|
|
.ok_or(HttpError::NotFound)?;
|
|
// Do they own the stream key
|
|
if stream_key.user_id != auth.0.id {
|
|
return Err(HttpError::Forbidden);
|
|
}
|
|
|
|
let lock = state.appstate.lock().await;
|
|
let mut ses = { lock.stream_sessions.get_mut(&payload.id) };
|
|
// drop(lock);
|
|
|
|
let mut am = stream_key.into_active_model();
|
|
if let Some(custom_id) = payload.custom_id {
|
|
am.custom_id = if custom_id.is_empty() {
|
|
if let Some(ref mut ses) = ses {
|
|
ses.custom_id = None;
|
|
}
|
|
Set(None)
|
|
} else {
|
|
// Check for conflict.
|
|
if stream_key::Entity::find_by_custom_id(&state.db, custom_id.clone())
|
|
.await?
|
|
.is_some()
|
|
{
|
|
return Err(HttpError::Conflict);
|
|
}
|
|
if let Some(ref mut ses) = ses {
|
|
ses.custom_id = Some(custom_id.clone());
|
|
}
|
|
Set(Some(custom_id))
|
|
};
|
|
}
|
|
if let Some(label) = new_label {
|
|
if let Some(ref mut ses) = ses {
|
|
ses.stream_key_label = label.to_string();
|
|
}
|
|
am.label = Set(label.to_string());
|
|
}
|
|
if let Some(pwd) = payload.password {
|
|
am.password = if pwd.is_empty() {
|
|
if let Some(ref mut ses) = ses {
|
|
ses.password = None;
|
|
}
|
|
Set(None)
|
|
} else {
|
|
if let Some(ref mut ses) = ses {
|
|
ses.password = Some(pwd.clone());
|
|
}
|
|
Set(Some(pwd))
|
|
};
|
|
}
|
|
if let Some(unlisted) = payload.unlisted {
|
|
if let Some(ref mut ses) = ses {
|
|
ses.is_unlisted = unlisted;
|
|
}
|
|
am.is_unlisted = Set(unlisted);
|
|
}
|
|
// This hopefully will not fail, if it does, our values for the stream key will be mismatched,
|
|
// that would be bad
|
|
am.update(&state.db).await?;
|
|
|
|
Ok(StatusCode::OK)
|
|
}
|
|
|
|
async fn catalog_handler(
|
|
State(state): State<Arc<HttpServer>>,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
let streams = stream_session::Model::get_all_active_sessions(&state.db).await?;
|
|
debug!("{:#?}", streams);
|
|
let catalog: Vec<StreamListing> = state
|
|
.appstate
|
|
.lock()
|
|
.await
|
|
.stream_sessions
|
|
.iter()
|
|
.filter(|x| !x.is_unlisted)
|
|
.map(|x| StreamListing {
|
|
id: x.stream_key_id,
|
|
label: x.stream_key_label.clone(),
|
|
custom_url_label: x.custom_id.clone().unwrap_or_default(),
|
|
user: x.stream_key_user.clone(),
|
|
started_at: x.started_at,
|
|
is_password_protected: x.password.is_some(),
|
|
})
|
|
.collect::<Vec<_>>();
|
|
Ok(Json(catalog))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct CreateStreamKeyBody {
|
|
label: String,
|
|
}
|
|
|
|
struct AuthUser(entity::users::Model);
|
|
|
|
impl FromRequestParts<Arc<HttpServer>> for AuthUser {
|
|
type Rejection = HttpError;
|
|
|
|
async fn from_request_parts(
|
|
parts: &mut Parts,
|
|
state: &Arc<HttpServer>,
|
|
) -> Result<Self, HttpError> {
|
|
let jar = CookieJar::from_headers(&parts.headers);
|
|
let session = jar.get("session").ok_or(HttpError::Unauthorized)?;
|
|
let session_val = session.value().to_string();
|
|
tracing::debug!(?session_val, "AuthUser: extracted session cookie");
|
|
|
|
let user = match users::Entity::find_by_auth_session(&state.db, session_val.clone()).await {
|
|
Ok(u) => u.ok_or(HttpError::Unauthorized)?,
|
|
Err(e) => {
|
|
tracing::error!(?session_val, error = %e, "AuthUser: db query failed");
|
|
return Err(HttpError::DbErr(e));
|
|
}
|
|
};
|
|
|
|
Ok(AuthUser(user))
|
|
}
|
|
}
|
|
|
|
async fn create_stream_key_handler(
|
|
State(state): State<Arc<HttpServer>>,
|
|
auth: AuthUser,
|
|
Json(payload): Json<CreateStreamKeyBody>,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
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?
|
|
.len();
|
|
if payload.label.is_empty() || payload.label.len() > MAX_LABEL_LEN {
|
|
warn!(
|
|
user_id = auth.0.id,
|
|
label_len = payload.label.len(),
|
|
max = MAX_LABEL_LEN,
|
|
"stream key creation rejected: label length invalid"
|
|
);
|
|
return Err(HttpError::Unprocessable("label length invalid".into()));
|
|
}
|
|
if key_amount >= auth.0.stream_key_limit.try_into().unwrap() {
|
|
warn!(
|
|
user_id = auth.0.id,
|
|
limit = auth.0.stream_key_limit,
|
|
"stream key limit reached"
|
|
);
|
|
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?;
|
|
|
|
info!(user_id = auth.0.id, stream_key_id = key.id, label = %key.label, "stream key created");
|
|
Ok(StatusCode::CREATED)
|
|
}
|
|
|
|
async fn get_all_stream_keys(
|
|
State(state): State<Arc<HttpServer>>,
|
|
auth: AuthUser,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
let keys = stream_key::Entity::find_by_user(&state.db, auth.0.id).await?;
|
|
Ok(Json(keys))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct LoginForm {
|
|
username: String,
|
|
password: String,
|
|
}
|
|
|
|
struct DevFlag(bool);
|
|
|
|
impl<S> FromRequestParts<S> for DevFlag
|
|
where
|
|
S: Send + Sync,
|
|
{
|
|
type Rejection = std::convert::Infallible;
|
|
|
|
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
|
|
let query = parts.uri.query().unwrap_or("");
|
|
Ok(DevFlag(query.contains("dev=1")))
|
|
}
|
|
}
|
|
|
|
fn cookie_for_token(token: &str, dev: bool) -> Cookie<'static> {
|
|
let token = token.to_owned();
|
|
let mut cookie = Cookie::build(("session", token))
|
|
.path("/")
|
|
.max_age(time::Duration::days(30))
|
|
.same_site(if dev {
|
|
axum_extra::extract::cookie::SameSite::None
|
|
} else {
|
|
axum_extra::extract::cookie::SameSite::Lax
|
|
});
|
|
if dev {
|
|
cookie = cookie.secure(true);
|
|
}
|
|
cookie.build()
|
|
}
|
|
|
|
#[axum::debug_handler]
|
|
async fn login_handler(
|
|
State(state): State<Arc<HttpServer>>,
|
|
DevFlag(dev): DevFlag,
|
|
Json(payload): Json<LoginForm>,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
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");
|
|
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 cookie = cookie_for_token(&token, dev);
|
|
Ok((
|
|
StatusCode::OK,
|
|
[(axum::http::header::SET_COOKIE, cookie.to_string())],
|
|
))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct CreateUserForm {
|
|
username: String,
|
|
password: String,
|
|
ref_token: String,
|
|
}
|
|
|
|
async fn create_user_handler(
|
|
State(state): State<Arc<HttpServer>>,
|
|
DevFlag(dev): DevFlag,
|
|
Json(payload): Json<CreateUserForm>,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
if state.signup_code.is_empty() || payload.ref_token != state.signup_code {
|
|
warn!(username = %payload.username, "signup rejected: invalid signup code");
|
|
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 Err(HttpError::Unprocessable("username length invalid".into()));
|
|
}
|
|
|
|
let user = users::Entity::create(
|
|
&state.db,
|
|
payload.username.clone(),
|
|
hash_password(&payload.password).map_err(|_| HttpError::Hash)?,
|
|
)
|
|
.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
|
|
})?;
|
|
|
|
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 cookie = cookie_for_token(&token, dev);
|
|
Ok((
|
|
StatusCode::OK,
|
|
[(axum::http::header::SET_COOKIE, cookie.to_string())],
|
|
))
|
|
}
|
|
|
|
async fn stream_handler(
|
|
State(state): State<Arc<HttpServer>>,
|
|
Path(slug): Path<String>,
|
|
headers: HeaderMap,
|
|
body: String,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
let request_id_clone = state.request_count.fetch_add(1, Ordering::Relaxed);
|
|
let app = state.appstate.lock().await;
|
|
|
|
let stream_key_id = {
|
|
app.stream_sessions.iter().find(|e| {
|
|
e.value().stream_key_id.to_string() == slug
|
|
|| e.value().custom_id.as_deref() == Some(slug.as_str())
|
|
})
|
|
// .map(|e| *e.key())
|
|
};
|
|
|
|
let stream_key_id = stream_key_id.ok_or_else(|| {
|
|
// warn!(slug = %slug, "WHEP request for unknown or inactive stream");
|
|
HttpError::NotFound
|
|
})?;
|
|
|
|
// Dont return stream by its id in the db if its unlisted
|
|
if stream_key_id.is_unlisted && stream_key_id.custom_id.clone().ok_or("") != Ok(slug.clone()) {
|
|
return Err(HttpError::NotFound);
|
|
}
|
|
|
|
let auth_header = headers.get("auth");
|
|
|
|
if let Some(password) = &stream_key_id.password {
|
|
if let Some(auth) = auth_header {
|
|
if auth.to_str().unwrap() != password {
|
|
return Err(HttpError::Unauthorized);
|
|
}
|
|
} else {
|
|
return Err(HttpError::Unauthorized);
|
|
};
|
|
};
|
|
|
|
info!(request_id = request_id_clone, slug = %slug, stream_key_id.stream_key_id, "WHEP offer received");
|
|
let accept_rx = state.accept_rx.activate_cloned();
|
|
// The webrtc worker owning the receiver died if this fails.
|
|
state
|
|
.offer_tx
|
|
.send((request_id_clone, stream_key_id.stream_key_id, body))
|
|
.await
|
|
.map_err(|_| HttpError::Internal)?;
|
|
debug!(
|
|
request_id = request_id_clone,
|
|
"offer sent, waiting for answer"
|
|
);
|
|
|
|
// Bound the wait: a WebRTC setup failure (malformed SDP, codec mismatch)
|
|
// would otherwise leave this HTTP request hanging forever.
|
|
match tokio::time::timeout(Duration::from_secs(10), async {
|
|
let mut accept_rx = accept_rx;
|
|
loop {
|
|
match accept_rx.recv().await {
|
|
Ok(answer) if answer.0 == request_id_clone => return Some(answer.1),
|
|
Ok(_) => continue,
|
|
Err(_) => return None,
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
{
|
|
Ok(Some(Ok(reply))) => {
|
|
return Ok(Response::builder()
|
|
.status(StatusCode::CREATED)
|
|
.header("content-type", "application/sdp")
|
|
.body(reply)
|
|
.unwrap());
|
|
}
|
|
Ok(Some(Err(codec))) => {
|
|
return Err(HttpError::WhepCodecError(codec));
|
|
}
|
|
Ok(None) => {
|
|
info!(
|
|
request_id = request_id_clone,
|
|
"answer channel closed without a match"
|
|
);
|
|
}
|
|
Err(_) => {
|
|
warn!(
|
|
request_id = request_id_clone,
|
|
"timed out waiting for WHEP answer"
|
|
);
|
|
}
|
|
};
|
|
|
|
Ok(Response::builder()
|
|
.status(StatusCode::GATEWAY_TIMEOUT)
|
|
.body(String::new())
|
|
.unwrap())
|
|
}
|
|
|
|
async fn meow_handler() -> &'static str {
|
|
"meow"
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct HealthResponse {
|
|
status: &'static str,
|
|
uptime_seconds: u64,
|
|
version: &'static str,
|
|
}
|
|
|
|
async fn health_handler(
|
|
State(state): State<Arc<HttpServer>>,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
let uptime = state.start_time.elapsed().as_secs();
|
|
Ok((
|
|
StatusCode::OK,
|
|
Json(HealthResponse {
|
|
status: "ok",
|
|
uptime_seconds: uptime,
|
|
version: state.version,
|
|
}),
|
|
))
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct StatsResponse {
|
|
version: &'static str,
|
|
uptime_seconds: u64,
|
|
cpu_usage_percent: f32,
|
|
active_streams: usize,
|
|
request_count: i32,
|
|
}
|
|
|
|
async fn stats_handler(
|
|
State(state): State<Arc<HttpServer>>,
|
|
) -> Result<impl IntoResponse, HttpError> {
|
|
let uptime = state.start_time.elapsed().as_secs();
|
|
let mut system = System::new();
|
|
system.refresh_cpu_all();
|
|
let cpu_usage = system.global_cpu_usage();
|
|
let active_streams = {
|
|
let app = state.appstate.lock().await;
|
|
app.stream_sessions.len()
|
|
};
|
|
let request_count = state.request_count.load(Ordering::Relaxed);
|
|
Ok((
|
|
StatusCode::OK,
|
|
Json(StatsResponse {
|
|
version: state.version,
|
|
uptime_seconds: uptime,
|
|
cpu_usage_percent: cpu_usage,
|
|
active_streams,
|
|
request_count,
|
|
}),
|
|
))
|
|
}
|