a lot

crates/server/src/hash.rs

@@ -0,0 +1,15 @@
+use argon2::password_hash::{SaltString, rand_core::OsRng};
+use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier};
+
+pub fn hash_password(password: &str) -> Result<String, argon2::password_hash::Error> {
+    let salt = SaltString::generate(&mut OsRng);
+    let hash = Argon2::default().hash_password(password.as_bytes(), &salt)?;
+    Ok(hash.to_string())
+}
+
+pub fn verify_password(password: &str, stored_hash: &str) -> bool {
+    let parsed = PasswordHash::new(stored_hash).unwrap();
+    Argon2::default()
+        .verify_password(password.as_bytes(), &parsed)
+        .is_ok()
+}

crates/server/src/http.rs

@@ -1,30 +1,30 @@
-use std::{
-    hash::{DefaultHasher, Hash, Hasher},
-    net::SocketAddr,
-    sync::Arc,
-};
+use std::{net::SocketAddr, sync::Arc};
 
 use axum::{
     Json, Router,
-    body::Bytes,
-    extract::{Form, Query, State},
-    http::StatusCode,
+    body::{Body, Bytes},
+    extract::{Form, FromRequestParts, Path, Query, State},
+    http::{HeaderMap, StatusCode, header::SET_COOKIE, request::Parts},
     response::{IntoResponse, Response},
     routing::{get, post},
 };
+use entity::{auth_session, stream_key, users};
+use sea_orm::DatabaseConnection;
 use serde::{Deserialize, Serialize};
 use tokio::{
     net::TcpListener,
     sync::{Mutex, mpsc::Sender},
 };
+use uuid::Uuid;
 
