Creation

src/server/api.rs

@@ -0,0 +1,177 @@
+use crate::{get_public_key, log};
+use crate::{
+    sql::{
+        iota_omikron_tracker::{get_omikron_for_iota, track_iota_omikron, untrack_iota},
+        sql::{
+            change_about, change_avatar, change_display_name, change_iota_id, change_iota_key,
+            change_keys, change_status, change_username, get_by_id, get_by_username,
+            get_iota_by_id, get_omikron_by_id, get_random_omikron, get_register_id,
+            register_complete_iota, register_complete_user,
+        },
+    },
+    util::crypto_helper::public_key_to_base64,
+};
+use axum::http::HeaderValue;
+use http_body_util::Full;
+use hyper::body::Bytes;
+use hyper::{HeaderMap, Response as HttpResponse, StatusCode};
+use json::JsonValue;
+
+pub async fn handle(
+    path: &str,
+    headers: HeaderMap<HeaderValue>,
+    body_string: Option<String>,
+) -> HttpResponse<Full<Bytes>> {
+    let path_parts: Vec<&str> = path.split("/").filter(|s| !s.is_empty()).collect();
+
+    let body: Option<JsonValue> = if body_string.is_some() {
+        if let Ok(body_json) = json::parse(&body_string.unwrap()) {
+            Some(body_json)
+        } else {
+            None
+        }
+    } else {
+        None
+    };
+    // api/
+    //   get/
+    //     omikron/
+    //     id/
+    //   register/
+    //     innit/
+    //     complete
+    log!("{}", path);
+    log!("{:?} .len = {}", path_parts, path_parts.len());
+    let (status, content, body_text) = if path_parts.len() >= 2 {
+        match path_parts[1] {
+            "get" => match path_parts[2] {
+                // api/get/omikron -> any omikron
+                // api/get/omikron/<id> -> omikron for id (user / iota / omikron)
+                "omikron" => {
+                    if path_parts.len() == 3 {
+                        if let Ok((id, public_key, ip_address)) = get_random_omikron().await {
+                            (
+                                StatusCode::OK,
+                                "application/json",
+                                format!(
+                                    "{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
+                                    id, public_key, ip_address
+                                ),
+                            )
+                        } else {
+                            not_found()
+                        }
+                    } else if path_parts.len() == 4 {
+                        let id = path_parts[3].parse::<i64>().unwrap_or(0);
+                        if id == 0 {
+                            not_found()
+                        } else if let Ok((omikron_id, public_key, ip_address)) =
+                            get_omikron_by_id(id).await
+                        {
+                            (
+                                StatusCode::OK,
+                                "application/json",
+                                format!(
+                                    "{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
+                                    omikron_id, public_key, ip_address
+                                ),
+                            )
+                        } else if let Some(omikron_id) = get_omikron_for_iota(id).await {
+                            if let Ok((omikron_id, public_key, ip_address)) =
+                                get_omikron_by_id(omikron_id).await
+                            {
+                                (
+                                    StatusCode::OK,
+                                    "application/json",
+                                    format!(
+                                        "{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
+                                        omikron_id, public_key, ip_address
+                                    ),
+                                )
+                            } else {
+                                not_found()
+                            }
+                        } else if let Ok((_, iota_id, _, _, _, _, _, _, _, _, _, _)) =
+                            get_by_id(id).await
+                        {
+                            if let Some(omikron_id) = get_omikron_for_iota(iota_id).await {
+                                if let Ok((omikron_id, public_key, ip_address)) =
+                                    get_omikron_by_id(omikron_id).await
+                                {
+                                    (
+                                        StatusCode::OK,
+                                        "application/json",
+                                        format!(
+                                            "{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
+                                            omikron_id, public_key, ip_address
+                                        ),
+                                    )
+                                } else {
+                                    not_found()
+                                }
+                            } else {
+                                not_found()
+                            }
+                        } else {
+                            not_found()
+                        }
+                    } else {
+                        bad_request()
+                    }
+                }
+                // get/id/<username>
+                "id" => {
+                    let username = path_parts[3];
+                    bad_request()
+                }
+                "public_key" => (
+                    StatusCode::OK,
+                    "application/json",
+                    public_key_to_base64(&get_public_key()),
+                ),
+                _ => {
+                    let id = path_parts[2];
+                    let id: i64 = id.parse().unwrap_or(0);
+                    if id == 0 {
+                        bad_request()
+                    } else {
+                        bad_request()
+                    }
+                }
+            },
+            _ => not_found(),
+        }
+    } else {
+        not_found()
+    };
+    let body = Full::new(Bytes::from(body_text.to_string()));
+    HttpResponse::builder().status(status).body(body).unwrap()
+}
+pub fn bad_request() -> (StatusCode, &'static str, String) {
+    (
+        StatusCode::BAD_REQUEST,
+        "text/text",
+        "400 Bad Request".to_string(),
+    )
+}
+pub fn unauthorized() -> (StatusCode, &'static str, String) {
+    (
+        StatusCode::UNAUTHORIZED,
+        "text/text",
+        "401 Unauthorized".to_string(),
+    )
+}
+pub fn forbidden() -> (StatusCode, &'static str, String) {
+    (
+        StatusCode::FORBIDDEN,
+        "text/text",
+        "403 Forbidden".to_string(),
+    )
+}
+pub fn not_found() -> (StatusCode, &'static str, String) {
+    (
+        StatusCode::NOT_FOUND,
+        "text/text",
+        "404 Not Found".to_string(),
+    )
+}

