From 058432130628d00764663ecf819a9adc6793768d Mon Sep 17 00:00:00 2001 From: salmonik Date: Sun, 8 Nov 2026 03:29:35 +0100 Subject: v5.0.0: protocol v5 and the browser frontend Protocol v5 replaces newline lines with bounded [u32 len][text] frames and REQ/RES/EVT envelopes. Requests carry correlation ids and the server replays the last 256 replies per connection, so a gateway retry cannot double-debit and a late reply cannot clobber a newer request. HELLO/PONG carry server time; crash rounds announce started_at and push coalesced 10 Hz ticks. Crash now commits sha256(seed) before a round and reveals the seed when it ends, so clients recompute the crash point. Adds the webd/ gateway: one blocking thread, one ClientCore and one pinned upstream TLS connection per session, an HttpOnly session cookie, takeover, and a 60-second grace window. The browser sends Command JSON and renders View JSON; it never speaks the game protocol or stores credentials. Adds the vanilla TypeScript browser app at /play/ (login, lobby, all four games, history, notices, crash verification), the nginx /play and /ws routes, and the headless DOM smoke test. webd and the app stay out of the TUI/GUI/server build graph. The client/server version moves to 5.0.0 in lockstep; v4 peers are refused at HELLO. --- src/bin/client/connection.rs | 6 +- src/bin/client/repl.rs | 95 +++++++++--------- src/bin/server/handler.rs | 47 ++++++--- src/bin/server/tls.rs | 27 +++--- src/client/core.rs | 182 +++++++++++++++++++++++++--------- src/client/games/manager.rs | 21 ++-- src/client/games/roulette.rs | 9 +- src/client/transport.rs | 171 ++++++++++++++++++++------------ src/client/view.rs | 60 +++++++++--- src/crash.rs | 54 +++++++++++ src/protocol.rs | 226 +++++++++++++++++++++++++++++++++++++------ src/server/core.rs | 81 ++++++++++++++-- src/server/games/crash.rs | 76 +++++++++++++-- src/version.rs | 34 ++++++- 14 files changed, 829 insertions(+), 260 deletions(-) (limited to 'src') diff --git a/src/bin/client/connection.rs b/src/bin/client/connection.rs index f88f62b..e808d90 100644 --- a/src/bin/client/connection.rs +++ b/src/bin/client/connection.rs @@ -8,7 +8,7 @@ use parking_lot::Mutex; use std::error::Error; use zagan_bet::client::transport::{self, TrustPrompt}; use zagan_bet::client::ui::first_contact_lines; -use zagan_bet::version::{is_newer, VERSION}; +use zagan_bet::version::{VERSION, is_newer}; pub use zagan_bet::client::transport::TlsStream; @@ -21,7 +21,7 @@ pub fn set_unsafe_mode(unsafe_mode: bool) { pub fn connect( addr: &str, conn: &rusqlite::Connection, -) -> Result<(TlsStream, Option), Box> { +) -> Result<(TlsStream, Option, u64), Box> { let unsafe_mode = *UNSAFE_MODE.lock(); // The terminal has no dialog: print the first-contact notice and trust. @@ -53,5 +53,5 @@ pub fn connect( println!("Authentication failed: {error}"); } - Ok((connected.stream, connected.user)) + Ok((connected.stream, connected.user, connected.server_time)) } diff --git a/src/bin/client/repl.rs b/src/bin/client/repl.rs index cfdcdd5..deb5743 100644 --- a/src/bin/client/repl.rs +++ b/src/bin/client/repl.rs @@ -14,15 +14,15 @@ use crossterm::{ use sha2::{Digest, Sha256}; use zagan_bet::client::core::{ClientCore, Effect}; -use zagan_bet::client::input::{game_input, Editor, GameInput, Input, Key}; -use zagan_bet::client::shell::{self, Shell}; -use zagan_bet::client::view::{RoulettePhase, View}; use zagan_bet::client::database; +use zagan_bet::client::input::{Editor, GameInput, Input, Key, game_input}; +use zagan_bet::client::shell::{self, Shell}; use zagan_bet::client::ui::{ - display_width, format_pong, help_lines, host_lines, input_cursor_column, - lines_to_move_up, terminal_size_warning, visual_line_count, MIN_COLUMNS, MIN_ROWS, + MIN_COLUMNS, MIN_ROWS, display_width, format_pong, help_lines, host_lines, input_cursor_column, + lines_to_move_up, terminal_size_warning, visual_line_count, }; -use zagan_bet::protocol::io_failure_message; +use zagan_bet::client::view::{RoulettePhase, View}; +use zagan_bet::protocol::{Packet, io_failure_message}; use zagan_bet::roulette::{WHEEL, color as roulette_color}; use crate::connection::TlsStream; @@ -78,9 +78,10 @@ impl ClientState { pub fn connect_to(&mut self, addr: &str) { self.addr = Some(addr.to_string()); match crate::connection::connect(addr, &self.conn) { - Ok((stream, authed_user)) => { + Ok((stream, authed_user, server_time)) => { self.stream = Some(stream); self.core.set_connected(true); + self.core.set_server_time(server_time); self.core.set_logged_in(authed_user); let fingerprint = database::get_fingerprint(&self.conn, addr); self.core.set_host_info(Some(addr.to_string()), fingerprint); @@ -126,9 +127,7 @@ impl ClientState { key, ) { GameInput::Quit => vec![Effect::Quit], - GameInput::Command(command) => { - self.core.command(command) - } + GameInput::Command(command) => self.core.command(command), GameInput::None => Vec::new(), } } else { @@ -189,10 +188,10 @@ impl ClientState { let Some(mut stream) = self.stream.take() else { return false; }; - let line = match stream.read_available_line() { - Ok(line) => { + let packet = match stream.read_available_packet() { + Ok(packet) => { self.stream = Some(stream); - line + packet } Err(_) => { self.core.set_connected(false); @@ -201,13 +200,15 @@ impl ClientState { } }; - let Some(line) = line else { + let Some(packet) = packet else { return false; }; - let effects = match self.core.handle_async_line(&line) { - Some(effects) => effects, - None => self.core.handle_reply(&line), + let effects = match packet { + Packet::Event { body, .. } => { + self.core.handle_async_line(&body).unwrap_or_default() + } + Packet::Reply { id, body } => self.core.handle_reply_id(id, &body), }; if self.apply_effects(effects) { return true; @@ -259,13 +260,13 @@ impl ClientState { self.draw_region(&lines); } Effect::ClearRegion => self.clear_region(), - Effect::SendLine(line) => { - let is_ping = line == "PING"; + Effect::SendRequest { id, payload } => { + let is_ping = payload == "PING"; let started = Instant::now(); - let reply = self.send_and_read(&line); + let reply = self.send_and_read(id, &payload); let elapsed = started.elapsed(); - let mut more = self.core.handle_reply(&reply); - if is_ping && reply.trim() == "PONG" { + let mut more = self.core.handle_reply_id(id, &reply); + if is_ping && reply.trim().starts_with("PONG") { for effect in &mut more { if let Effect::Print(lines) = effect { for text in lines { @@ -488,31 +489,30 @@ impl ClientState { self.prompt_text = text.to_string(); } - fn send_and_read(&mut self, line: &str) -> String { + fn send_and_read(&mut self, id: u64, line: &str) -> String { let Some(mut stream) = self.stream.take() else { return "ERROR Not connected".to_string(); }; // A crash event can be pushed while a request is in flight. Route those - // lines as they arrive and keep waiting for the actual reply. + // events as they arrive and keep waiting for the reply with this id. let mut pushed: Vec = Vec::new(); let result = (|| -> std::io::Result { - stream.write_line(line)?; + stream.write_request(id, None, line)?; loop { - let mut response = String::new(); - let read = stream.read_line(&mut response)?; - if read == 0 { - return Err(std::io::Error::new( - std::io::ErrorKind::UnexpectedEof, - "connection closed", - )); - } - let response = response.trim_end().to_string(); - if self.core.is_async(&response) { - pushed.push(response); - continue; + match stream.read_packet()? { + None => { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "connection closed", + )); + } + Some(Packet::Event { body, .. }) => pushed.push(body), + Some(Packet::Reply { id: reply_id, body }) if reply_id == id => { + return Ok(body); + } + Some(Packet::Reply { .. }) => continue, } - return Ok(response); } })(); @@ -614,7 +614,11 @@ impl ClientState { short_fingerprint(&identity), host ), - Ok(false) => println!("No saved login for server {} ({}).", short_fingerprint(&identity), host), + Ok(false) => println!( + "No saved login for server {} ({}).", + short_fingerprint(&identity), + host + ), Err(error) => println!("Failed to forget the saved login: {}", error), } } @@ -625,10 +629,9 @@ impl ClientState { return; }; match database::forget_host(&self.conn, &target) { - Ok(outcome) if outcome.credentials_removed => println!( - "Cleared trust and stored credentials for {}.", - target - ), + Ok(outcome) if outcome.credentials_removed => { + println!("Cleared trust and stored credentials for {}.", target) + } Ok(outcome) => { println!("Cleared trust for {}.", target); if !outcome.siblings.is_empty() { @@ -665,7 +668,8 @@ impl ClientState { let hashed = format!("{:x}", Sha256::digest(password.as_bytes())); let message = format!("AUTH {} {}", username, hashed); - let response = self.send_and_read(&message); + let id = self.core.allocate_request_id(); + let response = self.send_and_read(id, &message); let response = response.trim(); if response.starts_with("OK") { @@ -709,7 +713,8 @@ impl ClientState { let hashed = format!("{:x}", Sha256::digest(password.as_bytes())); let message = format!("REGISTER {} {}", username, hashed); - let response = self.send_and_read(&message); + let id = self.core.allocate_request_id(); + let response = self.send_and_read(id, &message); let response = response.trim(); if response == format!("REGISTERED {}", username) { diff --git a/src/bin/server/handler.rs b/src/bin/server/handler.rs index dffae69..e6b623f 100644 --- a/src/bin/server/handler.rs +++ b/src/bin/server/handler.rs @@ -1,6 +1,7 @@ use std::time::Duration; use crate::tls::TlsStream; +use zagan_bet::protocol::{event_line, parse_request, reply_line}; use zagan_bet::server::core::{ConnectionState, ServerCore}; /// How long the connection waits for a request before advancing timed games. @@ -30,24 +31,35 @@ pub fn handle_client_tls(mut stream: TlsStream) { } let mut state = ConnectionState::default(); + let mut event_seq = 0u64; println!("[TLS Client connected] {client_addr}"); 'connection: loop { - let mut line = String::new(); - match stream.read_line(&mut line) { + let mut frame = String::new(); + match stream.read_frame(&mut frame) { Ok(0) => break, Ok(_) => { - let line = line.trim(); - if !line.is_empty() { - let user = state.user().unwrap_or("guest").to_string(); - println!("[{user} @ {client_addr}] Received: {line}"); - let response = core.handle_message(&mut state, line); - if let Err(error) = stream.write_line(&response) { - eprintln!("Write error: {error}"); - break; + let user = state.user().unwrap_or("guest").to_string(); + println!("[{user} @ {client_addr}] Received: {frame}"); + match parse_request(&frame) { + Ok(request) => { + let response = core.handle_request(&mut state, &request); + let reply = reply_line(request.id, &response); + if let Err(error) = stream.write_frame(&reply) { + eprintln!("Write error: {error}"); + break; + } + let user = state.user().unwrap_or("guest"); + println!("[{user} @ {client_addr}] Sent: {reply}"); + } + Err(message) => { + let reply = reply_line(0, &format!("ERROR {message}")); + if let Err(error) = stream.write_frame(&reply) { + eprintln!("Write error: {error}"); + break; + } + eprintln!("Malformed request from {client_addr}: {reply}"); } - let user = state.user().unwrap_or("guest"); - println!("[{user} @ {client_addr}] Sent: {response}"); } } Err(error) @@ -65,12 +77,17 @@ pub fn handle_client_tls(mut stream: TlsStream) { } for push in core.poll(&state) { - if let Err(error) = stream.write_line(&push) { + event_seq = event_seq.wrapping_add(1); + let event = event_line(event_seq, &push); + if let Err(error) = stream.write_frame(&event) { eprintln!("Write error: {error}"); break 'connection; } - let user = state.user().unwrap_or("guest"); - println!("[{user} @ {client_addr}] Pushed: {push}"); + // Ticks are 10 Hz per live round; logging them floods the disk. + if !push.starts_with("crash.tick ") { + let user = state.user().unwrap_or("guest"); + println!("[{user} @ {client_addr}] Pushed: {event}"); + } } } diff --git a/src/bin/server/tls.rs b/src/bin/server/tls.rs index 420b5b4..eb94637 100644 --- a/src/bin/server/tls.rs +++ b/src/bin/server/tls.rs @@ -1,7 +1,7 @@ use std::io::{Read, Write}; use std::net::TcpStream; -use zagan_bet::protocol::LineBuffer; +use zagan_bet::protocol::{FrameBuffer, write_frame}; #[derive(Clone)] pub struct TlsServer { @@ -44,34 +44,31 @@ impl TlsServer { pub struct TlsStream { stream: native_tls::TlsStream, - lines: LineBuffer, + frames: FrameBuffer, } impl TlsStream { pub fn new(stream: native_tls::TlsStream) -> Self { Self { stream, - lines: LineBuffer::new(), + frames: FrameBuffer::new(), } } - pub fn read_line(&mut self, s: &mut String) -> std::io::Result { - match self.lines.read_line(&mut self.stream)? { - Some(line) => { - s.push_str(&line); - s.push('\n'); - Ok(line.len() + 1) + /// Read one protocol frame. `Ok(0)` is EOF with no partial frame. + pub fn read_frame(&mut self, out: &mut String) -> std::io::Result { + match self.frames.read_frame(&mut self.stream)? { + Some(frame) => { + let length = frame.len(); + out.push_str(&frame); + Ok(length) } None => Ok(0), } } - pub fn write_line(&mut self, msg: &str) -> std::io::Result<()> { - use std::io::Write; - self.stream.write_all(msg.as_bytes())?; - self.stream.write_all(b"\n")?; - self.stream.flush()?; - Ok(()) + pub fn write_frame(&mut self, msg: &str) -> std::io::Result<()> { + write_frame(&mut self.stream, msg) } pub fn peer_addr(&self) -> std::io::Result { diff --git a/src/client/core.rs b/src/client/core.rs index 6d8294c..278dc00 100644 --- a/src/client/core.rs +++ b/src/client/core.rs @@ -3,17 +3,16 @@ use crate::client::games::response::{ parse_blackjack_response, parse_crash_response, parse_history_response, parse_roulette_response, parse_spin_response, }; +use crate::client::ui::{game_lines, history_lines}; use crate::client::view::{ BlackjackView, Command, CrashView, HistoryView, LobbyView, RouletteView, SlotsView, View, }; -use crate::client::ui::{game_lines, history_lines}; use crate::crash::{CrashState, multiplier_at}; -use crate::history::{PlayRecord, MAX_HISTORY}; +use crate::history::{MAX_HISTORY, PlayRecord}; use crate::roulette::{BetTarget, Bets}; +use crate::version::{parse_pong, unix_time_ms}; use std::time::{Duration, Instant}; - - /// High-level actions produced by the client core. The terminal adapter is /// responsible for turning these into crossterm calls and network writes. #[derive(Debug, Clone, PartialEq)] @@ -24,8 +23,8 @@ pub enum Effect { Render(View), /// Remove the game board region. ClearRegion, - /// Send one protocol line and feed the reply to `handle_reply`. - SendLine(String), + /// Send one enveloped request and route its reply back by `id`. + SendRequest { id: u64, payload: String }, /// Drop the connection. Disconnect, /// Leave the program. @@ -61,6 +60,10 @@ pub struct ClientCore { connected: bool, logged_in: Option, pending: Option, + /// Id of the request whose reply fills `pending`. + pending_id: Option, + /// Next correlation id allocated on this connection. + next_request_id: u64, balance: Option, /// Account number from the last `INFO`, shown by both frontends. account: Option, @@ -68,6 +71,13 @@ pub struct ClientCore { fingerprint: Option, /// Local anchor for the crash multiplier, set when a round starts. crash_started: Option, + /// Server-time anchor for the crash multiplier, when the server supplied + /// `started_at`. + crash_started_ms: Option, + /// Server clock minus local clock, in milliseconds, from `HELLO`/`PONG`. + server_offset_ms: i64, + /// True once a frontend has supplied a server timestamp. + clock_synced: bool, /// App screen shown when no game is open. screen: Screen, /// Last plays fetched with `history`. @@ -83,11 +93,16 @@ impl ClientCore { connected: false, logged_in: None, pending: None, + pending_id: None, + next_request_id: 1, balance: None, account: None, host: None, fingerprint: None, crash_started: None, + crash_started_ms: None, + server_offset_ms: 0, + clock_synced: false, screen: Screen::Lobby, play_history: Vec::new(), login_error: None, @@ -99,11 +114,19 @@ impl ClientCore { if !connected { self.logged_in = None; self.pending = None; + self.pending_id = None; self.crash_started = None; + self.crash_started_ms = None; + self.clock_synced = false; self.screen = Screen::Lobby; } } + /// The signed-in username, if any. + pub fn user(&self) -> Option<&str> { + self.logged_in.as_deref() + } + pub fn set_logged_in(&mut self, user: Option) { if user.is_some() { self.login_error = None; @@ -147,7 +170,44 @@ impl ClientCore { self.board_effects() } + /// Record the server's clock and the local time it was received. The + /// offset is for display; cash-out still settles on the server. + pub fn set_server_time(&mut self, server_ms: u64) { + self.server_offset_ms = server_ms as i64 - unix_time_ms() as i64; + self.clock_synced = true; + } + + /// The game server's clock, as this client currently estimates it. + pub fn server_now_ms(&self) -> u64 { + (unix_time_ms() as i64 + self.server_offset_ms).max(0) as u64 + } + + /// Allocate a correlation id for a request the adapter sends directly + /// (the terminal's login prompt bypasses `ClientCore`). + pub fn allocate_request_id(&mut self) -> u64 { + let id = self.next_request_id; + self.next_request_id = self.next_request_id.wrapping_add(1).max(1); + id + } + + fn begin_request(&mut self, pending: Pending, payload: String) -> Effect { + let id = self.allocate_request_id(); + self.pending = Some(pending); + self.pending_id = Some(id); + Effect::SendRequest { id, payload } + } + + /// Route a reply by correlation id. A reply we no longer expect (a late + /// response after a timeout) is ignored instead of corrupting the board. + pub fn handle_reply_id(&mut self, id: u64, reply: &str) -> Vec { + if self.pending_id != Some(id) { + return Vec::new(); + } + self.handle_reply(reply) + } + pub fn handle_reply(&mut self, reply: &str) -> Vec { + self.pending_id = None; match self.pending.take() { Some(Pending::Spin) => match parse_spin_response(reply) { Ok(result) => { @@ -265,7 +325,8 @@ impl ClientCore { } }, Some(Pending::Ping) => { - if reply.trim() == "PONG" { + if let Some(server_ms) = parse_pong(reply) { + self.set_server_time(server_ms); print_line("Pong") } else { self.connected = false; @@ -305,10 +366,8 @@ impl ClientCore { if self.game.begin_spin().is_err() { return Vec::new(); } - self.pending = Some(Pending::Spin); - let mut effects = self.board_effects(); - effects.push(Effect::SendLine(request)); + effects.push(self.begin_request(Pending::Spin, request)); effects } @@ -322,24 +381,24 @@ impl ClientCore { return print_line("Not connected. Use 'connect :' first."); } self.login_error = None; - self.pending = Some(Pending::Login { user: user.clone() }); - vec![Effect::SendLine(format!("AUTH {user} {credential}"))] + vec![self.begin_request( + Pending::Login { user: user.clone() }, + format!("AUTH {user} {credential}"), + )] } Command::Register { user, credential } => { if !self.connected { return print_line("Not connected. Use 'connect :' first."); } self.login_error = None; - self.pending = Some(Pending::Register); - vec![Effect::SendLine(format!("REGISTER {user} {credential}"))] + vec![self.begin_request(Pending::Register, format!("REGISTER {user} {credential}"))] } Command::Games => vec![Effect::Print(game_lines(&self.games()))], Command::Ping => { if !self.connected { return print_line("Not connected. Use 'connect :' first."); } - self.pending = Some(Pending::Ping); - vec![Effect::SendLine("PING".to_string())] + vec![self.begin_request(Pending::Ping, "PING".to_string())] } Command::Echo(message) => { if !self.connected { @@ -348,8 +407,7 @@ impl ClientCore { if message.trim().is_empty() { return print_line("Usage: send "); } - self.pending = Some(Pending::Echo); - vec![Effect::SendLine(message)] + vec![self.begin_request(Pending::Echo, message)] } Command::Info => self.info_effects(), Command::Play { game, bet } => self.play_game(&game, bet), @@ -458,6 +516,7 @@ impl ClientCore { fn exit_game(&mut self) -> Vec { self.crash_started = None; + self.crash_started_ms = None; self.screen = Screen::Lobby; self.game.end_game(); vec![ @@ -475,19 +534,17 @@ impl ClientCore { return self.exit_game(); }; self.game.set_blackjack_status("forfeiting"); - self.pending = Some(Pending::BlackjackForfeit); - vec![Effect::SendLine(format!( - "BLACKJACK_ACTION {} forfeit", - hand.session_id - ))] + vec![self.begin_request( + Pending::BlackjackForfeit, + format!("BLACKJACK_ACTION {} forfeit", hand.session_id), + )] } fn forfeit_crash(&mut self) -> Vec { let Some(session) = self.game.crash_session_id() else { return self.exit_game(); }; - self.pending = Some(Pending::CrashForfeit); - vec![Effect::SendLine(format!("CRASH_FORFEIT {session}"))] + vec![self.begin_request(Pending::CrashForfeit, format!("CRASH_FORFEIT {session}"))] } fn place_roulette_chip_on(&mut self, target: BetTarget) -> Vec { @@ -533,10 +590,8 @@ impl ClientCore { if self.game.begin_spin().is_err() { return Vec::new(); } - self.pending = Some(Pending::RouletteSpin); - let mut effects = self.board_effects(); - effects.push(Effect::SendLine(request)); + effects.push(self.begin_request(Pending::RouletteSpin, request)); effects } @@ -566,9 +621,8 @@ impl ClientCore { } self.game.set_blackjack_status("dealing"); - self.pending = Some(Pending::BlackjackStart); let mut effects = self.board_effects(); - effects.push(Effect::SendLine(request)); + effects.push(self.begin_request(Pending::BlackjackStart, request)); effects } @@ -582,15 +636,21 @@ impl ClientCore { if action == "double" && !hand.actions.iter().any(|available| available == "double") { return Vec::new(); } - self.pending = Some(Pending::BlackjackAction); - vec![Effect::SendLine(format!( - "BLACKJACK_ACTION {} {}", - hand.session_id, action - ))] + vec![self.begin_request( + Pending::BlackjackAction, + format!("BLACKJACK_ACTION {} {}", hand.session_id, action), + )] } /// Redraw the rising multiplier. Called by the terminal loop on a timer. pub fn tick(&mut self) -> Vec { + if self.clock_synced { + if let Some(started_ms) = self.crash_started_ms { + let now = unix_time_ms() as i64 + self.server_offset_ms; + let elapsed_ms = (now - started_ms as i64).max(0) as u64; + return self.tick_at(Duration::from_millis(elapsed_ms)); + } + } let Some(started) = self.crash_started else { return Vec::new(); }; @@ -619,12 +679,16 @@ impl ClientCore { } pub fn is_async(&self, line: &str) -> bool { - line.trim_start().starts_with("CRASH_BUSTED ") + let line = line.trim_start(); + line.starts_with("CRASH_BUSTED ") || line.starts_with("crash.tick ") } /// Handle a server-initiated line. Returns `None` when the line is a normal /// request/response reply and belongs to `handle_reply`. pub fn handle_async_line(&mut self, line: &str) -> Option> { + if let Some(tick) = line.trim_start().strip_prefix("crash.tick ") { + return Some(self.apply_crash_tick(tick)); + } if !self.is_async(line) { return None; } @@ -643,13 +707,43 @@ impl ClientCore { if self.crash_started.is_none() { self.crash_started = Some(Instant::now()); } + if state.started_at > 0 { + self.crash_started_ms = Some(state.started_at); + } } else { self.crash_started = None; + self.crash_started_ms = None; } self.game.apply_crash_state(state); self.board_effects() } + /// Apply `crash.tick session= t= m=`. + fn apply_crash_tick(&mut self, tick: &str) -> Vec { + let mut session = None; + let mut multiplier = None; + for token in tick.split_whitespace() { + if let Some(value) = token.strip_prefix("session=") { + session = Some(value); + } else if let Some(value) = token.strip_prefix("m=") { + multiplier = value.parse::().ok(); + } + } + let Some(multiplier) = multiplier else { + return Vec::new(); + }; + if let Some(session) = session { + if self.game.crash_session_id().as_deref() != Some(session) { + return Vec::new(); + } + } + if self.game.crash_status() != "running" { + return Vec::new(); + } + self.game.set_crash_multiplier(multiplier); + self.board_effects() + } + fn start_crash(&mut self) -> Vec { let status = self.game.crash_status(); if !matches!(status.as_str(), "ready" | "cashed" | "crashed" | "forfeit") { @@ -671,9 +765,8 @@ impl ClientCore { } self.game.set_crash_status("starting"); - self.pending = Some(Pending::CrashStart); let mut effects = self.board_effects(); - effects.push(Effect::SendLine(request)); + effects.push(self.begin_request(Pending::CrashStart, request)); effects } @@ -681,8 +774,7 @@ impl ClientCore { let Some(session) = self.game.crash_session_id() else { return Vec::new(); }; - self.pending = Some(Pending::CrashCashout); - vec![Effect::SendLine(format!("CRASH_CASHOUT {session}"))] + vec![self.begin_request(Pending::CrashCashout, format!("CRASH_CASHOUT {session}"))] } fn play_game(&mut self, game_type: &str, bet: Option) -> Vec { @@ -706,8 +798,7 @@ impl ClientCore { } let mut effects = self.board_effects(); - self.pending = Some(Pending::Info); - effects.push(Effect::SendLine("INFO".to_string())); + effects.push(self.begin_request(Pending::Info, "INFO".to_string())); effects } Err(_) => print_line("Failed to enter game"), @@ -726,13 +817,12 @@ impl ClientCore { } self.screen = Screen::History; - self.pending = Some(Pending::History); let request = if limit == crate::history::DEFAULT_HISTORY { "HISTORY".to_string() } else { format!("HISTORY {limit}") }; - vec![Effect::SendLine(request)] + vec![self.begin_request(Pending::History, request)] } fn info_effects(&mut self) -> Vec { @@ -754,8 +844,7 @@ impl ClientCore { ])), } - self.pending = Some(Pending::Info); - effects.push(Effect::SendLine("INFO".to_string())); + effects.push(self.begin_request(Pending::Info, "INFO".to_string())); effects } @@ -809,7 +898,6 @@ impl ClientCore { } self.view().map(Effect::Render).into_iter().collect() } - } fn auth_result(reply: &str) -> Result<(), String> { diff --git a/src/client/games/manager.rs b/src/client/games/manager.rs index 24ceeaa..d35308b 100644 --- a/src/client/games/manager.rs +++ b/src/client/games/manager.rs @@ -5,7 +5,7 @@ use super::roulette::RouletteGame; use super::slots::SlotsGame; use super::traits::{Game, GameError, GamePhase, GameState}; use crate::blackjack::BlackjackState; -use crate::crash::{payout, CrashState}; +use crate::crash::{CrashState, payout}; use crate::roulette::Bets; use std::collections::HashMap; @@ -244,9 +244,7 @@ impl GameManager { pub fn set_roulette_cursor(&mut self, cursor: &crate::client::games::roulette::Cursor) { if let Some(session) = self.current_session.as_mut() { - session - .data - .insert("cursor".to_string(), cursor.encoded()); + session.data.insert("cursor".to_string(), cursor.encoded()); } } @@ -339,7 +337,7 @@ impl GameManager { }; session.bet_amount = state.bet; session.data.remove("error"); - for key in ["crash_point", "payout"] { + for key in ["crash_point", "payout", "commit", "seed"] { session.data.remove(key); } session @@ -360,10 +358,21 @@ impl GameManager { session .data .insert("payout".to_string(), state.payout.to_string()); + session + .data + .insert("started_at".to_string(), state.started_at.to_string()); + if let Some(commit) = &state.commit { + session.data.insert("commit".to_string(), commit.clone()); + } + if let Some(seed) = &state.seed { + session.data.insert("seed".to_string(), seed.clone()); + } if let Some(crash_point) = state.crash_point { + // Keep full precision: the browser recomputes this from the + // revealed seed, and the GUI charts it. session .data - .insert("crash_point".to_string(), format!("{crash_point:.2}")); + .insert("crash_point".to_string(), crash_point.to_string()); } } diff --git a/src/client/games/roulette.rs b/src/client/games/roulette.rs index f0ed816..15cc6ed 100644 --- a/src/client/games/roulette.rs +++ b/src/client/games/roulette.rs @@ -1,6 +1,7 @@ use super::traits::{Game, GameState}; use crate::client::view::{RoulettePhase, RouletteView}; use crate::roulette::{BetTarget, Bets, Outside, is_even, is_odd}; +use serde::{Deserialize, Serialize}; pub const ROULETTE_BOARD_LINES: usize = 12; @@ -21,7 +22,7 @@ const OUTSIDE_ORDER: [Outside; 6] = [ pub struct RouletteGame; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum Direction { Up, Down, @@ -41,7 +42,7 @@ pub enum Direction { /// odd,even line between rows split of the two numbers /// even,even corner of four cells corner /// ``` -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum Cursor { Zero, Lattice { x: usize, y: usize }, @@ -67,7 +68,9 @@ impl Cursor { match parts.as_slice() { ["zero"] => Cursor::Zero, ["lattice", x, y] => match (x.parse::(), y.parse::()) { - (Ok(x), Ok(y)) if (1..=FELT_X_MAX).contains(&x) && (1..=FELT_Y_MAX).contains(&y) => { + (Ok(x), Ok(y)) + if (1..=FELT_X_MAX).contains(&x) && (1..=FELT_Y_MAX).contains(&y) => + { Cursor::Lattice { x, y } } _ => Cursor::default(), diff --git a/src/client/transport.rs b/src/client/transport.rs index 97c9227..60e7ab6 100644 --- a/src/client/transport.rs +++ b/src/client/transport.rs @@ -7,20 +7,27 @@ use std::fmt; use std::io::{Read, Write}; use std::net::TcpStream; -use std::sync::mpsc; use std::sync::Arc; +use std::sync::mpsc; use std::thread; use std::time::{Duration, Instant}; use parking_lot::Mutex; -use rustls::client::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; -use rustls::DigitallySignedStruct; use rusqlite::Connection; +use rustls::DigitallySignedStruct; +use rustls::client::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; use crate::client::database; -use crate::protocol::LineBuffer; -use crate::tofu::{decide, spki_fingerprint, TrustDecision}; -use crate::version::{parse_hello, HelloReply, VERSION}; +use crate::protocol::{ + FrameBuffer, Packet, parse_packet, request_line, write_frame as write_protocol_frame, +}; +use crate::tofu::{TrustDecision, decide, spki_fingerprint}; +use crate::version::{HelloReply, VERSION, parse_hello}; + +/// Reserved request ids for the transport handshake. Application requests +/// start at 1, so the two ranges cannot collide on one connection. +const HELLO_REQUEST_ID: u64 = u64::MAX; +const AUTH_REQUEST_ID: u64 = u64::MAX - 1; /// How long to wait for TCP + TLS before giving up. const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); @@ -43,6 +50,8 @@ pub struct Connected { pub server_version: String, pub latest: String, pub url: String, + /// Server Unix time reported by `HELLO`, milliseconds. + pub server_time: u64, /// Round trip of the HELLO exchange. pub elapsed: Duration, } @@ -51,10 +60,7 @@ pub struct Connected { pub enum TransportError { Timeout, TrustDeclined, - KeyChanged { - expected: String, - observed: String, - }, + KeyChanged { expected: String, observed: String }, VersionMismatch(String), Closed, Unexpected(String), @@ -121,7 +127,7 @@ pub fn connect( return Err(TransportError::KeyChanged { expected, observed: actual, - }) + }); } } } @@ -131,12 +137,7 @@ pub fn connect( .map_err(io_failure)?; let start = Instant::now(); - tls.write_line(&format!("HELLO {VERSION}")) - .map_err(io_failure)?; - let mut response = String::new(); - if tls.read_line(&mut response).map_err(io_failure)? == 0 { - return Err(TransportError::Closed); - } + let response = exchange(&mut tls, HELLO_REQUEST_ID, &format!("HELLO {VERSION}"))?; let elapsed = start.elapsed(); let response = response.trim(); @@ -151,10 +152,11 @@ pub fn connect( let mut user = None; let mut auth_error = None; if let Some((username, password)) = database::get_credentials(conn, &observed) { - tls.write_line(&format!("AUTH {username} {password}")) - .map_err(io_failure)?; - let mut auth = String::new(); - tls.read_line(&mut auth).map_err(io_failure)?; + let auth = exchange( + &mut tls, + AUTH_REQUEST_ID, + &format!("AUTH {username} {password}"), + )?; if auth.trim().starts_with("OK") { user = Some(username); } else { @@ -169,19 +171,31 @@ pub fn connect( server_version: hello.version, latest: hello.latest, url: hello.url, + server_time: hello.time, elapsed, }) } +/// Send one request and wait for the reply with the same id, ignoring events +/// that arrive in between. Used only for the handshake. +fn exchange(stream: &mut TlsStream, id: u64, payload: &str) -> Result { + stream + .write_request(id, None, payload) + .map_err(io_failure)?; + loop { + match stream.read_packet().map_err(io_failure)? { + None => return Err(TransportError::Closed), + Some(Packet::Reply { id: reply_id, body }) if reply_id == id => return Ok(body), + Some(Packet::Reply { .. }) | Some(Packet::Event { .. }) => continue, + } + } +} + /// TCP + TLS on a thread so a blocked handshake can time out. fn tls_connect(addr: &str) -> Result<(TlsStream, Option), TransportError> { let (sender, receiver) = mpsc::channel(); let addr = addr.to_string(); - let server_name = addr - .split(':') - .next() - .unwrap_or("127.0.0.1") - .to_string(); + let server_name = addr.split(':').next().unwrap_or("127.0.0.1").to_string(); thread::spawn(move || { let outcome = (|| -> Result<(TlsStream, Option), TransportError> { @@ -241,7 +255,7 @@ impl TlsConnector { session: Arc::new(Mutex::new(Session { sock, tls: connection, - lines: LineBuffer::new(), + frames: FrameBuffer::new(), read_timeout: Some(READ_TIMEOUT), })), }) @@ -268,8 +282,9 @@ impl ServerCertVerifier for RecordingVerifier { _ocsp_response: &[u8], _now: std::time::SystemTime, ) -> Result { - let fingerprint = spki_fingerprint(&end_entity.0) - .map_err(|error| rustls::Error::General(format!("certificate parse failed: {error}")))?; + let fingerprint = spki_fingerprint(&end_entity.0).map_err(|error| { + rustls::Error::General(format!("certificate parse failed: {error}")) + })?; *self.observed.lock() = Some(fingerprint); Ok(ServerCertVerified::assertion()) } @@ -313,7 +328,7 @@ impl ServerCertVerifier for RecordingVerifier { struct Session { sock: TcpStream, tls: rustls::ClientConnection, - lines: LineBuffer, + frames: FrameBuffer, read_timeout: Option, } @@ -329,37 +344,58 @@ pub struct TlsStream { } impl TlsStream { - pub fn read_line(&mut self, buf: &mut String) -> std::io::Result { + /// Write one enveloped request. Application code allocates `id`. + pub fn write_request( + &mut self, + id: u64, + idem: Option<&str>, + payload: &str, + ) -> std::io::Result<()> { + self.write_frame(&request_line(id, idem, payload)) + } + + /// Write one raw frame payload. + pub fn write_frame(&mut self, payload: &str) -> std::io::Result<()> { + let mut s = self.session.lock(); + write_protocol_frame(&mut s.tls.writer(), payload)?; + s.drive() + } + + /// Block until one whole frame arrives. + pub fn read_frame(&mut self) -> std::io::Result> { let mut s = self.session.lock(); loop { - if let Some(line) = s.lines.next_line() { - buf.push_str(&line); - buf.push('\n'); - return Ok(line.len() + 1); + match s.frames.next_frame() { + Some(Ok(frame)) => return Ok(Some(frame)), + Some(Err(error)) => { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error)); + } + None => {} } let mut chunk = [0u8; 4096]; match s.tls.reader().read(&mut chunk) { - Ok(0) => { - if let Some(rest) = s.lines.take_rest() { - let read = rest.len(); - buf.push_str(&rest); - return Ok(read); - } - return Ok(0); - } - Ok(n) => s.lines.push(&chunk[..n]), + Ok(0) => return Ok(None), + Ok(n) => s.frames.push(&chunk[..n]), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => s.drive()?, Err(error) => return Err(error), } } } - pub fn write_line(&mut self, message: &str) -> std::io::Result<()> { - let mut s = self.session.lock(); - s.tls.writer().write_all(message.as_bytes())?; - s.tls.writer().write_all(b"\n")?; - s.drive() + /// Block until one `RES` or `EVT` arrives. + pub fn read_packet(&mut self) -> std::io::Result> { + loop { + let Some(frame) = self.read_frame()? else { + return Ok(None); + }; + match parse_packet(&frame) { + Ok(packet) => return Ok(Some(packet)), + Err(error) => { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error)); + } + } + } } pub fn set_read_timeout(&self, duration: Option) -> std::io::Result<()> { @@ -368,24 +404,36 @@ impl TlsStream { s.sock.set_read_timeout(duration) } - /// Return one complete line if the socket already has one, without - /// blocking. Used to pick up server pushes between requests. - pub fn read_available_line(&mut self) -> std::io::Result> { + /// Return one packet if the socket already has one, without blocking. + /// Used to pick up server pushes between requests. + pub fn read_available_packet(&mut self) -> std::io::Result> { let mut s = self.session.lock(); - if let Some(line) = s.lines.next_line() { - return Ok(Some(line)); - } // `set_read_timeout(Some(0))` is rejected by std, so poll with the // shortest timeout the socket accepts instead of a nonblocking mode. s.sock .set_read_timeout(Some(std::time::Duration::from_millis(1)))?; - let drained = (|| -> std::io::Result<()> { + let drained = (|| -> std::io::Result> { loop { + match s.frames.next_frame() { + Some(Ok(frame)) => { + return match parse_packet(&frame) { + Ok(packet) => Ok(Some(packet)), + Err(error) => { + Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error)) + } + }; + } + Some(Err(error)) => { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error)); + } + None => {} + } + let mut buf = [0u8; 4096]; match s.tls.reader().read(&mut buf) { - Ok(0) => return Ok(()), - Ok(n) => s.lines.push(&buf[..n]), + Ok(0) => return Ok(None), + Ok(n) => s.frames.push(&buf[..n]), // The reader only drains plaintext rustls has already // decrypted; `drive` moves bytes off the socket. Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { @@ -397,7 +445,7 @@ impl TlsStream { std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut ) => { - return Ok(()); + return Ok(None); } Err(error) => return Err(error), } @@ -408,15 +456,14 @@ impl TlsStream { std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut ) => { - return Ok(()); + return Ok(None); } Err(error) => return Err(error), } } })(); s.sock.set_read_timeout(s.read_timeout)?; - drained?; - Ok(s.lines.next_line()) + drained } } diff --git a/src/client/view.rs b/src/client/view.rs index 4615720..0886d64 100644 --- a/src/client/view.rs +++ b/src/client/view.rs @@ -7,15 +7,17 @@ use crate::blackjack::BlackjackState; use crate::client::games::roulette::{Cursor, Direction}; use crate::client::games::traits::{GamePhase, GameState}; use crate::roulette::{BetTarget, Bets}; +use serde::{Deserialize, Serialize}; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum RoulettePhase { Placing, Spinning, Result, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct RouletteView { pub bet_amount: u64, pub balance: Option, @@ -51,7 +53,10 @@ impl RouletteView { Some("result") => RoulettePhase::Result, _ => RoulettePhase::Placing, }; - let number = state.data.get("number").and_then(|value| value.parse().ok()); + let number = state + .data + .get("number") + .and_then(|value| value.parse().ok()); let color = state.data.get("color").cloned(); let winnings = state .data @@ -82,14 +87,15 @@ impl RouletteView { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum SlotsPhase { Ready, Spinning, Result, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct SlotsView { pub bet: u64, pub balance: Option, @@ -149,7 +155,7 @@ impl SlotsView { } } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct BlackjackView { pub status: String, pub balance: Option, @@ -186,7 +192,7 @@ impl BlackjackView { } } -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct CrashView { pub status: String, pub balance: Option, @@ -195,6 +201,11 @@ pub struct CrashView { pub potential: u64, pub crash_point: Option, pub payout: u64, + pub started_at: u64, + /// `sha256(seed)` for this round, before it ends. + pub commit: Option, + /// The revealed seed, after the round ends. + pub seed: Option, pub error: Option, } @@ -232,6 +243,13 @@ impl CrashView { .get("payout") .and_then(|value| value.parse::().ok()) .unwrap_or(0), + started_at: state + .data + .get("started_at") + .and_then(|value| value.parse::().ok()) + .unwrap_or(0), + commit: state.data.get("commit").cloned(), + seed: state.data.get("seed").cloned(), error: state .data .get("error") @@ -242,7 +260,7 @@ impl CrashView { } /// The lobby: who is logged in, the balance, and what can be played. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct LobbyView { pub connected: bool, pub user: Option, @@ -254,13 +272,14 @@ pub struct LobbyView { } /// The play log, already parsed. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct HistoryView { pub records: Vec, } /// Everything a frontend can draw. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "screen", rename_all = "snake_case")] pub enum View { Lobby(LobbyView), History(HistoryView), @@ -271,17 +290,28 @@ pub enum View { } /// Semantic input. Frontends turn keys and clicks into these. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum Command { /// The credential is already hashed by the frontend. - Login { user: String, credential: String }, + Login { + user: String, + credential: String, + }, Games, Ping, Echo(String), Info, - Register { user: String, credential: String }, - Play { game: String, bet: Option }, - History { limit: u32 }, + Register { + user: String, + credential: String, + }, + Play { + game: String, + bet: Option, + }, + History { + limit: u32, + }, ShowLobby, Logout, SetBet(u64), diff --git a/src/crash.rs b/src/crash.rs index 9d6e98c..40b8743 100644 --- a/src/crash.rs +++ b/src/crash.rs @@ -6,6 +6,7 @@ //! clock, never from anything the client sends. use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use std::time::Duration; /// `exp(GROWTH_RATE * t)` doubles the multiplier every five seconds. @@ -47,6 +48,53 @@ pub fn crash_point_from_uniform(u: f64) -> f64 { ((1.0 - HOUSE_EDGE) / (1.0 - u)).clamp(1.0, MAX_MULTIPLIER) } +/// How many bytes a round seed contains. The first eight bytes choose the +/// crash point; the rest are revealed and hashed with it. +pub const SEED_LEN: usize = 32; + +/// Hex-encoded SHA-256 of a round seed, sent before the round starts. +pub fn commit_seed(seed: &[u8]) -> String { + let digest = Sha256::digest(seed); + format!("{digest:x}") +} + +/// Uniform draw encoded by a seed: the first eight bytes, big-endian. +pub fn seed_to_uniform(seed: &[u8]) -> f64 { + let mut bytes = [0u8; 8]; + let take = seed.len().min(8); + bytes[..take].copy_from_slice(&seed[..take]); + u64::from_be_bytes(bytes) as f64 / 18_446_744_073_709_551_616.0 +} + +/// The crash point a seed commits to. A client recomputes this after the +/// reveal and checks it against the published commit. +pub fn crash_point_from_seed(seed: &[u8]) -> f64 { + crash_point_from_uniform(seed_to_uniform(seed)) +} + +/// Decode a lowercase or uppercase hex seed. Returns `None` for odd lengths, +/// bad digits or a seed that is not exactly [`SEED_LEN`] bytes. +pub fn seed_from_hex(raw: &str) -> Option> { + if raw.len() != SEED_LEN * 2 { + return None; + } + let mut seed = Vec::with_capacity(SEED_LEN); + for pair in raw.as_bytes().chunks_exact(2) { + let high = (pair[0] as char).to_digit(16)?; + let low = (pair[1] as char).to_digit(16)?; + seed.push((high * 16 + low) as u8); + } + Some(seed) +} + +/// Check a revealed seed against the commit published before the round. +pub fn verify_seed(seed_hex: &str, commit: &str) -> bool { + let Some(seed) = seed_from_hex(seed_hex) else { + return false; + }; + commit_seed(&seed).eq_ignore_ascii_case(commit) +} + /// Whole-Kr cashout value. Balances are integers, so the payout is floored. pub fn payout(bet: u64, multiplier: f64) -> u64 { if bet == 0 || !multiplier.is_finite() || multiplier <= 0.0 { @@ -72,10 +120,16 @@ pub struct CrashState { pub multiplier: f64, /// `floor(bet * multiplier)` for the value above. pub potential: u64, + /// Server Unix time when the round started, milliseconds. + pub started_at: u64, /// Hidden until the round ends. pub crash_point: Option, pub payout: u64, pub result: String, + /// `sha256(seed)`, present for the whole round. + pub commit: Option, + /// The seed itself, present only after the round ends. + pub seed: Option, } impl CrashState { diff --git a/src/protocol.rs b/src/protocol.rs index 11f54a1..e838b08 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -1,16 +1,60 @@ -//! Newline-delimited protocol helpers shared by the client and server. +//! Protocol v5: bounded frames and the request/reply/event envelope. //! -//! TLS reads can return more than one line at a time. Keeping the leftover -//! bytes is essential: dropping them silently loses the next message. +//! Every packet is `[u32 length][UTF-8 payload]`. The length is big-endian and +//! capped, so a peer cannot make the reader allocate without bound. Inside a +//! frame the payload is text: +//! +//! ```text +//! REQ [idem=] +//! RES +//! EVT +//! ``` +//! +//! The command and body grammar is the same text protocol the games already +//! use (`HELLO`, `PLAY_START`, `GAME_RESULT`, ...). The envelope only adds +//! correlation, event numbering and a hard size limit. + +use std::fmt; +use std::io::{self, Read, Write}; + +/// Largest payload a peer may send in one frame. A game message is a few +/// hundred bytes; 64 KiB is generous and still bounds the reader. +pub const MAX_FRAME_LEN: usize = 64 * 1024; + +/// How many request ids a connection remembers for idempotent retries. +pub const REPLAY_CACHE_LEN: usize = 256; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameError { + /// The declared payload length exceeds [`MAX_FRAME_LEN`]. + TooLarge(usize), + /// The payload was not valid UTF-8. + InvalidUtf8, +} + +impl fmt::Display for FrameError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + FrameError::TooLarge(length) => { + write!( + f, + "frame of {length} bytes exceeds the {MAX_FRAME_LEN}-byte limit" + ) + } + FrameError::InvalidUtf8 => write!(f, "frame payload is not valid UTF-8"), + } + } +} -use std::io::{self, Read}; +impl std::error::Error for FrameError {} +/// A bounded frame decoder. Feed it bytes from a socket, take whole frames. #[derive(Debug, Default)] -pub struct LineBuffer { +pub struct FrameBuffer { buffer: Vec, } -impl LineBuffer { +impl FrameBuffer { pub fn new() -> Self { Self::default() } @@ -19,46 +63,172 @@ impl LineBuffer { self.buffer.extend_from_slice(bytes); } - /// Return the next complete line, including a line that ended at EOF. - pub fn next_line(&mut self) -> Option { - let end = self.buffer.iter().position(|byte| *byte == b'\n')?; - let mut line: Vec = self.buffer.drain(..=end).collect(); - line.pop(); // '\n' - if line.last() == Some(&b'\r') { - line.pop(); + /// Return the next complete frame, if the buffer holds one. + /// + /// A declared length above the limit is reported immediately instead of + /// waiting for bytes that will never be accepted; the caller should close + /// the connection. + pub fn next_frame(&mut self) -> Option> { + if self.buffer.len() < 4 { + return None; } - Some(String::from_utf8_lossy(&line).into_owned()) - } - /// Return the unterminated remainder at EOF, if any. - pub fn take_rest(&mut self) -> Option { - if self.buffer.is_empty() { + let length = u32::from_be_bytes([ + self.buffer[0], + self.buffer[1], + self.buffer[2], + self.buffer[3], + ]) as usize; + + if length > MAX_FRAME_LEN { + self.buffer.clear(); + return Some(Err(FrameError::TooLarge(length))); + } + if self.buffer.len() < 4 + length { return None; } - let mut rest = std::mem::take(&mut self.buffer); - if rest.last() == Some(&b'\r') { - rest.pop(); + + let payload: Vec = self.buffer[4..4 + length].to_vec(); + self.buffer.drain(..4 + length); + match String::from_utf8(payload) { + Ok(text) => Some(Ok(text)), + Err(_) => Some(Err(FrameError::InvalidUtf8)), } - Some(String::from_utf8_lossy(&rest).into_owned()) } - /// Read one line from a blocking reader, preserving any extra lines for - /// the next call. Returns `None` at EOF with nothing buffered. - pub fn read_line(&mut self, reader: &mut R) -> io::Result> { + /// Read one frame from a blocking reader, preserving leftover bytes. + /// `None` means EOF with no complete frame buffered. + pub fn read_frame(&mut self, reader: &mut R) -> io::Result> { loop { - if let Some(line) = self.next_line() { - return Ok(Some(line)); + match self.next_frame() { + Some(Ok(frame)) => return Ok(Some(frame)), + Some(Err(error)) => return Err(io::Error::new(io::ErrorKind::InvalidData, error)), + None => {} } let mut chunk = [0u8; 4096]; match reader.read(&mut chunk) { - Ok(0) => return Ok(self.take_rest()), + Ok(0) => return Ok(None), Ok(n) => self.push(&chunk[..n]), Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, Err(error) => return Err(error), } } } + + /// True when a partial frame is buffered (used by tests and diagnostics). + pub fn is_empty(&self) -> bool { + self.buffer.is_empty() + } +} + +/// Encode one payload as a length-prefixed frame. +pub fn encode_frame(payload: &str) -> Vec { + let length = payload.len() as u32; + let mut bytes = Vec::with_capacity(4 + payload.len()); + bytes.extend_from_slice(&length.to_be_bytes()); + bytes.extend_from_slice(payload.as_bytes()); + bytes +} + +/// Write one payload as a length-prefixed frame. +pub fn write_frame(writer: &mut W, payload: &str) -> io::Result<()> { + writer.write_all(&encode_frame(payload))?; + writer.flush() +} + +/// One client request. `id` is the correlation key; `idem` is optional +/// metadata for logs and a future cross-connection cache. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Request { + pub id: u64, + pub idem: Option, + pub payload: String, +} + +/// A frame received by a client. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Packet { + Reply { id: u64, body: String }, + Event { seq: u64, body: String }, +} + +pub fn request_line(id: u64, idem: Option<&str>, payload: &str) -> String { + match idem { + Some(idem) if !idem.is_empty() => format!("REQ {id} idem={idem} {payload}"), + _ => format!("REQ {id} {payload}"), + } +} + +pub fn reply_line(id: u64, body: &str) -> String { + format!("RES {id} {body}") +} + +pub fn event_line(seq: u64, body: &str) -> String { + format!("EVT {seq} {body}") +} + +/// Parse a `REQ` payload. Errors are safe to return to the peer. +pub fn parse_request(frame: &str) -> Result { + let frame = frame.trim_end(); + let rest = frame + .strip_prefix("REQ ") + .ok_or_else(|| "Malformed request: expected REQ".to_string())?; + + let (id_token, mut rest) = split_token(rest); + let id: u64 = id_token + .parse() + .map_err(|_| "Malformed request id".to_string())?; + + let mut idem = None; + let (next, after) = split_token(rest); + if let Some(value) = next.strip_prefix("idem=") { + if value.is_empty() { + return Err("Malformed idempotency key".to_string()); + } + idem = Some(value.to_string()); + rest = after; + } + + Ok(Request { + id, + idem, + payload: rest.to_string(), + }) +} + +/// Parse a `RES` or `EVT` payload on the client side. +pub fn parse_packet(frame: &str) -> Result { + let frame = frame.trim_end(); + if let Some(rest) = frame.strip_prefix("RES ") { + let (id_token, body) = split_token(rest); + let id: u64 = id_token + .parse() + .map_err(|_| "Malformed reply id".to_string())?; + return Ok(Packet::Reply { + id, + body: body.to_string(), + }); + } + if let Some(rest) = frame.strip_prefix("EVT ") { + let (seq_token, body) = split_token(rest); + let seq: u64 = seq_token + .parse() + .map_err(|_| "Malformed event sequence".to_string())?; + return Ok(Packet::Event { + seq, + body: body.to_string(), + }); + } + Err(format!("Unexpected packet: {frame}")) +} + +fn split_token(rest: &str) -> (&str, &str) { + let rest = rest.trim_start(); + match rest.find(' ') { + Some(index) => (&rest[..index], rest[index + 1..].trim_start()), + None => (rest, ""), + } } /// A stable, user-facing message for a failed request. diff --git a/src/server/core.rs b/src/server/core.rs index 3cd6469..0b2efb4 100644 --- a/src/server/core.rs +++ b/src/server/core.rs @@ -1,22 +1,26 @@ -use bcrypt::{hash, verify, DEFAULT_COST}; +use bcrypt::{DEFAULT_COST, hash, verify}; use rusqlite::Connection; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::sync::Arc; use crate::blackjack::BlackjackState; use crate::crash::CrashState; +use crate::protocol::{REPLAY_CACHE_LEN, Request}; use crate::server::database; -use crate::version::{download_url, hello_line, Version, VERSION}; use crate::server::games::blackjack::{BlackjackGame, BlackjackHand}; use crate::server::games::crash::{Clock, CrashGame, CrashRound, RealClock}; use crate::server::games::registry::GameRegistry; use crate::server::games::session::SessionManager; use crate::server::games::traits::{GameError, GamePhase, GameState}; +use crate::version::{VERSION, Version, download_url, hello_line, pong_line, unix_time_ms}; #[derive(Debug, Default)] pub struct ConnectionState { user: Option, greeted: bool, + /// Last replies on this connection, oldest first. Request ids are unique + /// while cached, so a retried request replays its first result. + replays: VecDeque<(u64, String, String)>, } /// Settle reservations left behind by a previous process. @@ -39,6 +43,27 @@ impl ConnectionState { pub fn greeted(&self) -> bool { self.greeted } + + fn replay(&self, id: u64, payload: &str) -> Option> { + self.replays + .iter() + .find(|(cached_id, _, _)| *cached_id == id) + .map(|(_, cached_payload, reply)| { + if cached_payload == payload { + Ok(reply.clone()) + } else { + Err("ERROR Request id reused with a different command".to_string()) + } + }) + } + + fn remember(&mut self, id: u64, payload: &str, reply: &str) { + if self.replays.len() >= REPLAY_CACHE_LEN { + self.replays.pop_front(); + } + self.replays + .push_back((id, payload.to_string(), reply.to_string())); + } } pub struct ServerCore { @@ -49,6 +74,8 @@ pub struct ServerCore { blackjack: BlackjackGame, crash: CrashGame, clock: Arc, + /// Last time a crash tick was pushed, in seconds on the game clock. + last_tick: HashMap, } impl ServerCore { @@ -65,6 +92,7 @@ impl ServerCore { blackjack: BlackjackGame::new(), crash: CrashGame::new(), clock: Arc::new(RealClock::new()), + last_tick: HashMap::new(), } } @@ -96,6 +124,21 @@ impl ServerCore { Ok(Self::new(conn)) } + /// Handle one enveloped request, replaying a cached reply for a repeated + /// request id. The payload grammar is unchanged; the envelope only adds + /// correlation and idempotency. + pub fn handle_request(&mut self, state: &mut ConnectionState, request: &Request) -> String { + if let Some(reply) = state.replay(request.id, &request.payload) { + return match reply { + Ok(reply) => reply, + Err(message) => message, + }; + } + let reply = self.handle_message(state, &request.payload); + state.remember(request.id, &request.payload, &reply); + reply + } + pub fn handle_message(&mut self, state: &mut ConnectionState, msg: &str) -> String { let parts: Vec<&str> = msg.splitn(5, ' ').collect(); @@ -201,7 +244,7 @@ impl ServerCore { } } } - "PING" => "PONG".to_string(), + "PING" => pong_line(unix_time_ms()), "INFO" => self.handle_info(state), _ => msg.to_string(), } @@ -212,7 +255,8 @@ impl ServerCore { return "ERROR Invalid HELLO format. Usage: HELLO ".to_string(); } - let (Some(client), Some(server)) = (Version::parse(parts[1]), Version::parse(VERSION)) else { + let (Some(client), Some(server)) = (Version::parse(parts[1]), Version::parse(VERSION)) + else { return "ERROR Invalid version".to_string(); }; @@ -225,7 +269,7 @@ impl ServerCore { } state.greeted = true; - hello_line(VERSION, VERSION, &download_url()) + hello_line(VERSION, VERSION, &download_url(), unix_time_ms()) } fn handle_register(&mut self, username: &str, credential: &str) -> String { @@ -529,8 +573,8 @@ impl ServerCore { // The session is the audit guard for the deduction, so it exists before // the balance is touched. - let crash_point = self.crash.next_crash_point(); - let round = CrashRound::new(bet as u64, crash_point, self.clock.now()); + let seed = self.crash.next_seed(); + let round = CrashRound::new(bet as u64, seed, self.clock.now(), unix_time_ms()); let mut data = HashMap::new(); data.insert("round".to_string(), round.to_json()); data.insert("balance".to_string(), balance.to_string()); @@ -643,11 +687,27 @@ impl ServerCore { if !round.is_running() { continue; } - let elapsed = (self.clock.now() - round.started_at).max(0.0); - if crate::crash::multiplier_at_secs(elapsed) < round.crash_point { + let now = self.clock.now(); + let elapsed = (now - round.started_at).max(0.0); + let multiplier = crate::crash::multiplier_at_secs(elapsed); + if multiplier < round.crash_point { + let last = self + .last_tick + .get(&session.session_id) + .copied() + .unwrap_or(f64::NEG_INFINITY); + if now - last >= 0.1 { + self.last_tick.insert(session.session_id.clone(), now); + out.push(format!( + "crash.tick session={} t={} m={multiplier:.4}", + session.session_id, + unix_time_ms() + )); + } continue; } round.crash(); + self.last_tick.remove(&session.session_id); match self.settle_crash(user, &session.session_id, &round, 0) { Ok(balance) => { let state = round.state(&session.session_id, balance); @@ -678,6 +738,7 @@ impl ServerCore { payout as i64, &details, ); + self.last_tick.remove(session_id); let _ = self.sessions.remove_session(session_id); result.map_err(|error| error.to_string()) } diff --git a/src/server/games/crash.rs b/src/server/games/crash.rs index ba73c7f..1728afc 100644 --- a/src/server/games/crash.rs +++ b/src/server/games/crash.rs @@ -1,5 +1,8 @@ use super::traits::{Game, GameError, GamePhase, GameResult, GameState}; -use crate::crash::{crash_point_from_uniform, payout, CrashState}; +use crate::crash::{ + CrashState, SEED_LEN, commit_seed, crash_point_from_seed, payout, seed_to_uniform, +}; +use rand::RngCore; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex}; @@ -64,7 +67,14 @@ pub struct CrashRound { pub bet: u64, /// Drawn at start, revealed only when the round ends. pub crash_point: f64, + /// The 32 random bytes the crash point commits to, hex encoded. Kept on + /// the server until the round ends. + pub seed: String, + /// `sha256(seed)` in hex, sent with the start reply. + pub commit: String, pub started_at: f64, + /// Server Unix time at start, milliseconds, for clock sync. + pub started_at_ms: u64, /// `running`, `cashed`, `crashed` or `forfeit`. pub phase: String, pub result: String, @@ -74,11 +84,15 @@ pub struct CrashRound { } impl CrashRound { - pub fn new(bet: u64, crash_point: f64, started_at: f64) -> Self { + pub fn new(bet: u64, seed: [u8; SEED_LEN], started_at: f64, started_at_ms: u64) -> Self { + let crash_point = crash_point_from_seed(&seed); Self { bet, crash_point, + seed: seed_hex(&seed), + commit: commit_seed(&seed), started_at, + started_at_ms, phase: "running".to_string(), result: "playing".to_string(), multiplier: 1.0, @@ -120,6 +134,7 @@ impl CrashRound { balance, multiplier: self.multiplier, potential: payout(self.bet, self.multiplier), + started_at: self.started_at_ms, crash_point: if self.is_running() { None } else { @@ -127,6 +142,12 @@ impl CrashRound { }, payout: self.payout, result: self.result.clone(), + commit: Some(self.commit.clone()), + seed: if self.is_running() { + None + } else { + Some(self.seed.clone()) + }, } } @@ -146,6 +167,34 @@ pub struct CrashGame { uniforms: VecDeque, } +/// Hex encode a round seed. Lowercase, no separators. +fn seed_hex(seed: &[u8]) -> String { + let mut out = String::with_capacity(seed.len() * 2); + for byte in seed { + use std::fmt::Write as _; + let _ = write!(out, "{byte:02x}"); + } + out +} + +/// Draw a seed whose committed point matches a queued test uniform. +/// +/// The first eight bytes are the big-endian integer closest to +/// `uniform * 2^64`; the remaining bytes are zero. This keeps the existing +/// deterministic crash-point tests meaningful while every round still has a +/// real seed. +fn seed_from_uniform(uniform: f64) -> [u8; SEED_LEN] { + let uniform = if uniform.is_finite() { + uniform.clamp(0.0, 0.999_999_999_999_999_9) + } else { + 0.0 + }; + let value = (uniform * 18_446_744_073_709_551_616.0) as u64; + let mut seed = [0u8; SEED_LEN]; + seed[..8].copy_from_slice(&value.to_be_bytes()); + seed +} + impl CrashGame { pub fn new() -> Self { Self { @@ -163,12 +212,25 @@ impl CrashGame { } } + /// Draw the next round seed. Tests supply uniforms; production uses the + /// operating system's random source. + pub fn next_seed(&mut self) -> [u8; SEED_LEN] { + if let Some(uniform) = self.uniforms.pop_front() { + return seed_from_uniform(uniform); + } + let mut seed = [0u8; SEED_LEN]; + rand::thread_rng().fill_bytes(&mut seed); + seed + } + + /// Kept for callers that only need the point a queued uniform encodes. pub fn next_crash_point(&mut self) -> f64 { - let uniform = self - .uniforms - .pop_front() - .unwrap_or_else(rand::random::); - crash_point_from_uniform(uniform) + crash_point_from_seed(&self.next_seed()) + } + + /// The uniform encoded by a seed, exposed for verification tests. + pub fn uniform_of(seed: &[u8; SEED_LEN]) -> f64 { + seed_to_uniform(seed) } } diff --git a/src/version.rs b/src/version.rs index c1dcee5..adc2a8d 100644 --- a/src/version.rs +++ b/src/version.rs @@ -5,6 +5,7 @@ //! accepted until the server has answered it. use std::env; +use std::time::{SystemTime, UNIX_EPOCH}; /// Version this build reports. The client and the server share it. pub const VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -73,9 +74,9 @@ pub fn is_newer(latest: &str, current: &str) -> bool { } } -/// The server line that accepts a handshake. -pub fn hello_line(version: &str, latest: &str, url: &str) -> String { - format!("HELLO {version} latest={latest} url={url}") +/// The server line that accepts a handshake. `time` is Unix milliseconds. +pub fn hello_line(version: &str, latest: &str, url: &str, time: u64) -> String { + format!("HELLO {version} latest={latest} url={url} time={time}") } #[derive(Debug, Clone, PartialEq, Eq)] @@ -83,9 +84,11 @@ pub struct HelloReply { pub version: String, pub latest: String, pub url: String, + /// Server clock at the moment the reply was built, Unix milliseconds. + pub time: u64, } -/// Parse `HELLO latest= url=`. +/// Parse `HELLO latest= url= time=`. pub fn parse_hello(line: &str) -> Option { let rest = line.trim().strip_prefix("HELLO ")?; let mut tokens = rest.split_whitespace(); @@ -93,11 +96,14 @@ pub fn parse_hello(line: &str) -> Option { let mut latest = None; let mut url = None; + let mut time = None; for token in tokens { if let Some(value) = token.strip_prefix("latest=") { latest = Some(value.to_string()); } else if let Some(value) = token.strip_prefix("url=") { url = Some(value.to_string()); + } else if let Some(value) = token.strip_prefix("time=") { + time = value.parse().ok(); } } @@ -105,9 +111,29 @@ pub fn parse_hello(line: &str) -> Option { version, latest: latest?, url: url?, + time: time?, }) } +/// The reply to `PING`: `PONG `. +pub fn pong_line(time: u64) -> String { + format!("PONG {time}") +} + +/// Parse a `PONG` reply's server time. +pub fn parse_pong(line: &str) -> Option { + line.trim().strip_prefix("PONG ")?.trim().parse().ok() +} + +/// Current Unix time in milliseconds. Used for handshake timestamps and crash +/// anchors; a system clock before the epoch reads as zero. +pub fn unix_time_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64) + .unwrap_or(0) +} + /// Download page advertised by the server. `DOWNLOAD_URL` wins when set. pub fn download_url() -> String { env::var("DOWNLOAD_URL") -- cgit v1.2.3