Merge remote-tracking branch 'origin/master'

unencrypted text group chat functional

.gitignore

@@ -59,3 +59,78 @@ Icon
 Network Trash Folder
 Temporary Items
 .apdisk
+
+### JetBrains template
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio, WebStorm and Rider
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/**/usage.statistics.xml
+.idea/**/dictionaries
+.idea/**/shelf
+
+# Generated files
+.idea/**/contentModel.xml
+
+# Sensitive or high-churn files
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+.idea/**/dbnavigator.xml
+
+# Gradle
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# Gradle and Maven with auto-import
+# When using Gradle or Maven with auto-import, you should exclude module files,
+# since they will be recreated, and may cause churn.  Uncomment if using
+# auto-import.
+# .idea/artifacts
+# .idea/compiler.xml
+# .idea/jarRepositories.xml
+# .idea/modules.xml
+# .idea/*.iml
+# .idea/modules
+# *.iml
+# *.ipr
+
+# CMake
+cmake-build-*/
+
+# Mongo Explorer plugin
+.idea/**/mongoSettings.xml
+
+# File-based project format
+*.iws
+
+# IntelliJ
+out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Cursive Clojure plugin
+.idea/replstate.xml
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+
+# Editor-based Rest Client
+.idea/httpRequests
+
+# Android studio 3.1+ serialized cache file
+.idea/caches/build_file_checksums.ser
+
+.vscode/launch.json

Cargo.toml

@@ -6,3 +6,4 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+crypto_box = "0.8.2"

protokoll.txt

@@ -0,0 +1,129 @@
+Nachrichtaufbau:
+================
+
+  u32 src_id  // inserted by server
+  u32 dest_id // 0: to all
+  u16 size    // min: 4
+  u8 payload[size]
+
+  payload[0]: type
+  KEY_LENGTH=32
+
+Payloads:
+=========
+
+  0-31 system messages (plaintext)
+
+  0   reserved
+  1   PING
+      size: 4
+      u8 type, pad1, pad2, pad3;
+  2   PONG
+      size: 4
+      u8 type, pad1, pad2, pad3;
+  3   ERROR
+      size: 4+len
+      u8 type, severity;
+      u16 code;
+      u8 msg[len];
+
+  32-63 server side messages (currently plaintext)
+
+  32  INIT           // first msg from server to new connection
+      size: 4 + 4*count
+      u8 type, pad1;
+      u16 count;     // including own; own id is in dest_id
+      u32 ids[count];
+  33  JOIN           // sent by server; joining id is src_id
+      size: 4
+      u8 type, pad1, pad2, pad3;
+  34  EXIT           // sent by server; exiting id is src_id
+      size: 4
+      u8 type, pad1, pad2, pad3;
+
+  64-95 initial setup messages (plaintext)
+
+  64  DH_SETUP  // setup secure channel client->client
+      size: 4 + ?
+      u8 type, cryptotype, pad2, pad3;
+      remainder impl-dependend
+  65  DH_RETURN
+      size: 4 + ?
+      u8 type, cryptotype, pad2, pad3;
+      remainder impl-dependend
+
+  96-127 client-client setup messages (DH or old KEY encrypted)
+
+  96  AUTHORIZE
+      T.B.D.
+  97  KEY_CURRENT
+      size: 4 + KEY_LENGTH
+      u8 type, pad1, pad2, pad3;
+      u8 key[KEY_LENGTH];
+  98  KEY_NEW
+      size: 4 + KEY_LENGTH
+      u8 type, pad1, pad2, pad3;
+      u8 key[KEY_LENGTH];
+
+  128-159 main messages (shared KEY encrypted)
+
+  128 MSG
+      size: 4 + ?
+      u8 type, pad1;
+      u16 generation;  // which key generation is used. Increasing monotonously, until overflow
+      u8 msg[];
+
+  160-255 reserved
+
+
+Funktionalität:
+===============
+
+- Server: Pakete an alle weiterleiten (dest_id==0) bzw. an nur einen Client (dest_id)
+          Bei Verbindungsaufbau: Liste aller Clients (Nummern) und eigene Nr (Zufallszahl), JOIN an alle
+          Bei Verbindungsverlust: EXIT an alle
+
+- Client: Eigentliche Funktionalität
+
+
+Funktionsweise:
+
+Client: Neue Verbindung
+  - warten auf INIT
+    Liste der Clients merken
+  - Jetzt nur Kommunikation mit zufälligem Client
+  - -> DH_SETUP
+    <- DH_RETURN   -> client-client DH_KEY
+    <- KEY_CURRENT (mit DH_KEY verschlüsselt)
+    -> KEY_NEW an alle (mit KEY_CURRENT verschlüsselt)
+  - Wenn an irgendeiner Stelle Timeout -> nochmal von vorne mit neuem Client
+
+Nachrichten:
+  - JOIN
+    Liste der Clients updaten
+  - EXIT
+    Liste der Clients updaten
+  - DH_SETUP
+    siehe oben -> DH_RETURN, KEY_CURRENT
+  - KEY_NEW
+    Alter KEY xor KEY_NEW -> neuer KEY, generation++
+  - MSG
+    generation prüfen, wenn > aktueller, warten auf KEY_NEW etc. (Zwischenspeichern, TODO)
+    entschlüsseln, MAC prüfen, anzeigen
+  - ERROR
+    code + msg ausgeben,
+    severity >= 64  Key neu aushandeln (TODO),
+             >= 96  Verbindung neu aufbauen (TODO),
+             >= 128 Exit
+
+Nachricht schicken:
+  MSG erstellen, Plaintext mit KEY verschlüsseln + MAC, generation einfügen
+
+
+Was fehlt / mögliche Fehler und Sicherheitsprobleme:
+====================================================
+
+- Keine Authentisierung - jeder kann reingrätschen
+- server side messages derzeit plaintext
+- keine Validierung (Zertifikate)
+- ...

