diff --git a/Cargo.lock b/Cargo.lock index 4b58524..ba72ade 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -390,6 +390,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8" dependencies = [ "axum-core", + "base64 0.22.1", "bytes", "form_urlencoded", "futures-util", @@ -408,8 +409,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", diff --git a/Cargo.toml b/Cargo.toml index 22121da..f89a8a4 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,7 +39,7 @@ async-tungstenite = { version = "0.32.0", features = [ "verbose-logging", "webpki-roots", ] } -axum = "0.8.8" +axum = { version = "0.8.8", features = [ "ws" ] } base64 = "0.22.1" bytes = "1.11.1" color-eyre = "0.6.5" diff --git a/src/server/api.rs b/src/server/api.rs index 35ee9ec..ae30a24 100644 --- a/src/server/api.rs +++ b/src/server/api.rs @@ -10,14 +10,12 @@ use crate::{ }; use axum::http::HeaderValue; use base64::Engine as _; -use http_body_util::{Full, StreamBody}; -use hyper::body::{Body, Bytes, Frame}; +use http_body_util::Full; +use hyper::body::Bytes; use hyper::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE}; use hyper::{HeaderMap, Response as HttpResponse, StatusCode}; use json::JsonValue; use json::number::Number; -use tokio::fs::File; -use tokio_util::io::ReaderStream; pub async fn handle( path: &str, diff --git a/src/server/omikron_connection.rs b/src/server/omikron_connection.rs index 7ce8f8b..3eeb22b 100644 --- a/src/server/omikron_connection.rs +++ b/src/server/omikron_connection.rs @@ -6,29 +6,26 @@ use crate::sql::sql::{self, get_by_user_id, get_by_username, get_iota_by_id, get use crate::sql::user_online_tracker::{self}; use crate::util::crypto_helper::encrypt; use crate::util::logger::PrintType; -use crate::{get_private_key, get_public_key, log_in, log_out}; +use crate::{get_private_key, get_public_key, log_err, log_in, log_out}; +use axum::extract::ws::WebSocket; +use axum::extract::ws::{Message, Utf8Bytes}; use base64::{Engine as _, engine::general_purpose::STANDARD}; use dashmap::DashMap; -use futures::SinkExt; use futures::stream::SplitSink; use futures::stream::SplitStream; -use hyper::upgrade::Upgraded; -use hyper_util::rt::TokioIo; +use futures_util::SinkExt; use json::JsonValue; use json::number::Number; use rand::Rng; use rand::distributions::Alphanumeric; use std::sync::Arc; use tokio::sync::RwLock; -use tokio_tungstenite::WebSocketStream; -use tungstenite::Message; -use tungstenite::Utf8Bytes; use uuid::Uuid; use x448::PublicKey; pub struct OmikronConnection { - pub sender: Arc>, Message>>>, - pub receiver: Arc>>>>, + pub sender: Arc>>, + pub receiver: Arc>>, pub omikron_id: Arc>, pub pub_key: Arc>>>, identified: Arc>, @@ -43,8 +40,8 @@ pub struct OmikronConnection { impl OmikronConnection { pub fn new( - sender: SplitSink>, Message>, - receiver: SplitStream>>, + sender: SplitSink, + receiver: SplitStream, ) -> Arc { Arc::new(Self { sender: Arc::new(RwLock::new(sender)), @@ -66,10 +63,17 @@ impl OmikronConnection { *self.omikron_id.read().await, PrintType::Omikron, "{}", - message_text + cv.to_json().to_string() + ); + } + if let Err(e) = sender.send(message_text).await { + log_err!( + *self.omikron_id.read().await, + PrintType::Omikron, + "WebSocket send error: {}", + e ); } - let _ = sender.send(message_text).await; } pub async fn get_omikron_id(&self) -> i64 { *self.omikron_id.read().await @@ -77,9 +81,6 @@ impl OmikronConnection { pub async fn is_identified(&self) -> bool { *self.identified.read().await && *self.challenged.read().await } - pub async fn get_public_key(&self) -> PublicKey { - PublicKey::from_bytes(self.pub_key.read().await.as_ref().unwrap()).unwrap() - } pub async fn handle_message(self: Arc, message: String) { let cv = CommunicationValue::from_json(&message); @@ -619,7 +620,6 @@ impl OmikronConnection { if let Some(public_key) = cv.get_data(DataTypes::public_key).and_then(|v| v.as_str()) { if let Some(iota_id) = iota_id_opt { - // Existing logic to update iota match sql::register_complete_iota(iota_id, public_key.to_string()).await { Ok(_) => { let response = CommunicationValue::new(CommunicationType::success) @@ -792,7 +792,7 @@ impl OmikronConnection { ) { match sql::get_by_user_id(user_id).await { Ok(user) => { - let current_token = user.11; // token is the 12th element (index 11) + let current_token = user.11; if current_token == reset_token { let mut success = true; let mut error_message = String::new(); @@ -824,7 +824,6 @@ impl OmikronConnection { } else { self.send_error_response( &cv.get_id(), - // Using this for invalid token CommunicationType::error_invalid_challenge, ) .await; diff --git a/src/server/omikron_manager.rs b/src/server/omikron_manager.rs index 8440835..864c9d6 100644 --- a/src/server/omikron_manager.rs +++ b/src/server/omikron_manager.rs @@ -24,11 +24,3 @@ pub async fn get_random_omikron() -> Result, ()> { return Err(()); } } - -pub async fn get_omikron(omikron_id: i64) -> Option> { - if let Some(omikron) = OMIKRON_CONNECTIONS.get(&omikron_id) { - Some(omikron.clone()) - } else { - None - } -} diff --git a/src/server/server.rs b/src/server/server.rs index 3926f08..95af459 100755 --- a/src/server/server.rs +++ b/src/server/server.rs @@ -1,486 +1,106 @@ -use base64::Engine; -use base64::engine::general_purpose::STANDARD; -use bytes::Bytes; -use futures::StreamExt; -use futures_util::TryFutureExt; -use http_body_util::BodyExt; -use http_body_util::Full; -use hyper::server::conn::http2; -use hyper::{Method, StatusCode}; -use hyper::{ - Request as HttpRequest, Response as HttpResponse, body::Incoming, server::conn::http1, upgrade, +use axum::{ + Router, + body::Body, + extract::{ConnectInfo, OriginalUri, Path, ws::WebSocketUpgrade}, + response::{IntoResponse, Redirect}, + routing::get, }; -use hyper_util::rt::TokioExecutor; -use hyper_util::rt::tokio::TokioIo; -use hyper_util::service::TowerToHyperService; + use pnet::datalink::NetworkInterface; -use rustls::ServerConfig; -use rustls::pki_types::{CertificateDer, PrivateKeyDer}; -use sha1::{Digest, Sha1}; -use std::error::Error; -use std::io::ErrorKind; -use std::io::{self, BufReader}; -use std::net::{IpAddr, SocketAddr}; -use std::result::Result::Ok; -use std::sync::Arc; -use std::{future::Future, pin::Pin, time::Duration}; +use std::net::SocketAddr; +use std::time::Duration; use tokio::net::TcpListener; -use tokio::sync::broadcast; -use tokio_rustls::TlsAcceptor; -use tower::Service; use crate::log; use crate::server::api; use crate::server::short_link::get_short_link; use crate::server::socket; -use crate::util::file_util::load_file_buf; -#[derive(Clone)] -struct HttpService { - peer_addr: SocketAddr, + +pub async fn start(port: u16) -> bool { + let app = Router::new() + .route("/ws/omikron", get(ws_handler)) + .route("/direct/{short}", get(direct_handler)) + .fallback(fallback_handler); + run_http_server(port, app).await } -impl Service> for HttpService { - type Response = HttpResponse>; - type Error = io::Error; - type Future = Pin> + Send>>; +async fn ws_handler( + ws: WebSocketUpgrade, + OriginalUri(uri): OriginalUri, + ConnectInfo(_): ConnectInfo, +) -> impl IntoResponse { + log!("Attempting WebSocket upgrade on {}", uri.path()); + let path = uri.path().to_string(); - fn poll_ready( - &mut self, - _cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - std::task::Poll::Ready(std::io::Result::Ok(())) - } + ws.on_upgrade(async move |socket| socket::handle(path, socket)) +} - fn call(&mut self, req: HttpRequest) -> Self::Future { - let peer_ip = self.peer_addr.ip(); - - let (parts, body) = req.into_parts(); - - let method = parts.method.clone(); - let path = parts.uri.path().to_string(); - let headers = parts.headers.clone(); - - let fut = async move { - let is_websocket_upgrade = path.starts_with("/ws") - && method == Method::GET - && headers - .get("connection") - .map(|v| v.to_str().unwrap_or("").contains("Upgrade")) - .unwrap_or(false) - && headers.get("upgrade").map(|v| v.to_str().unwrap_or("")) == Some("websocket"); - - if is_websocket_upgrade { - log!("Attempting WebSocket upgrade on /ws"); - - if let Some(sec_websocket_key) = headers.get("sec-websocket-key") { - let sec_websocket_key = sec_websocket_key.to_str().unwrap_or("").to_string(); - let sec_websocket_accept = calculate_accept_key(&sec_websocket_key); - - let response = HttpResponse::builder() - .status(StatusCode::SWITCHING_PROTOCOLS) - .header("Upgrade", "websocket") - .header("Connection", "Upgrade") - .header("Sec-WebSocket-Accept", sec_websocket_accept) - .body(Full::new(Bytes::from(""))) - .unwrap(); - let req_for_upgrade = HttpRequest::from_parts(parts, body); - let upgrades = upgrade::on(req_for_upgrade); - - socket::handle(path, upgrades); - Ok(response) - } else { - log!("No Sec-WebSocket-Key found in request headers"); - let response = HttpResponse::builder() - .status(StatusCode::BAD_REQUEST) - .body(Full::new(Bytes::from("Missing Sec-WebSocket-Key"))) - .unwrap(); - Ok(response) - } - } else if path.starts_with("/api") { - let whole_body = - tokio::time::timeout(Duration::from_secs(10), body.collect()).await??; - let bytes = whole_body.to_bytes(); - - let body_string: Option = match String::from_utf8(bytes.to_vec()) { - Ok(s) => Some(s), - Err(_) => None, - }; - - Ok(api::handle(&path, headers.clone(), body_string).await) - } else if path.starts_with("/direct") { - let short = path.replace("/direct/", ""); - if let Ok(long) = get_short_link(&short).await { - let response = HttpResponse::builder() - .status(StatusCode::FOUND) - .header("Location", long) - .body(Full::new(Bytes::from(""))) - .unwrap(); - Ok(response) - } else { - let response = HttpResponse::builder() - .status(StatusCode::NOT_FOUND) - .body(Full::new(Bytes::from("Short link not found"))) - .unwrap(); - Ok(response) - } - } else { - let whole_body = match body.collect().await { - Ok(collected) => collected, - Err(e) => { - log!("Error collecting body: {}", e); - return Ok(HttpResponse::builder() - .status(StatusCode::INTERNAL_SERVER_ERROR) - .body(Full::new(Bytes::from(format!( - "Failed to read body: {}", - e - )))) - .unwrap()); - } - }; - let bytes = whole_body.to_bytes(); - - let body_string: Option = match String::from_utf8(bytes.to_vec()) { - Ok(s) => Some(s), - Err(_) => None, - }; - - Ok(api::handle(&path, headers, body_string).await) - } - }; - - Box::pin(fut.map_err(|err: color_eyre::eyre::ErrReport| { - io::Error::new( - io::ErrorKind::Other, - format!("Error in request handling: {}", err), - ) - })) +async fn direct_handler(Path(short): Path) -> impl IntoResponse { + match get_short_link(&short).await { + Ok(long) => Redirect::temporary(&long), + Err(_) => Redirect::temporary("https://tensamin.net"), } } -pub fn is_local_network(addr: IpAddr) -> bool { - match addr { - IpAddr::V4(v4) => { - let octets = v4.octets(); - if octets[0] == 10 { - return true; - } - if octets[0] == 172 && (16..=31).contains(&octets[1]) { - return true; - } - if octets[0] == 192 && octets[1] == 168 { - return true; - } - if octets[0] == 127 { - return true; - } - if octets[0] == 169 && octets[1] == 254 { - return true; - } +async fn fallback_handler( + OriginalUri(uri): OriginalUri, + headers: axum::http::HeaderMap, + body: Body, +) -> impl IntoResponse { + let path = uri.path().to_string(); - false - } + let whole_body = tokio::time::timeout( + Duration::from_secs(10), + axum::body::to_bytes(body, 1024 * 1024 * 10), + ) + .await; - IpAddr::V6(v6) => { - let segments = v6.segments(); - if (segments[0] & 0xfe00) == 0xfc00 { - return true; - } - if (segments[0] & 0xffc0) == 0xfe80 { - return true; - } - if v6.is_loopback() { - return true; - } + let body_string = match whole_body { + Ok(Ok(bytes)) => String::from_utf8(bytes.to_vec()).ok(), + _ => None, + }; - false - } - } + api::handle(&path, headers, body_string).await } -async fn run_http_server(port: u16) -> bool { - let mut ip = "0.0.0.0".to_string(); - for iface in pnet::datalink::interfaces() { - let iface: NetworkInterface = iface; - if iface.ips.len() > 0 { - let ipsv = format!("{}", iface.ips[0]); - let ips: &str = ipsv.split('/').next().unwrap(); - if format!("{}", ips).starts_with("10.") || format!("{}", ips).starts_with("192.") { - ip = ips.to_string(); - } +async fn run_http_server(port: u16, app: Router) -> bool { + let ip = find_local_ip(); + let listener = match TcpListener::bind(format!("0.0.0.0:{}", port)).await { + Ok(l) => l, + Err(e) => { + log!("Failed to bind to port {}: {:?}", port, e); + return false; } - } - let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await; - if let Err(e) = listener { - log!("Failed to bind to port {}: {:?}", port, e); - return false; - } - let listener = listener.unwrap(); + }; + log!( "Standard Server listening for HTTP and WS on {}:{}", ip, port ); - // Create a broadcast channel for graceful shutdown signal - let (shutdown_tx, _) = broadcast::channel::<()>(1); - - tokio::spawn(async move { - loop { - tokio::select! { - // Monitor for shutdown signal - _ = async { - loop { - tokio::time::sleep(Duration::from_millis(100)).await; - } - } => { - log!("Standard Server received shutdown signal."); - // Send kill signal to all active connection tasks - let _ = shutdown_tx.send(()); - break; - } - - // Accept new connections - accepted = listener.accept() => { - match accepted { - std::result::Result::Ok((stream, addr)) => { - let service = HttpService { peer_addr: addr }; - let io = TokioIo::new(stream); - - // Subscribe to the shutdown signal for this specific connection - let mut rx = shutdown_tx.subscribe(); - - tokio::spawn(async move { - use hyper::server::conn::http2; - - let conn = http2::Builder::new(TokioExecutor::new()) - .serve_connection(io, TowerToHyperService::new(service)); - - - // Wait for either the connection to finish naturally OR the shutdown signal - tokio::select! { - res = conn => { - if let Err(err) = res { - if let Some(io_err) = err.source().and_then(|e| e.downcast_ref::()) { - if io_err.kind() != io::ErrorKind::ConnectionReset - && io_err.kind() != io::ErrorKind::BrokenPipe - { - log!("Error serving connection: {:?}", err); - } - } - } - } - _ = rx.recv() => { - // Shutdown signal received. - // Dropping the 'conn' future here closes the socket immediately. - } - } - }); - } - Err(e) => { - log!("Error accepting connection: {:?}", e); - tokio::time::sleep(Duration::from_millis(500)).await; - } - } - } - } - } - - log!("Standard Server shutdown complete."); - }); - - true + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .map(|_| true) + .unwrap_or_else(|e| { + log!("Server error: {}", e); + false + }) } -/// Runs the encrypted HTTPS/WSS server loop using the provided TLS config. -async fn run_tls_server(port: u16, tls_config: Arc) -> bool { - let mut ip = "0.0.0.0".to_string(); +fn find_local_ip() -> String { for iface in pnet::datalink::interfaces() { let iface: NetworkInterface = iface; - let ipsv = format!("{}", iface.ips[0]); - let ips: &str = ipsv.split('/').next().unwrap(); - log!("{}", ips); - if format!("{}", ips).starts_with("10.") { - ip = ips.to_string(); - } - } - - let acceptor = TlsAcceptor::from(tls_config); - - let listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await; - if let Err(e) = listener { - log!("Failed to bind to port {}: {:?}", port, e); - return false; - } - let listener = listener.unwrap(); - log!( - "Encrypted Server listening for HTTPS and WSS on {}:{}", - ip, - port - ); - - // Create a broadcast channel for graceful shutdown signal - let (shutdown_tx, _) = broadcast::channel::<()>(1); - - tokio::spawn(async move { - loop { - tokio::select! { - // Monitor for shutdown signal - _ = async { - loop { - tokio::time::sleep(Duration::from_millis(100)).await; - } - } => { - log!("Encrypted Server received shutdown signal."); - // Send kill signal to all active connection tasks - let _ = shutdown_tx.send(()); - break; - } - - // Accept new connections - accepted = listener.accept() => { - match accepted { - std::result::Result::Ok((stream, addr)) => { - let service = HttpService { peer_addr: addr }; - let acceptor = acceptor.clone(); - - // Subscribe to the shutdown signal for this specific connection - let mut rx = shutdown_tx.subscribe(); - - tokio::spawn(async move { - // Perform TLS handshake - let tls_stream = match acceptor.accept(stream).await { - Ok(s) => s, - Err(e) => { - if e.kind() != io::ErrorKind::Interrupted { - log!("TLS Handshake error: {:?}", e); - } - return; - } - }; - let io = TokioIo::new(tls_stream); - - // Prepare connection future - let conn = http1::Builder::new() - .preserve_header_case(true) - .title_case_headers(true) - .serve_connection(io, TowerToHyperService::new(service)) - .with_upgrades(); - - // Wait for either the connection to finish naturally OR the shutdown signal - tokio::select! { - res = conn => { - if let Err(err) = res { - if let Some(io_err) = err.source().and_then(|e| e.downcast_ref::()) { - if io_err.kind() != io::ErrorKind::ConnectionReset - && io_err.kind() != io::ErrorKind::BrokenPipe - { - log!("Error serving connection: {:?}", err); - } - } - } - } - _ = rx.recv() => { - // Shutdown signal received. - // Dropping the 'conn' future here closes the socket immediately. - } - } - }); - } - Err(e) => { - log!("Error accepting connection: {:?}", e); - tokio::time::sleep(Duration::from_millis(500)).await; - } - } - } + if !iface.ips.is_empty() { + let ipsv = format!("{}", iface.ips[0]); + let ips: &str = ipsv.split('/').next().unwrap(); + if ips.starts_with("10.") || ips.starts_with("192.") { + return ips.to_string(); } } - - log!("Encrypted Server shutdown complete."); - }); - true -} - -pub async fn start(port: u16) -> bool { - let tls_result = load_tls_config(); - - match tls_result { - Ok(Some(tls_config)) => run_tls_server(port, tls_config).await, - Ok(_) => run_http_server(port).await, - Err(e) => { - log!("Fatal error during TLS config load: {}", e); - false - } } -} -fn calculate_accept_key(key: &str) -> String { - let websocket_guid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; - let mut sha1 = Sha1::new(); - sha1.update(key.as_bytes()); - sha1.update(websocket_guid.as_bytes()); - let result = sha1.finalize(); - STANDARD.encode(result) // Base64 encode the result -} - -/// Loads TLS config. Returns Ok(None) if cert files are not found, and an error if parsing fails. -fn load_tls_config() -> Result>, Box> { - let cert_file_res = load_file_buf("certs", "cert.pem"); - let key_file_res = load_file_buf("certs", "cert.key"); - - // Check if certificate files are present. If not, return None. - let cert_file_buf = match cert_file_res { - Ok(b) => b, - Err(e) if e.kind() == ErrorKind::NotFound => { - log!("TLS certificate 'certs/cert.pem' not found."); - return Ok(None); - } - Err(e) => return Err(e.into()), // Other IO error - }; - - let key_file_buf = match key_file_res { - Ok(b) => b, - Err(e) if e.kind() == ErrorKind::NotFound => { - log!("TLS key 'certs/cert.key' not found."); - return Ok(None); - } - Err(e) => return Err(e.into()), // Other IO error - }; - - // Continue with configuration if both files were found - let mut cert_reader = BufReader::new(cert_file_buf); - let cert_ders = rustls_pemfile::certs(&mut cert_reader) - .collect::, io::Error>>()?; - - // PKCS8 - let mut key_reader = BufReader::new(key_file_buf); - let mut key_ders = rustls_pemfile::pkcs8_private_keys(&mut key_reader) - .map(|r| r.map(Into::into)) // Explicit conversion - .collect::, io::Error>>()?; - - if key_ders.is_empty() { - // RSA - key_reader = BufReader::new(load_file_buf("certs", "cert.key")?); // Re-read key file - key_ders = rustls_pemfile::rsa_private_keys(&mut key_reader) - .map(|r| r.map(Into::into)) - .collect::, io::Error>>()?; - } - - if key_ders.is_empty() { - // EC - key_reader = BufReader::new(load_file_buf("certs", "cert.key")?); // Re-read key file - key_ders = rustls_pemfile::ec_private_keys(&mut key_reader) - .map(|r| r.map(Into::into)) - .collect::, io::Error>>()?; - } - - if key_ders.is_empty() { - return Err("No private keys found in key file. (Tried PKCS8, RSA, and EC)".into()); - } - - let mut config = rustls::ServerConfig::builder() - .with_no_client_auth() - .with_single_cert(cert_ders, key_ders.remove(0))?; - - config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; - - Ok(Some(Arc::new(config))) + "0.0.0.0".to_string() } diff --git a/src/server/socket.rs b/src/server/socket.rs index 5a8b815..25a00f7 100644 --- a/src/server/socket.rs +++ b/src/server/socket.rs @@ -1,51 +1,31 @@ use std::sync::Arc; +use axum::extract::ws::{Message, Utf8Bytes, WebSocket}; use futures::StreamExt; -use hyper::upgrade::OnUpgrade; -use hyper_util::rt::TokioIo; -use tokio_tungstenite::WebSocketStream; -use tungstenite::{Message, Utf8Bytes}; use crate::data::communication::{CommunicationType, CommunicationValue}; use crate::log; use crate::server::omikron_connection::OmikronConnection; -pub fn handle(path: String, upgrades: OnUpgrade) { +pub fn handle(path: String, upgrades: WebSocket) { tokio::spawn(async move { log!( "[ws] Spawning new task to handle WebSocket upgrade for path: {}", path ); - match upgrades.await { - Ok(upgraded_stream) => { - log!("[ws] WebSocket upgrade successful for path: {}", path); - let raw_stream = TokioIo::new(upgraded_stream); + log!("[ws] WebSocket upgrade successful for path: {}", path); - let ws_stream = WebSocketStream::from_raw_socket( - raw_stream, - tungstenite::protocol::Role::Server, - None, - ) - .await; - log!( - "[ws] WebSocket handshake successful, handling connection for {}", - path - ); + log!( + "[ws] WebSocket handshake successful, handling connection for {}", + path + ); - let (writer, reader) = ws_stream.split(); - if path == "/ws/omikron" { - let connection = OmikronConnection::new(writer, reader); - tokio::spawn(start_connecteable_handler(connection)); - } - } - Err(e) => { - log!( - "[ERROR] WebSocket upgrade failed for path {}: {:?}", - path, - e - ); - } + let (writer, reader) = upgrades.split(); + if path == "/ws/omikron" { + let connection = OmikronConnection::new(writer, reader); + tokio::spawn(start_connecteable_handler(connection)); } + log!( "[ws] WebSocket handling task for path: {} is finished.", path @@ -89,7 +69,7 @@ pub async fn start_connecteable_handler(connection: Arc) { log!("[ERROR] WS Error: {}. Breaking loop.", e); break; } - Ok(None) => { + Ok(_) => { log!("[ws_handler] WebSocket stream closed by peer. Breaking loop."); break; }