char/live.umm.gay

git clone https://git.t4t.associates/char/live.umm.gay

videogame hacker[wish-server-rs] Check stream keys against databasebcf019e

main
branchesmainall refs →
4.2 KiB143 linesraw
1use std::sync::Arc;
2
3use axum::{
4    extract::State,
5    http::{header, HeaderMap, Method, StatusCode},
6    response::{IntoResponse, Response},
7};
8use miette::{miette, Context, IntoDiagnostic, Result};
9use tracing::instrument;
10
11use crate::{
12    streams::{get_ongoing_stream, OngoingStream},
13    util::log_http_error,
14    AppState,
15};
16
17use webrtc::{
18    api::API as WebRTC,
19    peer_connection::{sdp::session_description::RTCSessionDescription, RTCPeerConnection},
20    track::track_local::TrackLocalWriter,
21};
22
23use super::create_rtc_config;
24
25async fn setup_whip_connection(
26    channel: &str,
27    webrtc: Arc<WebRTC>,
28) -> Result<Arc<RTCPeerConnection>> {
29    let rtc_config = create_rtc_config();
30
31    let peer_connection = Arc::new(
32        webrtc
33            .new_peer_connection(rtc_config)
34            .await
35            .into_diagnostic()?,
36    );
37
38    let OngoingStream {
39        video_track,
40        audio_track,
41        ..
42    } = get_ongoing_stream(channel).await?;
43
44    peer_connection.on_track(Box::new(move |track, _recv, _tx| {
45        let local_track = if track.codec().capability.mime_type.starts_with("audio/") {
46            &audio_track
47        } else {
48            &video_track
49        }
50        .clone();
51
52        tokio::spawn(async move {
53            let mut rtp_buf = vec![0u8; 1500];
54            while let Ok((rtp_read, _)) = track.read(&mut rtp_buf).await {
55                let (Ok(_) | Err(webrtc::Error::ErrClosedPipe)) = local_track.write(&rtp_buf[..rtp_read]).await else { break };
56            }
57            Result::<()>::Ok(())
58        });
59
60        Box::pin(async {})
61    }));
62
63    Ok(peer_connection)
64}
65
66#[instrument(skip_all)]
67#[axum::debug_handler]
68pub async fn handle_whip(
69    method: Method,
70    headers: HeaderMap,
71    State(app): State<AppState>,
72    offer: String,
73) -> Response {
74    if method != Method::POST {
75        return (StatusCode::METHOD_NOT_ALLOWED, "Please use POST!").into_response();
76    }
77
78    let Some(auth) = headers.get(header::AUTHORIZATION) else { return (StatusCode::UNAUTHORIZED, "Authorization was not set").into_response() };
79    let auth = auth
80        .to_str()
81        .into_diagnostic()
82        .wrap_err("Failed to decode auth header")
83        .unwrap();
84    let auth = auth.strip_prefix("Bearer ").unwrap_or(auth);
85
86    let Some((channel, key)) = auth.split_once(':') else { return (StatusCode::UNAUTHORIZED, "Invalid Authorization header").into_response() };
87
88    let mut db = app
89        .db
90        .acquire()
91        .await
92        .expect("SQLite connections never fail");
93    match sqlx::query!(
94        "SELECT stream FROM streams WHERE stream = ? AND password = ?",
95        channel,
96        key
97    )
98    .fetch_one(&mut db)
99    .await
100    {
101        Err(sqlx::Error::RowNotFound) => {
102            return log_http_error(StatusCode::UNAUTHORIZED, "Invalid stream key.")
103        }
104        Err(e) => return log_http_error(StatusCode::INTERNAL_SERVER_ERROR, e),
105        _ => {}
106    };
107
108    let peer_connection = match setup_whip_connection(channel, app.webrtc)
109        .await
110        .wrap_err("Failed to initialize peer connection")
111    {
112        Ok(peer_connection) => peer_connection,
113        Err(e) => {
114            return log_http_error(StatusCode::INTERNAL_SERVER_ERROR, e);
115        }
116    };
117
118    let Ok(description) = RTCSessionDescription::offer(offer) else { return log_http_error(StatusCode::BAD_REQUEST, "Malformed SDP offer") };
119    if let Err(e) = peer_connection.set_remote_description(description).await {
120        return log_http_error(StatusCode::INTERNAL_SERVER_ERROR, e);
121    };
122
123    let mut gather_complete = peer_connection.gathering_complete_promise().await;
124
125    let answer = match peer_connection.create_answer(None).await {
126        Ok(answer) => answer,
127        Err(e) => return log_http_error(StatusCode::INTERNAL_SERVER_ERROR, e),
128    };
129
130    if let Err(e) = peer_connection.set_local_description(answer).await {
131        return log_http_error(StatusCode::INTERNAL_SERVER_ERROR, e);
132    }
133
134    let _ = gather_complete.recv().await;
135
136    match peer_connection.local_description().await {
137        Some(desc) => (StatusCode::CREATED, desc.sdp).into_response(),
138        None => log_http_error(
139            StatusCode::INTERNAL_SERVER_ERROR,
140            miette!("No local description exists!"),
141        ),
142    }
143}