src/bin/client.rs

@@ -1,26 +1,88 @@
+use crypto_box::{
+    aead::{Aead, AeadCore, OsRng},
+    PublicKey, SalsaBox, SecretKey,
+};
+use std::io::Write;
 use std::net::TcpStream;
+use crate::message::*;
+
+mod message;
 
 fn main() {
-    let stream = TcpStream::connect("127.0.0.1:8080");
-    if stream.is_ok() {
-        println!("Connected to the server!");
-    } else {
-        println!("Couldn't connect to server...");
-    }
+    // Server Port
+    let port: u32 = 7878;
 
-    loop {
-        let mut buffer = String::new();
-        match std::io::stdin().read_line(&mut buffer) {
-            Ok(n) => {
-                if n == 0 {
-                    break;
+    match TcpStream::connect(format!("localhost:{}", port)) {
+        Ok(mut stream) => {
+            println!("Successfully connected to server");
+
+            let bob_init_pub_key = PublicKey::from([
+                0xe8, 0x98, 0xc, 0x86, 0xe0, 0x32, 0xf1, 0xeb, 0x29, 0x75, 0x5, 0x2e, 0x8d, 0x65,
+                0xbd, 0xdd, 0x15, 0xc3, 0xb5, 0x96, 0x41, 0x17, 0x4e, 0xc9, 0x67, 0x8a, 0x53, 0x78,
+                0x9d, 0x92, 0xc7, 0x54,
+            ]);
+
+            // Generate a random secret key.
+            // NOTE: The secret key bytes can be accessed by calling `secret_key.as_bytes()`
+            let own_secret_key = SecretKey::generate(&mut OsRng);
+
+            // Get the public key for the secret key we just generated
+            let own_public_key = own_secret_key.public_key().clone();
+
+            // Create a `SalsaBox` by performing Diffie-Hellman key agreement between
+            // the two keys.
+            let salsa_box = SalsaBox::new(&bob_init_pub_key, &own_secret_key);
+
+            loop {
+                let mut buffer = String::new();
+                match std::io::stdin().read_line(&mut buffer) {
+                    Ok(n) => {
+                        if n == 0 {
+                            break; // Beenden bei EOF
+                        }
+                        //
+                        // Encryption
+                        //
+
+                        // Get a random nonce to encrypt the message under
+                        let nonce = SalsaBox::generate_nonce(&mut OsRng);
+
+                        // Message to encrypt
+                        let plaintext = buffer.trim().as_bytes();
+
+                        // Encrypt the message using the box
+                        let ciphertext = salsa_box.encrypt(&nonce, &plaintext[..]).expect("Fehler");
+                        let message = Message::new(0,0,4, PayloadType::Msg);
+
+                        println!("Sending {0} as {1:?}", buffer.trim(), plaintext);
+                        stream.write(buffer.as_bytes()).unwrap();
+
+                        //
+                        // Decryption
+                        //
+
+                        // Decrypt the message, using the same randomly generated nonce
+                        let decrypted_plaintext =
+                            salsa_box.decrypt(&nonce, &ciphertext[..]).expect("Fehler");
+                        let dec_plain_plaintext =
+                            std::str::from_utf8(&*decrypted_plaintext).expect("Nö");
+                        assert_eq!(&plaintext[..], &decrypted_plaintext[..]);
+
+                        println!(
+                            "Sent {0:?} as cypher: {1:?}, decrypted: {2:?}, {3}",
+                            plaintext, ciphertext, decrypted_plaintext, dec_plain_plaintext
+                        );
+                    }
+                    Err(error) => {
+                        println!("error: {error}");
+                        break;
+                    }
                 }
-                println!("{}", buffer);
-            }
-            Err(_error) => {
-                println!("Fehlerhafte Eingabe!");
-                break;
             }
         }
+        Err(e) => {
+            println!("Failed to connect: {}", e);
+        }
     }
-} // the stream is closed here
+    println!("Terminated.");
+}

src/bin/connection_pool.rs

@@ -0,0 +1,69 @@
+use std::{
+    sync::{Arc, Mutex},
+    collections::HashMap,
+    net::{TcpStream},
+    io::{Write},
+};
+
+pub struct ConnectionPool{
+    conns: Arc<Mutex<HashMap<u32, TcpStream>>>,
+}
+
+impl ConnectionPool{
+
+    pub fn new() -> Self {
+        ConnectionPool{
+            conns: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    pub fn add_connection(&self, iid: u32, conn: TcpStream) {
+        self.conns.lock().unwrap().insert(iid, conn);
+    }
+
+    pub fn del_connection(&self, iid: u32) {
+        self.conns.lock().unwrap().remove(&iid);
+    }
+
+    pub fn unicast(&self, recv_iid: &u32, msg: &String) {
+        let mut del_conn = false;
+
+        match self.conns.lock().unwrap().get(recv_iid) {
+            Some(mut conn) => {
+                if conn.write_all(msg.as_bytes()).is_err() {
+                    del_conn = true;
+                }
+            },
+            None => {
+                println!("Unknown recv_iid: {:#?}", recv_iid);
+                return;
+            }
+        }
+        if del_conn{
+            self.del_connection(*recv_iid);
+        }
+    }
+
+    pub fn broadcast(&self, sender_iid: &u32, msg: &String) {
+        let mut del_conns: Vec<u32> = Vec::new();
+        let mut msg1: String;
+
+        msg1 = format!("User {sender_iid}: ").parse().unwrap();
+        msg1.push_str(msg);
+
+        for (iid, mut conn) in self.conns.lock().unwrap().iter() {
+            if sender_iid.ne(iid){
+                if conn.write_all(msg1.as_bytes()).is_err() {
+                    del_conns.push(*iid);
+                    continue;
+                }
+                else if conn.write_all(b"\n").is_err() {
+                    del_conns.push(*iid);
+                }
+            }
+        }
+        for iid in del_conns {
+            self.del_connection(iid);
+        }
+    }
+}

src/bin/constants.rs

@@ -0,0 +1,2 @@
+pub const KEYBYTE: u8 = 0b1;
+pub const MSGBYTE: u8 = 0b01;

src/bin/message.rs

@@ -0,0 +1,107 @@
+use std::{
+    io::{BufReader, Read},
+    net::TcpStream,
+};
+
+const KEYLENGTH: u8 = 32;
+
+pub enum PayloadType {
+    //0-31 system messages (plaintext)
+    Ping = 1,
+    Pong = 2,
+    Error = 3,
+    //32 - 63 Serverside messages (currently plaintext)
+    Init = 32,
+    Join = 33,
+    Exit = 34,
+    //64 -95 initial setup messages(plaintext)
+    DhSetup = 64,
+    DhReturn = 65,
+    //96 - 127 client-client setup messages(DH or old KEY encrypted)
+    AUTHORIZE = 96,
+    KeyCurrent = 97,
+    KeyNew = 98,
+    //128-159 main messages (shared KEY encrypted)
+    Msg = 128,
+    //160-255 reserved
+}
+
+pub struct Message {
+    // Eigenschaften der Klasse
+    src_id: u32,
+    dest_id: u32,
+    size: u16,
+    payload: Vec<u8>,
+}
+
+impl Message {
+    pub fn new(src_id: u32, dest_id: u32, size: u16, msg_type: PayloadType) -> Message {
+        let mut msg = Message {
+            src_id: src_id,
+            dest_id: dest_id,
+            size: size,
+            payload: Vec::with_capacity(size.into()),
+        };
+        msg.payload[0] = msg_type as u8;
+        msg
+    }
+    fn creat_vec(&mut self) -> Vec<u8> {
+        let mut vec: Vec<u8> = vec![0; (10 + self.size).into()];
+
+        vec[0] = ((self.src_id >> 24) & 0xff) as u8;
+        vec[1] = ((self.src_id >> 16) & 0xff) as u8;
+        vec[2] = ((self.src_id >> 8) & 0xff) as u8;
+        vec[3] = (self.src_id & 0xff) as u8;
+        vec[4] = ((self.dest_id >> 24) & 0xff) as u8;
+        vec[5] = ((self.dest_id >> 16) & 0xff) as u8;
+        vec[6] = ((self.dest_id >> 8) & 0xff) as u8;
+        vec[7] = (self.dest_id & 0xff) as u8;
+        vec[8] = ((self.size >> 8) & 0xff) as u8;
+        vec[9] = (self.size & 0xff) as u8;
+
+
+        for i in 10..(10 + self.size).into(){
+            vec[i] = self.payload[i - 10];
+        }
+        vec
+    }
+
+    fn receive(stream: TcpStream) -> Message {
+        let buffreader = BufReader::new(stream);
+        let mut byte_iterator = buffreader.bytes();
+        let byte1 = byte_iterator.next().unwrap().unwrap();
+        let byte2 = byte_iterator.next().unwrap().unwrap();
+        let byte3 = byte_iterator.next().unwrap().unwrap();
+        let byte4 = byte_iterator.next().unwrap().unwrap();
+
+        let src_id: u32 =
+            (byte1 << 24) as u32 | (byte2 << 16) as u32 | (byte3 << 8) as u32 | (byte4 << 0) as u32;
+
+        let byte5 = byte_iterator.next().unwrap().unwrap();
+        let byte6 = byte_iterator.next().unwrap().unwrap();
+        let byte7 = byte_iterator.next().unwrap().unwrap();
+        let byte8 = byte_iterator.next().unwrap().unwrap();
+
+        let dest_id: u32 =
+            (byte5 << 24) as u32 | (byte6 << 16) as u32 | (byte7 << 8) as u32 | (byte8 << 0) as u32;
+
+        let byte9 = byte_iterator.next().unwrap().unwrap();
+        let byte10 = byte_iterator.next().unwrap().unwrap();
+
+        let size = (byte9 << 8) as u16 | (byte10 << 0) as u16;
+        assert!(size > 0, "Ungültige size größe size = {}", size);
+
+        let mut payload = Vec::<u8>::new();
+        for i in 0..size {
+            payload.push(byte_iterator.next().unwrap().unwrap());
+        }
+
+        Message {
+            src_id: src_id,
+            dest_id: dest_id,
+            size: size,
+            payload: payload,
+        }
+    }
+    // Methoden der Klasse
+}

src/bin/server.rs

@@ -1,31 +1,70 @@
 use std::{
-    io::{prelude::*, BufReader},
+    io::{prelude::*, BufReader, Lines},
     net::{TcpListener, TcpStream},
 };
-use std::io::Lines;
+use std::sync::{Arc, Mutex};
+use std::thread::available_parallelism;
+use crate::connection_pool::ConnectionPool;
+use crate::thread_pool::ThreadPool;
+
+mod thread_pool;
+mod connection_pool;
+
 
 fn main() {
+    // start server on localhost (for remote access use: 0.0.0.0)
     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
+    
+    // create thread pool (size = no. of cpu cores)
+    let pool = ThreadPool::new(available_parallelism().unwrap().get());
+    //let pool = ThreadPool::new(2);
 
+    let conn_pool = Arc::new(Mutex::new(ConnectionPool::new()));
+    let mut cc: u32 = 0;
+
+    // handling function for incoming requests
     for stream in listener.incoming() {
         let stream = stream.unwrap();
 
-        handle_connection(stream);
+        conn_pool.lock().unwrap().add_connection(cc, stream.try_clone().unwrap());
+
+        let ccc = cc;
+        let cconn_pool = conn_pool.clone();
+
+        // pass request to thread pool
+        pool.execute(move || {
+            handle_connection(stream, ccc, cconn_pool);
+        });
+
+        cc += 1;
     }
+    println!("Shutting down.");
 }
 
-fn handle_connection(mut stream: TcpStream) {
-    let mut stream2: TcpStream = stream.try_clone().unwrap();
+
+fn handle_connection(mut stream: TcpStream, iid: u32, conn_pool: Arc<Mutex<ConnectionPool>>) {
+    // clone TcpStream for simultaneous receiving and sending
+    //let mut stream2 = stream.try_clone().unwrap();
+
+    // initialize reading buffer (for better performance!)
     let buf_reader: BufReader<&mut TcpStream> = BufReader::new(&mut stream);
+    // request: read received text line by line (continuous until connection closed)
     let request: Lines<BufReader<&mut TcpStream>> = buf_reader.lines();
+    // let request: Bytes<BufReader<&mut TcpStream>> = buf_reader.bytes();
 
     println!("Request: {:#?}", request);
 
+    // processing function for every text line received by request (terminates when conn closed)
     for elem in request
     {
+        // store received line in variable s
         let s = elem.unwrap();
         println!("{:?}", s);
-        stream2.write(s.as_bytes()).unwrap();
-        stream2.write_all(b"\n").unwrap();
+
+        conn_pool.lock().unwrap().broadcast(&iid, &s);
+
+        // send received line back to sender (append LF)
+        //stream2.write_all(s.as_bytes()).unwrap();
+        //stream2.write_all(b"\n").unwrap();
     }
 }

src/bin/thread_pool.rs

@@ -0,0 +1,116 @@
+use std::{
+    sync::{mpsc, Arc, Mutex},
+    thread
+};
+
+
+pub struct ThreadPool {
+    workers: Vec<Worker>,
+    sender: Option<mpsc::Sender<Job>>,
+}
+
+// Box: pointer for heap allocation (smart pointer); dyn: unknown type to compile time
+type Job = Box<dyn FnOnce() + Send + 'static>;
+
+
+impl ThreadPool {
+    /// Create a new ThreadPool.
+    ///
+    /// The size is the number of threads in the pool.
+    ///
+    /// # Panics
+    ///
+    /// The `new` function will panic if the size is zero.
+    pub fn new(size: usize) -> ThreadPool {
+        assert!(size > 0);
+
+        // create asynchronous channel with sender/receiver pair (for jobs!)
+        let (sender, receiver) = mpsc::channel();
+
+        // protect receiver with thread-safe shared (Arc) mutex to avoid multiple access
+        let receiver = Arc::new(Mutex::new(receiver));
+
+        // pre(!)allocate space for workers
+        let mut workers = Vec::with_capacity(size);
+
+        // initialize all workers...
+        for id in 0..size {
+            // create worker with unique id and pointer to the shared mutex
+            workers.push(Worker::new(id, Arc::clone(&receiver)));
+        }
+
+        // set thread pool variables
+        ThreadPool {
+            workers,
+            // Some --> enum variant (Option): indicates that sender may not exist (=be null)
+            sender: Some(sender),
+        }
+    }
+
+    pub fn execute<F>(&self, f: F)
+        where
+            // FnOnce() --> only callable once; Send --> transfer to another thread; 'static --> lifetime
+            F: FnOnce() + Send + 'static,
+    {
+        let job = Box::new(f);
+
+        // commit job to sender object --> pass to threads for execution
+        self.sender.as_ref().unwrap().send(job).unwrap();
+    }
+}
+
+impl Drop for ThreadPool {
+    fn drop(&mut self) {
+        // drop sender first
+        drop(self.sender.take());
+
+        // then drop workers
+        for worker in &mut self.workers {
+            println!("Shutting down worker {}", worker.id);
+
+            // join worker thread
+            if let Some(thread) = worker.thread.take() {
+                thread.join().unwrap();
+            }
+        }
+    }
+}
+
+
+struct Worker {
+    id: usize,
+    thread: Option<thread::JoinHandle<()>>,
+}
+
+impl Worker {
+    /// Create a new Worker..
+    ///
+    /// The id is a unique identifier.
+    /// The receiver is a shared pointer to a receiver protected by Mutex.
+    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
+        // spawn thread and enter infinite loop
+        let thread = thread::spawn(move || loop {
+            // lock mutex and wait for job
+            let message = receiver.lock().unwrap().recv();
+
+            match message {
+                // normal operation --> execute job
+                Ok(job) => {
+                    println!("Worker {id} got a job; executing.");
+
+                    job();
+                }
+                // exit gracefully when pool is closed --> recv will return error then
+                Err(_) => {
+                    println!("Worker {id} disconnected; shutting down.");
+                    break;
+                }
+            }
+        });
+
+        Worker {
+            id,
+            thread: Some(thread),
+        }
+    }
+}