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. --- webd/src/session.rs | 623 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 623 insertions(+) create mode 100644 webd/src/session.rs (limited to 'webd/src/session.rs') diff --git a/webd/src/session.rs b/webd/src/session.rs new file mode 100644 index 0000000..8a20721 --- /dev/null +++ b/webd/src/session.rs @@ -0,0 +1,623 @@ +//! One gateway session: a browser socket, one upstream TLS connection and one +//! `ClientCore`. +//! +//! The browser sends `Command` JSON and renders `View` JSON. All game state, +//! protocol framing and balance arithmetic stay in the Rust core. + +use std::collections::HashMap; +use std::net::TcpStream; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; +use std::thread; +use std::time::{Duration, Instant}; + +use rusqlite::Connection; +use serde::{Deserialize, Serialize}; +use tungstenite::error::Error as WsError; +use tungstenite::http::HeaderValue; +use tungstenite::http::header::ORIGIN; +use tungstenite::protocol::{WebSocket, WebSocketConfig}; +use tungstenite::{Message, accept_hdr_with_config}; + +use zagan_bet::client::core::{ClientCore, Effect}; +use zagan_bet::client::database; +use zagan_bet::client::transport::{self, TlsStream}; +use zagan_bet::client::view::{Command, View}; +use zagan_bet::protocol::Packet; +use zagan_bet::version::VERSION; + +use crate::config::Config; + +const COOKIE_NAME: &str = "zagan_session"; +/// On HTTPS, use the `__Host-` prefix so a sibling origin cannot set a +/// domain-scoped cookie and fixate the session. +const HOST_COOKIE_NAME: &str = "__Host-zagan_session"; +const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const POLL: Duration = Duration::from_millis(10); + +pub enum SocketEvent { + Attach(WebSocket), +} + +/// Session id → generation and the thread that owns it. +/// +/// The generation lets an ending thread remove its own entry without deleting +/// a replacement session that reused the same cookie. +pub type Registry = Arc)>>>; + +static NEXT_GENERATION: AtomicU64 = AtomicU64::new(1); + +pub fn new_registry() -> Registry { + Arc::new(Mutex::new(HashMap::new())) +} + +/// Complete the WebSocket handshake, then attach this socket to an existing +/// session or start one. +pub fn accept(stream: TcpStream, config: Config, registry: Registry) -> Result<(), String> { + let mut session_id = None; + let origin = config.origin.clone(); + let secure = config.cookie_secure; + + let ws_config = WebSocketConfig { + // A command or view is a few KiB at most; keep a compromised page from + // making the gateway buffer tens of megabytes. + max_message_size: Some(64 * 1024), + max_frame_size: Some(64 * 1024), + ..WebSocketConfig::default() + }; + let mut ws = accept_hdr_with_config( + stream, + |request: &tungstenite::handshake::server::Request, + mut response: tungstenite::handshake::server::Response| { + if let Some(header) = request.headers().get(ORIGIN) { + if header.to_str().ok() != Some(origin.as_str()) { + return Err(tungstenite::http::Response::builder() + .status(403) + .body(Some("origin not allowed".to_string())) + .expect("static response")); + } + } + + let found = request + .headers() + .get("cookie") + .and_then(|value| value.to_str().ok()) + .and_then(cookie_id); + let (id, fresh) = match found { + Some(id) => (id, false), + None => (random_id(), true), + }; + if fresh { + let mut cookie = format!("{COOKIE_NAME}={id}; Path=/; HttpOnly; SameSite=Strict"); + if secure { + cookie.push_str("; Secure"); + } + if let Ok(value) = HeaderValue::from_str(&cookie) { + response.headers_mut().append("set-cookie", value); + } + } + session_id = Some(id); + Ok(response) + }, + Some(ws_config), + ) + .map_err(|error| format!("WebSocket handshake failed: {error}"))?; + + let id = session_id.ok_or_else(|| "handshake produced no session id".to_string())?; + ws.get_ref() + .set_nonblocking(true) + .map_err(|error| format!("could not make the browser socket nonblocking: {error}"))?; + + let (sender, receiver) = mpsc::channel(); + let generation = NEXT_GENERATION.fetch_add(1, Ordering::Relaxed); + { + let mut sessions = registry.lock().map_err(|_| "session registry poisoned")?; + if let Some((_, existing)) = sessions.get(&id).cloned() { + match existing.send(SocketEvent::Attach(ws)) { + Ok(()) => return Ok(()), + Err(mpsc::SendError(SocketEvent::Attach(returned))) => { + sessions.remove(&id); + ws = returned; + } + } + } + sessions.insert(id.clone(), (generation, sender.clone())); + } + + thread::spawn(move || { + let session = match Session::new(id.clone(), config) { + Ok(session) => Some(session), + Err(error) => { + eprintln!("[webd {id}] could not start: {error}"); + None + } + }; + if let Some(mut session) = session { + session.attach(ws); + session.poll_loop(&receiver); + } + if let Ok(mut sessions) = registry.lock() { + let own = sessions + .get(&id) + .map(|(generation_now, _)| *generation_now == generation) + .unwrap_or(false); + if own { + sessions.remove(&id); + } + } + eprintln!("[webd {id}] session ended"); + }); + + Ok(()) +} + +fn cookie_id(raw: &str) -> Option { + for pair in raw.split(';') { + let (name, value) = pair.split_once('=')?; + if matches!(name.trim(), COOKIE_NAME | HOST_COOKIE_NAME) { + let value = value.trim(); + if value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Some(value.to_ascii_lowercase()); + } + return None; + } + } + None +} + +fn random_id() -> String { + use rand::RngCore; + let mut bytes = [0u8; 32]; + rand::thread_rng().fill_bytes(&mut bytes); + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +#[derive(Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum ClientMessage { + Command { command: Command }, + Ping, +} + +#[derive(Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum ServerMessage<'a> { + Hello { + version: &'a str, + grace_seconds: u64, + /// Signed-in user, so a refresh during a game can skip the login card. + user: Option<&'a str>, + /// Estimated game-server Unix time, milliseconds. + time: u64, + }, + View { + view: &'a View, + }, + Notice { + text: &'a str, + }, + Error { + text: &'a str, + }, + Replaced, + Pong, +} + +struct Session { + id: String, + config: Config, + core: ClientCore, + /// In-memory trust store seeded with the pinned fingerprint. No player + /// credentials are ever stored here. + trust: Connection, + upstream: Option, + ws: Option>, + pending_id: Option, + pending_since: Option, + grace_until: Option, + drop_upstream: bool, +} + +impl Session { + fn new(id: String, config: Config) -> Result { + let trust = Connection::open_in_memory() + .map_err(|error| format!("could not open the trust store: {error}"))?; + database::init_db(&trust); + database::save_fingerprint(&trust, &config.upstream, &config.fingerprint) + .map_err(|error| format!("could not pin the upstream key: {error}"))?; + + Ok(Self { + id, + config, + core: ClientCore::new(), + trust, + upstream: None, + ws: None, + pending_id: None, + pending_since: None, + grace_until: None, + drop_upstream: false, + }) + } + + fn attach(&mut self, ws: WebSocket) { + if let Some(mut old) = self.ws.take() { + send_json(&mut old, &ServerMessage::Replaced); + close_socket(&mut old); + } + eprintln!("[webd {}] browser attached", self.id); + self.ws = Some(ws); + self.grace_until = None; + let mut ws = self.ws.take().expect("just attached"); + if let Err(text) = self.ensure_upstream() { + send_json(&mut ws, &ServerMessage::Error { text: &text }); + } + send_json( + &mut ws, + &ServerMessage::Hello { + version: VERSION, + grace_seconds: self.config.grace.as_secs(), + user: self.core.user(), + time: self.core.server_now_ms(), + }, + ); + self.send_view(&mut ws); + self.ws = Some(ws); + } + + fn poll_loop(&mut self, receiver: &mpsc::Receiver) { + loop { + match receiver.recv_timeout(POLL) { + Ok(SocketEvent::Attach(ws)) => self.attach(ws), + Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => break, + } + + self.pump_browser(); + self.pump_upstream(); + self.check_request_timeout(); + + if self.ws.is_some() { + self.grace_until = None; + } else if let Some(deadline) = self.grace_until { + if Instant::now() >= deadline { + break; + } + } else { + self.grace_until = Some(Instant::now() + self.config.grace); + } + } + + // Dropping the TLS stream closes upstream; the server forfeits any + // live hand, exactly like a desktop client disconnecting. + self.upstream = None; + eprintln!("[webd {}] grace expired", self.id); + } + + fn pump_browser(&mut self) { + let Some(mut ws) = self.ws.take() else { + return; + }; + let mut keep = true; + + loop { + match ws.read() { + Ok(Message::Text(text)) => { + if !self.handle_browser_text(&text, &mut ws) { + keep = false; + break; + } + } + Ok(Message::Close(_)) => { + keep = false; + break; + } + Ok(Message::Ping(payload)) => { + let _ = send_message(&mut ws, Message::Pong(payload)); + } + Ok(Message::Binary(_)) | Ok(Message::Pong(_)) | Ok(Message::Frame(_)) => {} + Err(WsError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock => break, + Err(WsError::ConnectionClosed) | Err(WsError::AlreadyClosed) => { + keep = false; + break; + } + Err(_) => { + keep = false; + break; + } + } + } + + if keep { + self.ws = Some(ws); + } else { + self.grace_until = Some(Instant::now() + self.config.grace); + self.pending_id = None; + self.pending_since = None; + } + } + + fn handle_browser_text(&mut self, text: &str, ws: &mut WebSocket) -> bool { + let message = match serde_json::from_str::(text) { + Ok(message) => message, + Err(error) => { + send_json( + ws, + &ServerMessage::Error { + text: &format!("Malformed message: {error}"), + }, + ); + return true; + } + }; + + match message { + ClientMessage::Ping => send_json(ws, &ServerMessage::Pong), + ClientMessage::Command { command } => { + if let Err(text) = self.ensure_upstream() { + send_json(ws, &ServerMessage::Error { text: &text }); + self.send_view(ws); + return true; + } + let effects = self.core.command(command); + self.apply_effects(effects, Some(ws)); + } + } + true + } + + fn ensure_upstream(&mut self) -> Result<(), String> { + if self.upstream.is_some() { + return Ok(()); + } + + let mut trust = |_prompt: &transport::TrustPrompt| false; + match transport::connect(&self.config.upstream, &self.trust, false, &mut trust) { + Ok(connected) => { + self.core.set_connected(true); + self.core.set_server_time(connected.server_time); + self.core.set_host_info( + Some(self.config.upstream.clone()), + Some(self.config.fingerprint.clone()), + ); + self.upstream = Some(connected.stream); + eprintln!( + "[webd {}] upstream connected: {} ({}), ping {:.1}ms", + self.id, + self.config.upstream, + connected.server_version, + connected.elapsed.as_secs_f64() * 1000.0 + ); + Ok(()) + } + Err(error) => { + self.core.reset_after_disconnect(); + Err(format!("Upstream connection failed: {error}")) + } + } + } + + fn pump_upstream(&mut self) { + let Some(mut upstream) = self.upstream.take() else { + return; + }; + let mut ws = self.ws.take(); + let mut disconnected = None; + + loop { + match upstream.read_available_packet() { + Ok(Some(Packet::Event { body, .. })) => { + if let Some(effects) = self.core.handle_async_line(&body) { + self.apply_effects(effects, ws.as_mut()); + } + } + Ok(Some(Packet::Reply { id, body })) => { + if self.pending_id == Some(id) { + self.pending_id = None; + self.pending_since = None; + } + let effects = self.core.handle_reply_id(id, &body); + self.apply_effects(effects, ws.as_mut()); + } + Ok(None) => break, + Err(error) => { + disconnected = Some(error.to_string()); + break; + } + } + } + + if self.drop_upstream { + self.drop_upstream = false; + drop(upstream); + } else if disconnected.is_none() { + self.upstream = Some(upstream); + } + + if let Some(message) = disconnected { + eprintln!("[webd {}] upstream lost: {message}", self.id); + self.core.reset_after_disconnect(); + self.pending_id = None; + self.pending_since = None; + if let Some(ws) = ws.as_mut() { + send_json( + ws, + &ServerMessage::Error { + text: "Connection to the game server was lost", + }, + ); + self.send_view(ws); + } + } + + self.ws = ws; + } + + fn check_request_timeout(&mut self) { + let Some(sent) = self.pending_since else { + return; + }; + if sent.elapsed() < REQUEST_TIMEOUT { + return; + } + + let id = self.pending_id.take(); + self.pending_since = None; + self.upstream = None; + let mut ws = self.ws.take(); + if let Some(id) = id { + let effects = self.core.handle_reply_id(id, "ERROR Request timed out"); + self.apply_effects(effects, ws.as_mut()); + if let Some(ws) = ws.as_mut() { + send_json( + ws, + &ServerMessage::Notice { + text: "Request timed out. Disconnected.", + }, + ); + } + } + self.core.reset_after_disconnect(); + if let Some(ws) = ws.as_mut() { + self.send_view(ws); + } + self.ws = ws; + } + + fn apply_effects(&mut self, effects: Vec, mut ws: Option<&mut WebSocket>) { + for effect in effects { + match effect { + Effect::Print(lines) => { + // A multi-line print is a terminal table; the structured + // `View` carries that data to the browser. + if lines.len() == 1 { + if let (Some(socket), Some(line)) = (ws.as_deref_mut(), lines.first()) { + send_json(socket, &ServerMessage::Notice { text: line }); + } + } + } + Effect::SendRequest { id, payload } => { + self.pending_id = Some(id); + self.pending_since = Some(Instant::now()); + match self.ensure_upstream() { + Ok(()) => { + if let Some(upstream) = self.upstream.as_mut() { + if let Err(error) = upstream.write_request(id, None, &payload) { + let text = format!("Could not send request: {error}"); + if let Some(socket) = ws.as_deref_mut() { + send_json(socket, &ServerMessage::Error { text: &text }); + } + self.drop_upstream = true; + self.upstream = None; + } + } + } + Err(text) => { + self.pending_id = None; + self.pending_since = None; + if let Some(socket) = ws.as_deref_mut() { + send_json(socket, &ServerMessage::Error { text: &text }); + } + } + } + } + Effect::Disconnect => { + self.upstream = None; + self.drop_upstream = true; + self.core.reset_after_disconnect(); + } + Effect::Render(_) | Effect::ClearRegion | Effect::Quit => {} + } + } + + if let Some(socket) = ws { + self.send_view(socket); + } + } + + fn send_view(&self, ws: &mut WebSocket) { + let Some(view) = self.core.view() else { + return; + }; + send_json(ws, &ServerMessage::View { view: &view }); + } +} + +fn send_message(ws: &mut WebSocket, message: Message) -> Result<(), WsError> { + // The read side is nonblocking; writes are small and should not fail + // half-way because a browser is briefly not draining its socket. + let _ = ws.get_ref().set_nonblocking(false); + let result = ws.send(message); + let _ = ws.get_ref().set_nonblocking(true); + result +} + +fn send_json(ws: &mut WebSocket, message: &ServerMessage<'_>) { + let Ok(text) = serde_json::to_string(message) else { + return; + }; + let _ = send_message(ws, Message::Text(text)); +} + +fn close_socket(ws: &mut WebSocket) { + let _ = ws.get_ref().set_nonblocking(false); + let _ = ws.close(None); + let _ = ws.get_ref().set_nonblocking(true); +} + +#[cfg(test)] +mod tests { + use super::*; + use zagan_bet::client::view::{LobbyView, View}; + + #[test] + fn reads_only_well_formed_session_cookies() { + let id = "a".repeat(64); + assert_eq!( + cookie_id(&format!("other=1; {COOKIE_NAME}={id}; theme=dark")), + Some(id) + ); + assert_eq!(cookie_id(&format!("{COOKIE_NAME}=short")), None); + assert_eq!(cookie_id("other=1"), None); + assert_eq!(cookie_id(""), None); + } + + #[test] + fn parses_a_browser_command_and_rejects_nonsense() { + let message: ClientMessage = serde_json::from_str( + r#"{"type":"command","command":{"Login":{"user":"alice","credential":"abc"}}}"#, + ) + .expect("login command"); + match message { + ClientMessage::Command { command } => { + assert_eq!( + command, + Command::Login { + user: "alice".to_string(), + credential: "abc".to_string() + } + ); + } + _ => panic!("expected a command"), + } + + assert!(serde_json::from_str::("{}").is_err()); + assert!(serde_json::from_str::(r#"{"type":"nope"}"#).is_err()); + } + + #[test] + fn serializes_a_view_for_the_browser() { + let view = View::Lobby(LobbyView { + connected: true, + user: Some("alice".to_string()), + account: Some("1".to_string()), + balance: Some(1000), + games: vec![("slots".to_string(), "Slots".to_string())], + login_error: None, + }); + let json = serde_json::to_string(&ServerMessage::View { view: &view }).unwrap(); + assert!(json.contains(r#""type":"view""#)); + assert!(json.contains(r#""screen":"lobby""#)); + assert!(json.contains(r#""user":"alice""#)); + } +} -- cgit v1.2.3