src/server/mod.rs

@@ -0,0 +1,5 @@
+pub mod api;
+pub mod omikron_connection;
+pub mod omikron_manager;
+pub mod server;
+pub mod socket;

src/server/omikron_connection.rs

@@ -0,0 +1,258 @@
+use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
+use crate::get_public_key;
+use crate::sql::sql::get_omikron_by_id;
+use crate::util::crypto_helper::encrypt;
+use crate::{get_private_key, log_in_from, log_out_from};
+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 json::JsonValue;
+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<RwLock<SplitSink<WebSocketStream<TokioIo<Upgraded>>, Message>>>,
+    pub receiver: Arc<RwLock<SplitStream<WebSocketStream<TokioIo<Upgraded>>>>>,
+    pub omikron_id: Arc<RwLock<i64>>,
+    pub pub_key: Arc<RwLock<Option<Vec<u8>>>>,
+    identified: Arc<RwLock<bool>>,
+    challenged: Arc<RwLock<bool>>,
+    challenge: Arc<RwLock<String>>,
+    pub ping: Arc<RwLock<i64>>,
+    waiting_tasks: DashMap<
+        Uuid,
+        Box<dyn Fn(Arc<OmikronConnection>, CommunicationValue) -> bool + Send + Sync>,
+    >,
+}
+
+impl OmikronConnection {
+    pub fn new(
+        sender: SplitSink<WebSocketStream<TokioIo<Upgraded>>, Message>,
+        receiver: SplitStream<WebSocketStream<TokioIo<Upgraded>>>,
+    ) -> Arc<Self> {
+        Arc::new(Self {
+            sender: Arc::new(RwLock::new(sender)),
+            receiver: Arc::new(RwLock::new(receiver)),
+            omikron_id: Arc::new(RwLock::new(0)),
+            pub_key: Arc::new(RwLock::new(None)),
+            identified: Arc::new(RwLock::new(false)),
+            challenged: Arc::new(RwLock::new(false)),
+            challenge: Arc::new(RwLock::new(String::new())),
+            ping: Arc::new(RwLock::new(-1)),
+            waiting_tasks: DashMap::new(),
+        })
+    }
+    pub async fn send_message(&self, message: &CommunicationValue) {
+        let mut sender = self.sender.write().await;
+        let message_text = Message::Text(Utf8Bytes::from(message.to_json().to_string()));
+        log_out_from!(*self.omikron_id.read().await, "{}", message_text);
+        sender.send(message_text).await.unwrap();
+    }
+    pub async fn get_user_id(&self) -> i64 {
+        *self.omikron_id.read().await
+    }
+    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<Self>, message: String) {
+        let cv = CommunicationValue::from_json(&message);
+        if cv.is_type(CommunicationType::ping) {
+            self.handle_ping(cv).await;
+            return;
+        }
+
+        log_in_from!(*self.omikron_id.read().await, "{}", message);
+
+        if !*self.identified.read().await && cv.is_type(CommunicationType::identification) {
+            let omikron_id = cv
+                .get_data(DataTypes::omikron)
+                .unwrap_or(&JsonValue::Null)
+                .as_i64()
+                .unwrap_or(0);
+            if let Ok((_, public_key, _)) = get_omikron_by_id(omikron_id).await {
+                // Generate Challenge, encrypt it and send it to the omikron
+                *self.omikron_id.write().await = omikron_id;
+                *self.identified.write().await = true;
+
+                let challenge_str: String = rand::thread_rng()
+                    .sample_iter(&Alphanumeric)
+                    .take(32)
+                    .map(char::from)
+                    .collect();
+
+                *self.challenge.write().await = challenge_str.clone();
+
+                let user_public_key_bytes = match STANDARD.decode(&public_key) {
+                    Ok(bytes) => bytes,
+                    Err(_) => {
+                        self.send_error_response(
+                            &cv.get_id(),
+                            CommunicationType::error_invalid_user_id,
+                        )
+                        .await;
+                        return;
+                    }
+                };
+                *self.pub_key.write().await = Some(user_public_key_bytes.clone());
+
+                let omikron_pub_key: PublicKey = match PublicKey::from_bytes(&user_public_key_bytes)
+                {
+                    Some(key) => key,
+                    None => {
+                        self.send_error_response(
+                            &cv.get_id(),
+                            CommunicationType::error_invalid_user_id,
+                        )
+                        .await;
+                        return;
+                    }
+                };
+
+                let encrypted_challenge =
+                    encrypt(get_private_key(), omikron_pub_key, &challenge_str)
+                        .unwrap_or("".to_string());
+
+                let response = CommunicationValue::new(CommunicationType::challenge)
+                    .add_data_str(
+                        DataTypes::public_key,
+                        STANDARD.encode(get_public_key().as_bytes()),
+                    )
+                    .add_data_str(DataTypes::challenge, encrypted_challenge)
+                    .with_id(cv.get_id());
+
+                self.send_message(&response).await;
+                // prepare Challenge Response handling
+                self.waiting_tasks.insert(
+                    cv.get_id(),
+                    Box::new(
+                        |selfc: Arc<OmikronConnection>, cv: CommunicationValue| -> bool {
+                            tokio::spawn(async move {
+                                let client_challenge_response_b64 =
+                                    match cv.get_data(DataTypes::challenge) {
+                                        Some(data) => data.to_string(),
+                                        None => {
+                                            selfc
+                                                .send_error_response(
+                                                    &cv.get_id(),
+                                                    CommunicationType::error,
+                                                )
+                                                .await;
+                                            return;
+                                        }
+                                    };
+
+                                let challenge_response_bytes =
+                                    match STANDARD.decode(&client_challenge_response_b64) {
+                                        Ok(bytes) => bytes,
+                                        Err(_) => {
+                                            selfc
+                                                .send_error_response(
+                                                    &cv.get_id(),
+                                                    CommunicationType::error,
+                                                )
+                                                .await;
+                                            return;
+                                        }
+                                    };
+
+                                if challenge_response_bytes.len() < 12 {
+                                    selfc
+                                        .send_error_response(&cv.get_id(), CommunicationType::error)
+                                        .await;
+                                    return;
+                                }
+
+                                let client_response = cv
+                                    .get_data(DataTypes::challenge)
+                                    .unwrap_or(&JsonValue::Null)
+                                    .as_str()
+                                    .unwrap_or("");
+
+                                let expected_challenge = selfc.challenge.read().await.clone();
+
+                                if client_response != expected_challenge {
+                                    selfc
+                                        .send_error_response(
+                                            &cv.get_id(),
+                                            CommunicationType::error_invalid_challenge,
+                                        )
+                                        .await;
+                                    selfc.close().await;
+                                    return;
+                                }
+
+                                *selfc.challenged.write().await = true;
+
+                                let response = CommunicationValue::new(
+                                    CommunicationType::identification_response,
+                                )
+                                .with_id(cv.get_id());
+
+                                selfc.send_message(&response).await;
+                                return;
+                            });
+                            return true;
+                        },
+                    ),
+                );
+            } else {
+                self.send_error_response(&cv.get_id(), CommunicationType::error_not_found)
+                    .await;
+            }
+            return;
+        }
+
+        if self.waiting_tasks.contains_key(&cv.get_id()) {
+            let (_, task) = self.waiting_tasks.remove(&cv.get_id()).unwrap();
+            let _ = task(self.clone(), cv.clone());
+        }
+
+        if !self.is_identified().await {
+            self.send_error_response(&cv.get_id(), CommunicationType::error_not_found)
+                .await;
+            return;
+        }
+    }
+
+    async fn send_error_response(&self, message_id: &Uuid, error_type: CommunicationType) {
+        let error = CommunicationValue::new(error_type).with_id(*message_id);
+        self.send_message(&error).await;
+    }
+    pub async fn close(&self) {
+        let mut sender = self.sender.write().await;
+        let _ = sender.close().await;
+    }
+    pub async fn handle_close(self: Arc<Self>) {
+        if self.is_identified().await {
+            if self.get_user_id().await != 0 {}
+        }
+    }
+
+    async fn handle_ping(&self, cv: CommunicationValue) {
+        if let Some(last_ping) = cv.get_data(DataTypes::last_ping) {
+            if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
+                let mut ping_guard = self.ping.write().await;
+                *ping_guard = ping_val;
+            }
+        }
+
+        let response = CommunicationValue::new(CommunicationType::pong).with_id(cv.get_id());
+
+        self.send_message(&response).await;
+    }
+}

