Tahinli/rust_webrtc
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: ✨ token extractor middleware

Browse source
This commit is contained in:
Ahmet Kaan Gümüş 2025-04-16 06:27:08 +03:00
parent 96199f71ef
commit 4f874d8789
9 changed files with 247 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

3
client/Cargo.toml
View file

@ -21,8 +21,11 @@ web-sys = { version = "0.3.77", features = [
"MediaTrackConstraintSet",
"Navigator",
"RtcConfiguration",
"RtcIceCandidate",
"RtcIceCandidateInit",
"RtcIceServer",
"RtcPeerConnection",
"RtcPeerConnectionState",
"RtcSdpType",
"RtcSessionDescription",
"RtcSessionDescriptionInit",

23
client/src/rtc.rs
View file

@ -48,18 +48,25 @@ pub async fn offer(username: String) {
if let Err(err_val) = send_offer(&username, &data).await {
log!("Error: Send Offer | {}", err_val)
}
for _ in 0..10 {
match receive_answer(&username).await {
Ok(received_answer) => {
log!("{:#?}", received_answer);
Ok(received_user_and_signal_answer) => {
log!("{:#?}", received_user_and_signal_answer);
let peer_connection_session_answer =
RtcSessionDescriptionInit::new(RtcSdpType::Answer);
peer_connection_session_answer.set_sdp(received_answer.get_data().as_str());
peer_connection_session_answer
.set_sdp(received_user_and_signal_answer.signal.get_data().as_str());
JsFuture::from(
peer_connection.set_remote_description(&peer_connection_session_answer),
)
.await
.unwrap();
for _ in 0..100 {
log!("{:#?}", peer_connection.connection_state());
sleep(1000).await;
}
break;
}
Err(err_val) => log!("Error: Receive Answer | {}", err_val),
@ -71,11 +78,12 @@ pub async fn offer(username: String) {
pub async fn answer(username: String) {
for _ in 0..10 {
match receive_offer(&username).await {
Ok(offer) => {
Ok(received_user_and_signal_offer) => {
let peer_connection = create_peer_connection_with_configuration().await;
let peer_connection_session_offer =
RtcSessionDescriptionInit::new(RtcSdpType::Offer);
peer_connection_session_offer.set_sdp(offer.get_data().as_str());
peer_connection_session_offer
.set_sdp(received_user_and_signal_offer.signal.get_data().as_str());
JsFuture::from(
peer_connection.set_remote_description(&peer_connection_session_offer),
)
@ -99,6 +107,11 @@ pub async fn answer(username: String) {
let data = session_answer;
send_answer(&username, &data).await.unwrap();
for _ in 0..100 {
log!("{:#?}", peer_connection.connection_state());
sleep(1000).await;
}
break;
}
Err(err_val) => log!("Error: Receive Offer | {}", err_val),

50
client/src/signal.rs
View file

@ -1,11 +1,11 @@
use std::sync::LazyLock;
use leptos::logging::log;
use protocol::{Signal, SignalType};
use protocol::{Signal, SignalType, UserAndSignal};
use reqwest::{Response, header::HeaderMap};
use serde_json::{Value, json};
const SIGNALLING_ADDRESS: &str = "http://127.0.0.1:4546";
const SIGNALLING_ADDRESS: &str = "http://192.168.1.3:4546";
static REQUEST_CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| reqwest::Client::new());
async fn create_headers(headers: Vec<(&'static str, String)>) -> HeaderMap {
@ -58,15 +58,17 @@ pub async fn send_offer(username: &String, data: &String) -> Result<(), reqwest:
.map(|_| Ok(()))?
}
pub async fn receive_offer(username: &String) -> Result<Signal, Box<dyn std::error::Error>> {
pub async fn receive_offer(username: &String) -> Result<UserAndSignal, Box<dyn std::error::Error>> {
let result = get_json(username, SignalType::Offer)
.await
.map(async |response| response.json::<Signal>().await)?
.map(async |response| response.json::<UserAndSignal>().await)?
.await?;
if result.get_signal_type() == SignalType::Offer {
if result.signal.get_signal_type() == SignalType::Offer {
Ok(result)
} else {
Err(protocol::Error::SignalType(result.get_signal_type()))?
Err(protocol::Error::UnexpectedSignalType(
result.signal.get_signal_type(),
))?
}
}
@ -78,14 +80,42 @@ pub async fn send_answer(username: &String, data: &String) -> Result<(), reqwest
.map(|_| Ok(()))?
}
pub async fn receive_answer(username: &String) -> Result<Signal, Box<dyn std::error::Error>> {
pub async fn receive_answer(
username: &String,
) -> Result<UserAndSignal, Box<dyn std::error::Error>> {
let result = get_json(username, SignalType::Answer)
.await
.map(async |response| response.json::<Signal>().await)?
.map(async |response| response.json::<UserAndSignal>().await)?
.await?;
if result.get_signal_type() == SignalType::Answer {
if result.signal.get_signal_type() == SignalType::Answer {
Ok(result)
} else {
Err(protocol::Error::SignalType(result.get_signal_type()))?
Err(protocol::Error::UnexpectedSignalType(
result.signal.get_signal_type(),
))?
}
}
pub async fn send_ice_candidate(username: &String, data: &String) -> Result<(), reqwest::Error> {
let rtc_session_answer_signal = Signal::new(&SignalType::ICECandidate, data);
let rtc_session_answer_signal = json!(rtc_session_answer_signal);
post_json(username, &rtc_session_answer_signal)
.await
.map(|_| Ok(()))?
}
pub async fn receive_ice_candidate(
username: &String,
) -> Result<UserAndSignal, Box<dyn std::error::Error>> {
let result = get_json(username, SignalType::Answer)
.await
.map(async |response| response.json::<UserAndSignal>().await)?
.await?;
if result.signal.get_signal_type() == SignalType::ICECandidate {
Ok(result)
} else {
Err(protocol::Error::UnexpectedSignalType(
result.signal.get_signal_type(),
))?
}
}

47
protocol/src/lib.rs
View file

@ -1,11 +1,28 @@
use std::fmt::Display;
use std::{fmt::Display, str::FromStr};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UserAndSignal {
pub user: User,
pub signal: Signal,
}
impl UserAndSignal {
pub async fn new(user: User, signal: Signal) -> Self {
UserAndSignal { user, signal }
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct User {
pub username: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum Error {
SignalType(SignalType),
UnexpectedSignalType(SignalType),
InvalidSignalType(String),
}
impl std::error::Error for Error {
@ -17,8 +34,11 @@ impl std::error::Error for Error {
impl Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Error::SignalType(signal_type) => {
write!(f, "Not Expected Signal Type: {}", signal_type)
Error::UnexpectedSignalType(signal_type) => {
write!(f, "Unexpected Signal Type: {}", signal_type)
}
Error::InvalidSignalType(invalid_signal_type) => {
write!(f, "Invalid Signal Type: {}", invalid_signal_type)
}
}
}
@ -29,9 +49,23 @@ pub enum SignalType {
Auth,
Offer,
Answer,
ICECandidate,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
impl FromStr for SignalType {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"Auth" => Ok(SignalType::Auth),
"Offer" => Ok(SignalType::Offer),
"Answer" => Ok(SignalType::Answer),
_ => Err(Error::InvalidSignalType(s.to_owned())),
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Signal {
signal_type: SignalType,
data: String,
@ -44,6 +78,7 @@ impl Display for SignalType {
SignalType::Auth => write!(f, "Auth"),
SignalType::Offer => write!(f, "Offer"),
SignalType::Answer => write!(f, "Answer"),
SignalType::ICECandidate => write!(f, "ICE Candidate"),
}
}
}

4
server/Cargo.toml
View file

@ -7,7 +7,11 @@ edition = "2024"
axum = { version = "0.8.3", features = ["json"] }
axum-macros = "0.5.0"
tokio = "1.42.1"
tower-http = { version = "0.6.2", features = ["cors", "trace"] }
tracing-subscriber = "0.3.19"
tracing = "0.1.41"
webrtc = "0.12.0"
serde = { workspace = true }
serde_json = { workspace = true }
chrono = { workspace = true }
protocol = { path = "../protocol" }

1
server/src/lib.rs
View file

@ -2,6 +2,7 @@ use std::sync::LazyLock;
use utils::naive_toml_parser;
mod middleware;
pub mod signal;
pub mod utils;

7
server/src/main.rs
View file

@ -1,8 +1,11 @@
use rust_communication_server::signal::start_signalling;
use tracing::Level;
#[tokio::main]
async fn main() {
println!("Hello, world!");
tokio::spawn(start_signalling()).await.unwrap();
tracing_subscriber::fmt()
.with_max_level(Level::TRACE)
.init();
start_signalling().await;
}

83
server/src/middleware.rs Normal file
View file

@ -0,0 +1,83 @@
use std::{str::FromStr, sync::Arc};
use axum::{
extract::Request,
http::{self, HeaderMap, StatusCode},
middleware::Next,
response::IntoResponse,
};
use protocol::{SignalType, User};
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize)]
pub struct UserAndExpectedSignal {
pub user: User,
pub expected_signal: SignalType,
}
async fn extract_user_from_authorization_header(headers: &HeaderMap) -> Option<User> {
if let Some(authorization_header) = headers.get(http::header::AUTHORIZATION) {
dbg!(authorization_header);
if let Ok(authorization_header) = authorization_header.to_str() {
if let Some((bearer, authorization_token)) = authorization_header.split_once(' ') {
dbg!(
"Info: Verify | Http Header | {} || {}",
bearer,
authorization_token
);
if bearer.to_lowercase() == "bearer" {
let user = User {
username: authorization_token.to_string(),
};
return Some(user);
}
}
}
}
None
}
pub async fn verify_then_get_user(mut request: Request, next: Next) -> impl IntoResponse {
let headers = request.headers();
dbg!(headers);
if let Some(user) = extract_user_from_authorization_header(headers).await {
let user = Arc::new(user);
request.extensions_mut().insert(user);
return next.run(request).await;
}
StatusCode::FORBIDDEN.into_response()
}
pub async fn verify_then_get_user_and_expected_signal(
mut request: Request,
next: Next,
) -> impl IntoResponse {
let headers = request.headers();
if let Some(user) = extract_user_from_authorization_header(headers).await {
if let Ok(expected_signal) = headers.get("EXPECTED_SIGNAL").unwrap().to_str() {
match SignalType::from_str(expected_signal) {
Ok(expected_signal) => {
let user_and_expected_signal = UserAndExpectedSignal {
user,
expected_signal,
};
let user_and_expected_signal = Arc::new(user_and_expected_signal);
request.extensions_mut().insert(user_and_expected_signal);
next.run(request).await
}
Err(err_val) => {
eprintln!(
"Error: Verify and Get Expected Signal | Signal Type Conversion | {}",
err_val
);
StatusCode::BAD_REQUEST.into_response()
}
}
} else {
StatusCode::BAD_REQUEST.into_response()
}
} else {
StatusCode::FORBIDDEN.into_response()
}
}

61
server/src/signal.rs
View file

@ -1,28 +1,45 @@
use std::sync::{Arc, LazyLock, RwLock};
use axum::{
Json, Router,
Extension, Json, Router,
http::StatusCode,
response::IntoResponse,
routing::{get, post},
};
use axum_macros::debug_handler;
use protocol::Signal;
use protocol::{Signal, User, UserAndSignal};
use tokio::net::TcpListener;
use tower_http::{cors::CorsLayer, trace::TraceLayer};
static SIGNALS: LazyLock<Arc<RwLock<Vec<Signal>>>> =
LazyLock::new(|| Arc::new(RwLock::new(vec![])));
use crate::middleware::{
UserAndExpectedSignal, verify_then_get_user, verify_then_get_user_and_expected_signal,
};
static USERS_AND_SIGNALS: LazyLock<RwLock<Vec<UserAndSignal>>> =
LazyLock::new(|| RwLock::new(vec![]));
pub async fn start_signalling() {
let route = route();
let listener = TcpListener::bind("0.0.0.0:4546").await.unwrap();
let route = route()
.layer(CorsLayer::permissive())
.layer(TraceLayer::new_for_http());
let listener = TcpListener::bind("192.168.1.3:4546").await.unwrap();
println!("http://192.168.1.3:4546");
axum::serve(listener, route).await.unwrap();
}
fn route() -> Router {
Router::new()
.route("/alive", get(alive))
.route("/", post(signal))
.route(
"/",
get(read_signal).route_layer(axum::middleware::from_fn(
verify_then_get_user_and_expected_signal,
)),
)
.route(
"/",
post(create_signal).route_layer(axum::middleware::from_fn(verify_then_get_user)),
)
}
async fn alive() -> impl IntoResponse {
@ -30,7 +47,33 @@ async fn alive() -> impl IntoResponse {
}
#[debug_handler]
async fn signal(Json(signal): Json<Signal>) -> impl IntoResponse {
SIGNALS.write().unwrap().push(signal);
async fn create_signal(
Extension(user): Extension<Arc<User>>,
Json(signal): Json<Signal>,
) -> impl IntoResponse {
let user = (*user).clone();
let user_and_signal = UserAndSignal::new(user, signal).await;
USERS_AND_SIGNALS.write().unwrap().push(user_and_signal);
StatusCode::OK
}
async fn read_signal(
Extension(user_and_expected_signal): Extension<Arc<UserAndExpectedSignal>>,
) -> impl IntoResponse {
let mut target_index = None;
let mut json_body = serde_json::json!("");
for (index, user_and_signal) in USERS_AND_SIGNALS.read().unwrap().iter().enumerate() {
if user_and_signal.signal.get_signal_type() == user_and_expected_signal.expected_signal {
json_body = serde_json::json!(user_and_signal);
target_index = Some(index);
}
}
match target_index {
Some(target_index) => {
USERS_AND_SIGNALS.write().unwrap().remove(target_index);
(StatusCode::OK, Json(json_body)).into_response()
}
None => StatusCode::BAD_REQUEST.into_response(),
}
}