1 files changed, 623 insertions, 0 deletions

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<TcpStream>),
+}
+
+/// 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<Mutex<HashMap<String, (u64, mpsc::Sender<SocketEvent>)>>>;
+
+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<String> {
+    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<TlsStream>,
+    ws: Option<WebSocket<TcpStream>>,
+    pending_id: Option<u64>,
+    pending_since: Option<Instant>,
+    grace_until: Option<Instant>,
+    drop_upstream: bool,
+}
+
+impl Session {
+    fn new(id: String, config: Config) -> Result<Self, String> {
+        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<TcpStream>) {
+        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<SocketEvent>) {
+        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<TcpStream>) -> bool {
+        let message = match serde_json::from_str::<ClientMessage>(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<Effect>, mut ws: Option<&mut WebSocket<TcpStream>>) {
+        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<TcpStream>) {
+        let Some(view) = self.core.view() else {
+            return;
+        };
+        send_json(ws, &ServerMessage::View { view: &view });
+    }
+}
+
+fn send_message(ws: &mut WebSocket<TcpStream>, 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<TcpStream>, 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<TcpStream>) {
+    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::<ClientMessage>("{}").is_err());
+        assert!(serde_json::from_str::<ClientMessage>(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""#));
+    }
+}