src/server/server.rs

@@ -0,0 +1,475 @@
+use crate::log;
+use crate::server::api;
+use crate::server::omikron_connection::OmikronConnection;
+use crate::util::file_util::load_file_buf;
+
+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::{Method, StatusCode};
+use hyper::{
+    Request as HttpRequest, Response as HttpResponse, body::Incoming, server::conn::http1, upgrade,
+};
+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::SocketAddr;
+use std::result::Result::Ok;
+use std::sync::Arc;
+use std::{future::Future, pin::Pin, time::Duration};
+use tokio::net::TcpListener;
+use tokio::sync::broadcast;
+use tokio_rustls::TlsAcceptor;
+use tokio_tungstenite::{WebSocketStream, accept_async};
+use tower::Service;
+use tungstenite::Message;
+#[derive(Clone)]
+struct HttpService {
+    peer_addr: SocketAddr,
+}
+
+impl Service<HttpRequest<Incoming>> for HttpService {
+    type Response = HttpResponse<Full<Bytes>>;
+    type Error = io::Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(
+        &mut self,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        std::task::Poll::Ready(std::io::Result::Ok(()))
+    }
+
+    fn call(&mut self, req: HttpRequest<Incoming>) -> 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 == "/ws/omikron"
+                && method == Method::GET
+                && headers
+                    .get("connection")
+                    .map(|v| {
+                        v.to_str()
+                            .unwrap_or("")
+                            .split(',')
+                            .any(|s| s.trim().eq_ignore_ascii_case("upgrade"))
+                    })
+                    .unwrap_or(false)
+                && headers
+                    .get("upgrade")
+                    .map(|v| v.to_str().unwrap_or("").eq_ignore_ascii_case("websocket"))
+                    .unwrap_or(false);
+
+            if is_websocket_upgrade {
+                log!("Attempting WebSocket upgrade on {}", path);
+
+                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);
+                    log!("Handling WebSocket upgrade");
+
+                    // Spawn upgrade handling to avoid blocking the service call
+                    tokio::spawn(async move {
+                        match upgrades.await {
+                            Ok(upgraded_stream) => {
+                                log!("Valid WebSocket upgrade");
+                                let raw_stream = TokioIo::new(upgraded_stream);
+
+                                let ws_stream = WebSocketStream::from_raw_socket(
+                                    raw_stream,
+                                    tungstenite::protocol::Role::Server,
+                                    None,
+                                )
+                                .await;
+                                log!(
+                                    "WebSocket handshake successful, handling connection for Omikron"
+                                );
+
+                                // Split stream for OmikronConnection
+                                let (writer, reader) = ws_stream.split();
+
+                                // INTEGRATION START
+                                // Erstelle die Connection und starte den Handler
+                                let connection = OmikronConnection::new(writer, reader);
+
+                                // Handler in separatem Task starten
+                                start_omikron_handler(connection).await;
+                                // INTEGRATION END
+                            }
+                            Err(e) => {
+                                log!("WebSocket upgrade failed after response: {:?}", e);
+                            }
+                        }
+                    });
+
+                    log!("Handled WebSocket connection initiation");
+                    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 = 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<String> = match String::from_utf8(bytes.to_vec()) {
+                    Ok(s) => Some(s),
+                    Err(_) => None,
+                };
+
+                Ok(api::handle(&path, headers.clone(), body_string).await)
+            } else {
+                let response = HttpResponse::builder()
+                    .status(StatusCode::BAD_REQUEST)
+                    .body(Full::new(Bytes::from("No path provided")))
+                    .unwrap();
+                Ok(response)
+            }
+        };
+
+        Box::pin(fut.map_err(|err: color_eyre::eyre::ErrReport| {
+            io::Error::new(
+                io::ErrorKind::Other,
+                format!("Error in request handling: {}", err),
+            )
+        }))
+    }
+}
+
+async fn run_http_server(port: u16) -> bool {
+    let ip = "0.0.0.0".to_string();
+    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! {
+
+                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 {
+                                // Prepare the 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::<io::Error>()) {
+                                                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;
+                        }
+                    }
+                }
+            }
+        }
+    });
+
+    true
+}
+
+/// Runs the encrypted HTTPS/WSS server loop using the provided TLS config.
+async fn run_tls_server(port: u16, tls_config: Arc<ServerConfig>) -> bool {
+    let mut ip = "0.0.0.0".to_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.to_string());
+        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::<io::Error>()) {
+                                                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!("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(None) => 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<Option<Arc<ServerConfig>>, Box<dyn Error>> {
+    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::<Result<Vec<CertificateDer>, 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::<Result<Vec<PrivateKeyDer>, 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::<Result<Vec<PrivateKeyDer>, 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::<Result<Vec<PrivateKeyDer>, io::Error>>()?;
+    }
+
+    if key_ders.is_empty() {
+        return Err("No private keys found in key file. (Tried PKCS8, RSA, and EC)".into());
+    }
+
+    let config = rustls::ServerConfig::builder()
+        .with_no_client_auth()
+        .with_single_cert(cert_ders, key_ders.remove(0))
+        .map_err(|e| io::Error::new(ErrorKind::Other, e.to_string()))?;
+
+    Ok(Some(Arc::new(config)))
+}
+// In deiner Server-Logik, wo OmikronConnection initialisiert wird:
+
+pub async fn start_omikron_handler(connection: Arc<OmikronConnection>) {
+    loop {
+        let msg = match {
+            let mut receiver = connection.receiver.write().await;
+            receiver.next().await
+        } {
+            Some(Ok(msg)) => msg,
+            Some(Err(e)) => {
+                log!("WS Error: {}", e);
+                break;
+            }
+            None => break, // Stream ended
+        };
+
+        match msg {
+            Message::Text(text) => {
+                let conn_clone = connection.clone();
+                tokio::spawn(async move {
+                    conn_clone.handle_message(text.to_string()).await;
+                });
+            }
+            Message::Close(_) => {
+                break;
+            }
+            // Other message types like Binary, Ping, Pong are ignored.
+            _ => {}
+        }
+    }
+    connection.handle_close().await;
+}