-use crate::AppState;
+use crate::{AppState, hash::hash_password};
 
 pub struct HttpServer {
-    pub offer_tx: Sender<(i32, String, String)>,
-    pub accept_rx: async_broadcast::Receiver<(i32, String)>,
+    pub offer_tx: Sender<(i32, i32, String)>,
+    pub accept_rx: async_broadcast::Receiver<(i32, Option<String>)>,
     pub appstate: Arc<Mutex<AppState>>,
     pub request_count: Mutex<i32>,
+    pub db: DatabaseConnection,
 }
 
 impl HttpServer {
@@ -35,7 +35,12 @@ impl HttpServer {
             let app = Router::new()
                 .route("/api/catalog", get(catalog_handler))
                 .route("/api/user", post(create_user_handler))
-                .route("/api/stream", get(stream_handler))
+                .route(
+                    "/api/stream-key",
+                    post(create_stream_key_handler).get(get_all_stream_keys),
+                )
+                .route("/api/login", post(login_handler))
+                .route("/api/stream/{slug}", post(stream_handler))
                 .route("/api/meow", get(meow_handler))
                 .with_state(state);
 
@@ -47,17 +52,103 @@ impl HttpServer {
     }
 }
 
-fn calculate_hash<T: Hash>(t: &T) -> u64 {
-    let mut s = DefaultHasher::new();
-    t.hash(&mut s);
-    s.finish()
-}
-
 async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoResponse {
     let catalog = catalog_from_state(&state.appstate).await;
     Json(catalog)
 }
 
+#[derive(Deserialize)]
+struct CreateStreamKeyBody {
+    label: String,
+}
+
+struct AuthUser(entity::users::Model);
+
+impl FromRequestParts<Arc<HttpServer>> for AuthUser {
+    type Rejection = StatusCode;
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        state: &Arc<HttpServer>,
+    ) -> Result<Self, StatusCode> {
+        let token = parts
+            .headers
+            .get("session")
+            // parse token...
+            .ok_or(StatusCode::UNAUTHORIZED)?;
+
+        let user =
+            users::Entity::find_by_auth_session(&state.db, token.to_str().unwrap().to_string())
+                .await
+                .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
+                .ok_or(StatusCode::UNAUTHORIZED)?;
+
+        Ok(AuthUser(user))
+    }
+}
+
+async fn create_stream_key_handler(
+    State(state): State<Arc<HttpServer>>,
+    auth: AuthUser,
+    Json(payload): Json<CreateStreamKeyBody>,
+) -> impl IntoResponse {
+    let uuid = uuid::Uuid::new_v4();
+    let value = format!("stream-key-{uuid}");
+    let key = stream_key::Entity::create(&state.db, auth.0.id, value, payload.label, false).await;
+
+    if let Ok(_key) = key {
+        StatusCode::CREATED
+    } else {
+        StatusCode::INTERNAL_SERVER_ERROR
+    }
+}
+
+struct StreamKeys {
+    keys: Vec<StreamKey>,
+}
+
+struct StreamKey {
+    id: i32,
+    label: String,
+    value: String,
+}
+
+async fn get_all_stream_keys(
+    State(state): State<Arc<HttpServer>>,
+    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(),
+    }
+}
+
+#[derive(Deserialize)]
+struct LoginForm {
+    username: String,
+    password: String,
+}
+
+#[derive(Serialize)]
+struct LoginResponse {
+    session_token: String,
+}
+
+async fn login_handler(
+    State(state): State<Arc<HttpServer>>,
+    Json(payload): Json<LoginForm>,
+) -> StatusCode {
+    // TODO: hash payload.password with a real password hasher (argon2/bcrypt), look up user by username
+    // TODO: verify hash matches stored hashed_password
+    // TODO: call auth_session::Entity::create, return session token
+    // TODO: return 401 on bad credentials
+    todo!("login not yet implemented")
+}
+
 #[derive(Deserialize)]
 struct CreateUserForm {
     username: String,
@@ -65,44 +156,91 @@ struct CreateUserForm {
     ref_token: String,
 }
 
-async fn create_user_handler(Form(form): Form<CreateUserForm>) -> StatusCode {
-    // TODO: ref_token validation, db integration
-    let _ = (
-        form.username,
-        calculate_hash(&form.password),
-        form.ref_token,
-    );
-    todo!("user creation not yet implemented")
-}
+async fn create_user_handler(
+    State(state): State<Arc<HttpServer>>,
+    Json(payload): Json<CreateUserForm>,
+) -> (HeaderMap, StatusCode) {
+    if payload.ref_token != "TEST" {
+        return (HeaderMap::new(), StatusCode::UNAUTHORIZED);
+    }
 
-#[derive(Deserialize)]
-struct StreamQuery {
-    stream_label: String,
+    let meow = users::Entity::create(
+        &state.db,
+        payload.username,
+        hash_password(&payload.password).unwrap(),
+    )
+    .await;
+
+    // Create session
+    let session: auth_session::Model = if let Ok(meow) = meow {
+        auth_session::Entity::create(&state.db, meow.id)
+            .await
+            .unwrap() // This should be ok (hopefully)
+    } else {
+        return (HeaderMap::new(), StatusCode::CONFLICT);
+    };
+    let token = session.value;
+    let mut headers = HeaderMap::new();
+    headers.insert(
+        SET_COOKIE,
+        format!("session={token}; HttpOnly; SameSite=Strict; Path=/")
+            .parse()
+            .unwrap(),
+    );
+    (headers, StatusCode::OK)
 }
 
 async fn stream_handler(
     State(state): State<Arc<HttpServer>>,
-    Query(query): Query<StreamQuery>,
-    body: Bytes,
+    Path(slug): Path<String>,
+    body: String,
 ) -> impl IntoResponse {
-    let body_str = String::from_utf8(body.to_vec()).unwrap_or_default();
-
     let mut request_id = state.request_count.lock().await;
     *request_id += 1;
     let request_id_clone = *request_id;
     drop(request_id);
 
+    let stream_key_id = {
+        let app = state.appstate.lock().await;
+        app.stream_sessions
+            .iter()
+            .find(|e| e.value().stream_key_label == slug)
+            .map(|e| *e.key())
+    };
+
+    let stream_key_id = if let Some(id) = stream_key_id {
+        id
+    } else {
+        return Response::builder()
+            .status(StatusCode::NOT_FOUND)
+            .header("content-type", "application/text")
+            .body("".to_string())
+            .unwrap();
+    };
+
     let mut accept_rx = state.accept_rx.new_receiver();
     let _ = state
         .offer_tx
-        .send((request_id_clone, query.stream_label, body_str))
-        .await;
+        .send((request_id_clone, stream_key_id, body))
+        .await
+        .unwrap();
 
     let mut reply_body = String::new();
     while let Ok(answer) = accept_rx.recv().await {
-        if answer.0 == request_id_clone {
-            reply_body = answer.1;
-            break;
+        if let Some(reply) = answer.1 {
+            if answer.0 == request_id_clone {
+                Response::builder()
+                    .status(StatusCode::CREATED)
+                    .header("content-type", "application/sdp")
+                    .body(reply)
+                    .unwrap();
+            }
+        } else {
+            Response::builder()
+                .status(StatusCode::NOT_FOUND)
+                // .header("content-type", "application/sdp")
+                .body("")
+                .unwrap();
         }
     }
 
@@ -119,9 +257,10 @@ async fn meow_handler() -> &'static str {
 
 #[derive(Serialize)]
 struct StreamCatalog {
-    active_streams: Vec<String>,
+    active_streams: Vec<i32>,
 }
 
+// TODO: Make this return streak labels and their id instead of the whole stream key omfg
 async fn catalog_from_state(state: &Arc<Mutex<AppState>>) -> StreamCatalog {
     let app = state.lock().await;
     let active_streams = app

crates/server/src/main.rs

@@ -2,10 +2,12 @@ use std::{error::Error, sync::Arc};
 
 use async_broadcast::broadcast;
 use dashmap::DashMap;
+use migration::{Migrator, MigratorTrait};
 use rml_rtmp::{
     handshake::{Handshake, HandshakeProcessResult, PeerType},
     sessions::{ServerSession, ServerSessionConfig, ServerSessionEvent, ServerSessionResult},
 };
+use sea_orm::{Database, sqlx::types::chrono};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::TcpListener,
@@ -17,33 +19,49 @@ use crate::{
     media::{H264Parser, VideoFrame},
 };
 
+mod hash;
 mod http;
 mod media;
 mod rtmp;
 mod webrtc;
 
 pub struct AppState {
-    pub stream_sessions: Arc<DashMap<String, StreamSession>>,
+    pub stream_sessions: Arc<DashMap<i32, StreamSession>>,
 }
 
 pub struct StreamSession {
-    pub stream_key: String,
+    pub stream_key_id: i32,
+    pub stream_key_label: String,
     pub frame_channel: async_broadcast::Sender<Arc<VideoFrame>>,
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     let listener = TcpListener::bind("0.0.0.0:8123").await?;
+    let db = Database::connect("sqlite://./db.sqlite?mode=rwc")
+        .await
+        .unwrap();
+    Migrator::up(&db, None).await.unwrap();
     let appstate = Arc::new(Mutex::new(AppState {
         stream_sessions: Arc::new(DashMap::new()),
     }));
-    let (offer_tx, offer_rx) = tokio::sync::mpsc::channel::<(String, String)>(4);
-    let (answer_tx, answer_rx) = tokio::sync::mpsc::channel::<(String, String)>(4);
+    let (offer_tx, offer_rx) = tokio::sync::mpsc::channel::<(i32, i32, String)>(4);
+    // Request_Id,
+    // String_Label,
+    // Offer_body
+    // --
+    // We are using broadcast instead of a regular channel because the way answers are processed are
+    // mulithreaded and we could easily process the answer of a completely different offer request
+    let (answer_tx, answer_rx) = broadcast::<(i32, Option<String>)>(4);
+    // Request_Id,
+    // Answer_body
 
-    let mut http = HttpServer {
+    let http = HttpServer {
         offer_tx,
         accept_rx: answer_rx,
         appstate: appstate.clone(),
+        request_count: Mutex::new(0),
+        db: db.clone(),
     };
     http.start()?;
 
@@ -52,6 +70,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         offer_rx,
         accept_tx: answer_tx,
         sessions_ref: app.stream_sessions.clone(),
+        db: db.clone(),
     };
     webrtc.start()?;
     drop(app);
@@ -59,6 +78,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     loop {
         let (stream, _) = listener.accept().await?;
         let appstate = appstate.clone();
+        let db = db.clone();
 
         tokio::spawn(async move {
             let mut stream = stream;
@@ -108,22 +128,45 @@ async fn main() -> Result<(), Box<dyn Error>> {
                         }
                         ServerSessionResult::RaisedEvent(x) => match x {
                             ServerSessionEvent::PublishStreamFinished { stream_key, .. } => {
-                                appstate.lock().await.stream_sessions.remove(&stream_key);
+                                let key = entity::stream_key::Entity::find_by_key(&db, &stream_key)
+                                    .await
+                                    .unwrap()
+                                    .unwrap();
+                                appstate.lock().await.stream_sessions.remove(&key.id);
                             }
                             ServerSessionEvent::PublishStreamRequested {
                                 request_id,
                                 stream_key,
                                 ..
                             } => {
+                                let key =
+                                    entity::stream_key::Entity::find_by_key(&db, &stream_key).await;
+
+                                let key = if let Ok(Some(key)) = key {
+                                    key
+                                } else {
+                                    // Early return, gives error to client
+                                    let reply = rtmp_session
+                                        .reject_request(request_id, "", "Stream key invalid")
+                                        .unwrap();
+                                    for x in reply {
+                                        if let ServerSessionResult::OutboundResponse(y) = x {
+                                            stream.write_all(&y.bytes).await.unwrap();
+                                        }
+                                    }
+                                    break;
+                                };
+
                                 let session = StreamSession {
-                                    stream_key: stream_key.clone(),
+                                    stream_key_id: key.id,
+                                    stream_key_label: key.label,
                                     frame_channel: video_channel.clone(),
                                 };
                                 appstate
                                     .lock()
                                     .await
                                     .stream_sessions
-                                    .insert(stream_key.clone(), session);
+                                    .insert(key.id, session);
 
                                 let reply = rtmp_session.accept_request(request_id).unwrap();
                                 for x in reply {

crates/server/src/webrtc.rs

@@ -1,13 +1,15 @@
 use dashmap::DashMap;
-use tokio::{
-    net::UdpSocket,
-    sync::mpsc::{Receiver, Sender},
-};
+use entity::stream_key;
+use sea_orm::DatabaseConnection;
 use std::{
     error::Error,
     sync::Arc,
     time::{Duration, Instant},
 };
+use tokio::{
+    net::UdpSocket,
+    sync::mpsc::{Receiver, Sender},
+};
 
 use str0m::{
     Candidate, Event, Input, Output, Rtc,
@@ -19,16 +21,28 @@ use str0m::{
 use crate::{StreamSession, media::VideoFrame};
 
 pub struct Webrtc {
-    pub offer_rx: Receiver<(String, String)>,
-    pub accept_tx: Sender<(String, String)>,
-    pub sessions_ref: Arc<DashMap<String, StreamSession>>,
+    pub offer_rx: Receiver<(i32, i32, String)>,
+    pub accept_tx: async_broadcast::Sender<(i32, Option<String>)>,
+    pub sessions_ref: Arc<DashMap<i32, StreamSession>>,
+    pub db: DatabaseConnection,
 }
 
+const PER_CLIENT_CONNECTION_BUF: usize = 65535;
+
 impl Webrtc {
     pub fn start(mut self) -> Result<(), Box<dyn Error>> {
         tokio::spawn(async move {
             while let Some(offer) = self.offer_rx.recv().await {
-                let (stream_key, sdp_body) = offer;
+                let (request_id, stream_id, sdp_body) = offer;
+
+                let stream_key =
+                    stream_key::Entity::find_by_key(&self.db, &stream_id.to_string()).await;
+
+                if let Ok(None) = stream_key {
+                    self.accept_tx.broadcast((request_id, None)).await.unwrap();
+                    continue;
+                }
+
                 let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
                 let local_addr = socket.local_addr().unwrap();
 
@@ -50,7 +64,7 @@ impl Webrtc {
                 let mid = changes.add_media(
                     MediaKind::Video,
                     str0m::media::Direction::SendOnly,
-                    Some(stream_key.clone()),
+                    Some(stream_id.to_string()),
                     Some("video0".to_string()),
                     None,
                 );
@@ -64,13 +78,13 @@ impl Webrtc {
                 let answer_sdp = offer_answer.to_sdp_string();
 
                 self.accept_tx
-                    .send((stream_key.clone(), answer_sdp))
+                    .broadcast((request_id, Some(answer_sdp)))
                     .await
                     .unwrap();
 
                 let sessions_ref = self.sessions_ref.clone();
                 tokio::spawn(async move {
-                    Webrtc::detach_connection(socket, rtc, sessions_ref, mid).await;
+                    Webrtc::detach_connection(socket, rtc, sessions_ref, stream_id, mid).await;
                 });
             }
         });
@@ -80,7 +94,8 @@ impl Webrtc {
     async fn detach_connection(
         socket: UdpSocket,
         mut rtc: Rtc,
-        sessions_ref: Arc<DashMap<String, StreamSession>>,
+        sessions_ref: Arc<DashMap<i32, StreamSession>>,
+        stream_id: i32,
         _hint_mid: Mid,
     ) {
         let mut video_mid: Option<Mid> = None;
@@ -88,7 +103,7 @@ impl Webrtc {
         let mut connected = false;
         let mut video_stream: Option<async_broadcast::Receiver<Arc<VideoFrame>>> = None;
 
-        let mut recv_buf = vec![0u8; 65535];
+        let mut recv_buf = vec![0u8; PER_CLIENT_CONNECTION_BUF];
         let local_addr = socket.local_addr().unwrap();
 
         loop {
@@ -104,11 +119,9 @@ impl Webrtc {
                         Event::MediaAdded(ma) => {
                             if ma.kind == MediaKind::Video {
                                 if let Some(writer) = rtc.writer(ma.mid) {
-                                    let best = writer
-                                        .payload_params()
-                                        .max_by_key(|p| {
-                                            p.spec().format.profile_level_id.unwrap_or(0)
-                                        });
+                                    let best = writer.payload_params().max_by_key(|p| {
+                                        p.spec().format.profile_level_id.unwrap_or(0)
+                                    });
                                     if let Some(params) = best {
                                         println!("Selected PT {:?}", params.pt());
                                         video_pt = Some(params.pt());
@@ -132,7 +145,7 @@ impl Webrtc {
 
             if connected {
                 if video_stream.is_none() {
-                    if let Some(session) = sessions_ref.get("test") {
+                    if let Some(session) = sessions_ref.get(&stream_id) {
                         video_stream = Some(session.frame_channel.new_receiver());
                     }
                 }
@@ -141,7 +154,8 @@ impl Webrtc {
                         match stream.try_recv() {
                             Ok(frame) => {
                                 let now = Instant::now();
-                                let rtp_time = MediaTime::from_90khz(frame.timestamp_ms as u64 * 90);
+                                let rtp_time =
+                                    MediaTime::from_90khz(frame.timestamp_ms as u64 * 90);
                                 if let (Some(pt), Some(writer)) =
                                     (video_pt, video_mid.and_then(|m| rtc.writer(m)))
                                 {
@@ -160,7 +174,9 @@ impl Webrtc {
                 }
             }
 
-            let wait_until = deadline.min(Instant::now() + Duration::from_millis(20)).max(Instant::now());
+            let wait_until = deadline
+                .min(Instant::now() + Duration::from_millis(20))
+                .max(Instant::now());
             let sleep = tokio::time::sleep_until(wait_until.into());
 
             tokio::select! {