kernel/networking/
server_sockets.rs

1use std::io::{Read, Write};
2use std::net::{Shutdown, TcpListener, TcpStream};
3use std::thread;
4use std::sync::atomic::Ordering;
5use std::sync::mpsc;
6use std::sync::mpsc::{Receiver,Sender};
7use std::time::Duration;
8use rand::{RngExt};
9use common::{r_log, r_debug_log};
10use common::logging::LogLevel::*;
11use common::networking::messages::{HandshakeRequest, HandshakeResponse, SessionID, TcpClientMessage, TcpServerMessage};
12use common::networking::messages::TcpServerMessage::{CommandOutput, ImplicitCommandOutput, LogoutOk};
13use common::networking::subscription_objects::UpdateMode;
14use crate::networking::connection_engine::{ClientSession, ConnectionID, NEXT_CONNECTION_ID, SERVER_STATE};
15use crate::networking::subscriptions::add_subscription;
16use crate::cli::run_command;
17use crate::cli::execute_implicit_cli_action;
18
19/// Binds the TCP server socket to the specified port and spawns the main incoming connection listener loop.
20///
21/// If binding fails (e.g., because another instance is already running), an error is logged and the process exits.
22/// Every accepted connection is assigned a unique [`ConnectionID`] and dispatched to its own dedicated worker thread.
23///
24/// # Arguments
25///
26/// * `port` - The network port number on which the server should listen for incoming client connections.
27pub fn activate_socket(port: u16) {
28    let address = format!("0.0.0.0:{}", port);
29    let listener = match TcpListener::bind(&address) {
30        Ok(l) => l,
31        Err(e) => {
32            r_log!(Error,"CRITICAL: Failed to bind TCP socket on address {}. Is another kernel instance already \
33                    running? To change port, use --port [port]. OS Error: {}",address, e);
34            // Give the Logger time to do its thing
35            thread::sleep(Duration::from_millis(50));
36            std::process::exit(1);
37        }
38    };
39
40    r_log!(Info,"Listening on {} for Clients", listener.local_addr().unwrap());
41
42    thread::spawn(move|| {
43        for stream in listener.incoming() {
44            match stream {
45                Ok(stream) => {
46                    let connection_id = NEXT_CONNECTION_ID.fetch_add(1, Ordering::Relaxed);
47                    r_log!(SuccessEvent,"[Conn {}] New client connected: {}",
48                        connection_id, stream.peer_addr().unwrap());
49
50                    thread::spawn(move || {
51                        handle_client(stream, connection_id);
52                    });
53                }
54                Err(e) => {
55                    r_log!(Error,"Got error trying to establish connection : {}", e);
56                }
57            }
58        }
59    });
60}
61
62/// Orchestrates the lifecycle of an individual client connection.
63///
64/// Performs the initial protocol handshake and version compatibility check, establishes
65/// independent channels for asynchronous communication, and spawns parallel reader and writer
66/// threads for the stream.
67///
68/// # Arguments
69///
70/// * `stream`        - The active `TcpStream` associated with the connected client.
71/// * `connection_id` - The unique identifier assigned to this physical connection.
72fn handle_client(mut stream: TcpStream, connection_id: ConnectionID) {
73
74    match check_version_compatibility(&mut stream) {
75        Err(HandshakeError::InvalidData(e)) => {
76            r_log!(Error,"[Conn {}] Handshake Error: {}. Connection Closed.", connection_id, e);
77            return;
78        }
79
80        Err(HandshakeError::VersionMismatch {client_version, server_version}) => {
81            r_log!(Warning,"[Conn {}] Version Mismatch: Client: {}, Server: {}. Connection Closed",
82                     connection_id, client_version, server_version);
83            return;
84        }
85
86        Ok(()) => {
87            r_log!(Info,"[Conn {}] Versions match. Handshake completed.", connection_id);
88        }
89    }
90
91    let (tx_channel, rx_channel) = mpsc::channel::<TcpServerMessage>();
92    let write_stream = stream.try_clone().unwrap();
93
94    thread::spawn(move|| {
95        write_thread(rx_channel, write_stream, connection_id);
96    });
97
98
99    read_thread(&mut stream, connection_id, &tx_channel);
100
101}
102
103/// Represents possible errors that can occur during the client handshake phase.
104enum HandshakeError {
105    VersionMismatch {
106        server_version: String,
107        client_version: String,
108    },
109    InvalidData(String),
110}
111
112/// Validates the initial handshake packet sent by a newly connected client.
113///
114/// Verifies the custom magic string protocol identifier, deserializes the client's version details,
115/// compares the protocol hash against the running server version, and transmits the appropriate
116/// [`HandshakeResponse`] back across the stream.
117///
118/// # Arguments
119///
120/// * `stream` - A mutable reference to the client's `TcpStream`.
121fn check_version_compatibility(stream: &mut TcpStream) -> Result<(), HandshakeError> {
122    let mut len_buffer = [0u8; 4];
123    let mut buffer = match stream.read_exact(&mut len_buffer) {
124        Ok(_) => {
125            let msg_len = u32::from_be_bytes(len_buffer) as usize;
126            vec![0u8; msg_len]
127        },
128        Err(e) => {
129            return Err(HandshakeError::InvalidData(format!("Len-Read Error: {}", e)));
130        }
131    };
132
133    let bytes = match stream.read_exact(&mut buffer) {
134        Err(e) => {
135            return Err(HandshakeError::InvalidData(format!("Read Error: {}", e)));
136        }
137        Ok(_) => buffer
138    };
139
140    match bincode::deserialize::<HandshakeRequest>(&bytes) {
141        Ok(request) => {
142            if request.magic_string != "REKTAL" {
143                return Err(HandshakeError::InvalidData("Wrong magic string. Client is not an Rektal-Client".into()));
144            }
145
146            let server_hash = common::networking::messages::get_protocol_version();
147            let server_version = env!("CARGO_PKG_VERSION").to_string();
148
149            if request.protocol_hash != server_hash {
150                let response = HandshakeResponse::Mismatch {
151                    server_version: server_version.clone(),
152                };
153
154                let payload = bincode::serialize(&response).unwrap();
155                let len = payload.len() as u32;
156
157                stream.write_all(&len.to_be_bytes()).unwrap();
158                stream.write_all(&payload).unwrap();
159
160                Err(HandshakeError::VersionMismatch {
161                    server_version,
162                    client_version: request.client_version
163                })
164            } else {
165                let response = HandshakeResponse::Ok;
166                let payload = bincode::serialize(&response).unwrap();
167                let len = payload.len() as u32;
168                stream.write_all(&len.to_be_bytes()).unwrap();
169                stream.write_all(&payload).unwrap();
170
171                Ok(())
172            }
173
174        }
175        Err(_) => Err(HandshakeError::InvalidData("Couldnt deserialize handshake".into())),
176    }
177
178}
179
180/// Continuously reads, deserializes, and processes incoming messages from an active client connection.
181///
182/// Handles session authorization enforcement, delegates requests to appropriate message handlers,
183/// and performs cleanup operations (updating session status to sleeping) when the client disconnects.
184///
185/// # Arguments
186///
187/// * `stream`        - A mutable reference to the client's `TcpStream`.
188/// * `connection_id` - The unique identifier of this connection.
189/// * `tx_channel`    - The message sender channel used to push asynchronous server responses to the tcp writer thread.
190fn read_thread(stream: &mut TcpStream, connection_id: ConnectionID, tx_channel: &Sender<TcpServerMessage>) {
191    let mut len_buffer = [0u8; 4];
192    let mut token: Option<SessionID> = None;
193
194    loop {
195        let mut buffer = match stream.read_exact(&mut len_buffer) {
196            Ok(_) => {
197                let msg_len = u32::from_be_bytes(len_buffer) as usize;
198                vec![0u8; msg_len]
199            },
200            Err(e) => {
201                if e.kind() == std::io::ErrorKind::UnexpectedEof {
202                    if token.is_none() {
203                        r_log!(SuccessEvent,"[Conn {}] Client disconnected successfully", connection_id);
204                    } else {
205                        r_log!(Error, "[Conn {}] Client disconnected without logging out. Session is still active",
206                            connection_id
207                        );
208                    }
209                } else {
210                    r_log!(Error, "[Conn {}] Length of read-stream-error: {}", connection_id, e);
211                }
212                break;
213            }
214        };
215
216        let bytes = match stream.read_exact(&mut buffer) {
217            Err(e) => {
218                r_log!(Error,"[Conn {}] Connection error : {}", connection_id, e);
219                break;
220            }
221            Ok(_) => buffer
222        };
223
224        let msg = bincode::deserialize::<TcpClientMessage>(&bytes).unwrap();
225        r_debug_log!(Info,"[Conn {}] Received Enum: {:?}", connection_id, msg);
226
227        token = update_login_status(&msg, token, &tx_channel, connection_id);
228
229        let Some(_) = token else {
230            if !matches!(msg, TcpClientMessage::Logout | TcpClientMessage::Relogin {..} | TcpClientMessage::Login {..}){
231                r_log!(UserError,"[Conn {}] Message discarded: Client is not authorized.", connection_id);
232                let _ = tx_channel.send(TcpServerMessage::Unauthenticated);
233            }
234            continue
235        };
236
237        match msg {
238            TcpClientMessage::Login { .. } | TcpClientMessage::Relogin { .. } => {
239                //Already got handled by update_login_status
240            },
241            TcpClientMessage::Logout => { unreachable!() },
242
243            _ => {
244                if let Some(response) = handle_messages(msg, connection_id, token.unwrap()) {
245                    if let Err(e) = tx_channel.send(response) {
246                        r_log!(Error,"[Conn {}] Got error while sending response to channel: {}", connection_id, e);
247                    }
248                }
249            }
250        }
251    }
252
253    //Clean Up
254    if let Some(token) = token {
255        let mut state = SERVER_STATE.write().unwrap();
256        if let Some(session) = state.get_mut(&token) {
257            match session.active_connection.as_ref().map(|tuple| tuple.1) {
258                Some(active_id) if active_id == connection_id => {
259                    session.active_connection = None;
260                    r_log!(Info,"[Conn {}] Cleanup complete. Session {} is now sleeping", connection_id, token);
261                }
262
263                Some(active_id) => {
264                    r_log!(Info,"[Conn {}] Cleanup for Session {} aborted. Session already has new Connection {}",
265                             connection_id, token, active_id);
266                }
267
268                None => {
269                    r_log!(Info,"[Conn {}] Cleanup for Session {} aborted. Session is already sleeping.",
270                             connection_id, token);
271                }
272            }
273        };
274    }
275    r_debug_log!(Info,"The Read-Thread  {} is dead! Long live the Read-Thread!", connection_id);
276}
277
278/// Manages outgoing data serialization and transmission to the client on a dedicated background thread.
279///
280/// Listens on the receive channel for outbound [`TcpServerMessage`] objects, serializes them,
281/// writes their length-prefixed binary representations to the TCP stream, and terminates upon
282/// encountering transmission errors or a kick notification.
283///
284/// # Arguments
285///
286/// * `rx_channel`    - The receiver channel for pulling queued server-to-client messages.
287/// * `write_stream`  - The isolated `TcpStream` clone dedicated to writing data.
288/// * `connection_id` - The unique identifier of this connection.
289fn write_thread(rx_channel: Receiver<TcpServerMessage>, mut write_stream: TcpStream, connection_id: ConnectionID) {
290    while let Ok(message) = rx_channel.recv() {
291        match bincode::serialize(&message) {
292            Ok(serialized_response) => {
293                let len = serialized_response.len() as u32;
294
295                if let Err(e) = write_stream.write_all(&len.to_be_bytes()) {
296                    r_log!(Warning,"[Conn {}] Got error while sending len to Client: {} Stopped write-Thread",
297                             connection_id, e);
298                    break;
299                }
300
301                if let Err(e) = write_stream.write_all(&serialized_response) {
302                    r_log!(Warning,"[Conn {}] Got error while sending to Client: {} Stopped write-Thread",
303                             connection_id, e);
304                    break;
305                }
306            }
307
308            Err(e) => {
309                r_log!(Error,"[Conn {}] Got error while serializing response: {}", connection_id, e);
310                break;
311            }
312        }
313
314        if let TcpServerMessage::Kicked { reason } = message {
315            r_log!(Info,"[Conn {}] Write thread terminating because client was kicked: {}", connection_id, reason);
316
317            break;
318        }
319    }
320
321    //Thread the Ripper, we kill the read-thread with us
322    let _ = write_stream.shutdown(Shutdown::Both);
323    r_debug_log!(Info,"The Write-Thread {} is dead! Long live the Write-Thread!", connection_id);
324}
325
326/// Evaluates authentication-related client messages (`Login`, `Relogin`, `Logout`) and updates global session states
327/// (via return).
328///
329/// # Arguments
330///
331/// * `message`       - The incoming client message payload.
332/// * `old_token`     - The existing session token associated with this connection context, if any.
333/// * `tx_channel`    - The transmission channel to send immediate auth feedback.
334/// * `connection_id` - The unique identifier of the connection.
335fn update_login_status(
336    message: &TcpClientMessage, old_token: Option<SessionID>, tx_channel: &Sender<TcpServerMessage>,
337    connection_id:ConnectionID
338) -> Option<SessionID> {
339    let (new_token, response) = match message {
340        TcpClientMessage::Login {password, user_name, user_role }  => {
341            if let Some(real_old_token) = old_token {
342                r_log!(UserError,
343                    "[Conn {}] User '{}' tried to login, but connection is already logged in with token {}. Ignored.",
344                    connection_id, user_name, real_old_token);
345            }
346
347            if password == "" {//TODO Save password somewhere and read it here
348                let mut state = SERVER_STATE.write().unwrap();
349
350                let mut rng = rand::rng();
351                let new_token = loop {
352                    let token:SessionID = rng.random();
353
354                    if !state.contains_key(&token) {
355                        break token;
356                    }
357
358                };
359
360                state.insert(new_token, ClientSession {
361                    _user_name: user_name.clone(),
362                    _user_role: *user_role,
363                    active_connection: Some((tx_channel.clone(), connection_id)),
364                    subscriptions: vec![]
365                });
366
367                r_log!(SuccessEvent,"[Conn {}] Logged in with Session-Token {}", connection_id, new_token);
368
369                (Some(new_token), Some(TcpServerMessage::LoginOk { token: new_token }))
370            } else {
371                r_log!(UserError,"[Conn {}] User {} wanted to login with wrong password",
372                         connection_id, user_name);
373                let _ = tx_channel.send(TcpServerMessage::LoginFailed {
374                    reason: String::from("Wrong password")
375                });
376
377                (None, Some(TcpServerMessage::LoginFailed { reason: "Wrong password".into() }))
378            }
379        }
380
381        TcpClientMessage::Relogin {user_id, clear_subscriptions} => {
382            if let Some(old_token) = old_token {
383                r_log!(UserError,"[Conn {}] User with ID {} wanted to relog with ID {}, but he was logged \
384                in. Relogin ignored", connection_id, old_token, user_id);
385                (Some(old_token), Some(TcpServerMessage::ReloginFailed { reason: "Already logged in".into()}))
386
387            }else if let Some(session) = SERVER_STATE.write().unwrap().get_mut(&user_id) {
388
389                if let Some((old_user_channel,old_connection_id)) = session.active_connection.take() {
390                    let _ = old_user_channel.send(TcpServerMessage::Kicked {
391                        reason: "Newer Connection relogged in with same token".into()
392                    });
393                    r_log!(Warning,"[Conn {}] User was kicked due to newer Connection with same token {}. \
394                    Killed old Connection {}", connection_id, user_id, old_connection_id);
395                }
396
397                session.active_connection = Some((tx_channel.clone(), connection_id));
398
399                if *clear_subscriptions {
400                    session.subscriptions.clear();
401                }
402
403                r_log!(SuccessEvent,"[Conn {}] User with ID {} relogged in  successfully", connection_id, user_id);
404                (Some(*user_id), Some(TcpServerMessage::ReloginOk { token: *user_id}))
405            } else {
406                r_log!(Warning,"[Conn {}] User wanted to relogin with ID {}, wich doesnt exist",
407                         connection_id, user_id);
408                (None, Some(TcpServerMessage::ReloginFailed { reason: "User doesnt exist".into() }))
409            }
410        }
411
412        TcpClientMessage::Logout => {
413            if let Some(old_token) = old_token {
414                let mut state = SERVER_STATE.write().unwrap();
415
416                state.remove(&old_token);
417                r_log!(SuccessEvent,"[Conn {}] Client logged out successfully", connection_id);
418            } else {
419                //User wanted to log out, but wasn't logged in in the first place
420                //No log message here, because everyone is happy
421            }
422            (None, Some(LogoutOk))
423        }
424
425        _ => (old_token, None)
426    };
427
428    if let Some(response) = response {
429        if let Err(e) = tx_channel.send(response) {
430            r_log!(Error,"[Conn {}] Error sending the auth response to the channel: {}", connection_id, e);
431        }
432    }
433
434    new_token
435}
436
437/// Routes general, authorized client messages to appropriate engine functions or command handlers.
438///
439/// # Arguments
440///
441/// * `msg`           - The deserialized command.
442/// * `connection_id` - The unique identifier of the active connection.
443/// * `token`         - The verified session identifier of the caller.
444fn handle_messages(msg: TcpClientMessage, connection_id: ConnectionID, token: SessionID) -> Option<TcpServerMessage> {
445    match msg {
446        TcpClientMessage::Login {..} | TcpClientMessage::Logout | TcpClientMessage::Relogin {..} => unreachable!(),
447
448
449        TcpClientMessage::Subscribe {topic, update_mode} => {
450            add_subscription(&token, &topic, &update_mode);
451            r_log!(Info,"[Conn {}] {}", connection_id, match update_mode {
452                UpdateMode::OnChange => format!("Client requested updates on changes for {}!", topic),
453                UpdateMode::Continuous => format!("Client requested continuous updates for {}", topic),
454            }); //TODO Normally we always should respond to this
455            None
456        }
457
458        TcpClientMessage::Unsubscribe { topic } => {
459            r_log!(Info, "[Conn {}] Client unsubscribed from {}.", connection_id, topic);
460            None
461        }
462
463        TcpClientMessage::ExecuteCommand{ command, response_id} => {
464            let answer = run_command(false, command);
465            r_log!(answer.0, "[Conn {}] {}", connection_id, answer.1);
466            Some(CommandOutput{
467                answer,
468                response_id
469            })
470        }
471
472        TcpClientMessage::ExecuteImplicitCommand { command, response_id} => {
473            let answer = execute_implicit_cli_action(&command);
474            //TODO Add logging for this
475            Some(ImplicitCommandOutput {
476                answer,
477                response_id
478            })
479        }
480
481        TcpClientMessage::RequestEdit(_) => {
482            r_log!(Info, "[Conn {}] Client requested resource edit lock", connection_id);
483            None
484        }
485
486        TcpClientMessage::SubmitEdit {resource:_, new_data:_} => {
487            r_log!(Info,"[Conn {}] User submitted change to resource", connection_id);
488            None
489        },
490        TcpClientMessage::DropEditLock(_) => {
491            r_log!(Info, "[Conn {}] User dropped resource edit lock", connection_id);
492            None
493        }
494    }
495}
kernel/networking/server_sockets.rs

kernel/networking/
server_sockets.rs