src/server/socket.rs

@@ -0,0 +1,58 @@
+use std::sync::Arc;
+
+use futures::StreamExt;
+use futures::stream::SplitSink;
+use futures::stream::SplitStream;
+use hyper::upgrade::Upgraded;
+use hyper_util::rt::TokioIo;
+use tungstenite::Message;
+
+use crate::log;
+use crate::server::omikron_connection::OmikronConnection;
+
+pub fn handle(
+    path: String,
+    writer: SplitSink<tokio_tungstenite::WebSocketStream<TokioIo<Upgraded>>, Message>,
+    reader: SplitStream<tokio_tungstenite::WebSocketStream<TokioIo<Upgraded>>>,
+) {
+    log!("handling");
+    tokio::spawn(async move {
+        if path.starts_with("/ws/phi/") {
+        } else if path.starts_with("/ws/omikron/") {
+            let community_conn: Arc<OmikronConnection> =
+                Arc::from(OmikronConnection::new(writer, reader));
+            loop {
+                let msg_result: Option<Result<_, _>> = {
+                    let mut session_lock = community_conn.receiver.write().await;
+                    session_lock.next().await
+                };
+
+                match msg_result {
+                    Some(Ok(msg)) => {
+                        if msg.is_text() {
+                            let text = msg.into_text().unwrap();
+                            community_conn
+                                .clone()
+                                .handle_message(text.to_string())
+                                .await;
+                        } else if msg.is_close() {
+                            log!("Closing: {}", msg);
+                            community_conn.handle_close().await;
+                            return;
+                        }
+                    }
+                    Some(Err(e)) => {
+                        log!("Closing ERR: {}", e);
+                        community_conn.handle_close().await;
+                        return;
+                    }
+                    None => {
+                        log!("Closed Session me!");
+                        community_conn.handle_close().await;
+                        return;
+                    }
+                }
+            }
+        }
+    });
+}