diff options
Diffstat (limited to 'crates/chat-server/src/app')
| -rw-r--r-- | crates/chat-server/src/app/cli.rs | 167 | ||||
| -rw-r--r-- | crates/chat-server/src/app/http.rs | 243 | ||||
| -rw-r--r-- | crates/chat-server/src/app/maintenance.rs | 72 | ||||
| -rw-r--r-- | crates/chat-server/src/app/service.rs | 438 | ||||
| -rw-r--r-- | crates/chat-server/src/app/store.rs | 182 |
5 files changed, 1102 insertions, 0 deletions
diff --git a/crates/chat-server/src/app/cli.rs b/crates/chat-server/src/app/cli.rs new file mode 100644 index 0000000..1296df4 --- /dev/null +++ b/crates/chat-server/src/app/cli.rs @@ -0,0 +1,167 @@ +// Copyright (C) 2026 Doan Luu Duc Tai <doan@luuduct.ai> +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <https://www.gnu.org/licenses/>. + +use clap::{Arg, ArgAction, Command}; +use std::ffi::OsString; +use std::net::SocketAddr; +use std::path::PathBuf; + +#[derive(Debug)] + +pub(super) struct Arguments { + pub(super) root: PathBuf, + + pub(super) listener: Listener, + + pub(super) healthcheck: bool, + + pub(super) init: bool, + + pub(super) reset_owner: Option<String>, + + pub(super) owner_fingerprint: bool, + + pub(super) snapshot: Option<PathBuf>, +} + +#[derive(Debug)] +pub(super) struct Listener { + pub(super) bind: Option<SocketAddr>, + + pub(super) proxy: bool, +} + +fn flag(name: &'static str) -> Arg { + Arg::new(name).long(name).action(ArgAction::SetTrue) +} + +fn path(name: &'static str) -> Arg { + Arg::new(name) + .long(name) + .value_parser(clap::builder::PathBufValueParser::new()) +} + +fn command() -> Command { + Command::new("stack-chat-server").about("Run private chat directly or behind the stack gateway") + .args([ + path("root").default_value("."), + Arg::new("bind").long("bind").value_parser(str::parse::<SocketAddr>), + flag("proxy").help("Run behind the stack gateway using its explicit container environment"), + flag("healthcheck").help("Check the local HTTP health endpoint without opening the database"), + flag("init").conflicts_with_all(["reset-owner", "owner-fingerprint", "snapshot"]), + Arg::new("reset-owner").long("reset-owner").value_name("FINGERPRINT").conflicts_with_all(["owner-fingerprint", "snapshot"]) + .help("Permanently clear the matching owner registration and all chat history after key loss"), + flag("owner-fingerprint").conflicts_with("snapshot").help("Print the registered owner's public fingerprint"), + path("snapshot"), + ]) +} + +impl Arguments { + pub(super) fn try_parse() -> Result<Self, clap::Error> { + Self::try_parse_from(std::env::args_os()) + } + + fn try_parse_from( + arguments: impl IntoIterator<Item = impl Into<OsString> + Clone>, + ) -> Result<Self, clap::Error> { + let mut values = command().try_get_matches_from(arguments)?; + let root = values.remove_one("root").ok_or_else(|| { + clap::Error::raw( + clap::error::ErrorKind::MissingRequiredArgument, + "Missing application root", + ) + })?; + Ok(Self { + root, + listener: Listener { + bind: values.remove_one("bind"), + proxy: values.get_flag("proxy"), + }, + healthcheck: values.get_flag("healthcheck"), + init: values.get_flag("init"), + reset_owner: values.remove_one("reset-owner"), + owner_fingerprint: values.get_flag("owner-fingerprint"), + snapshot: values.remove_one("snapshot"), + }) + } +} + +#[cfg(test)] +mod tests { + use super::{Arguments, command}; + use anyhow::{Result, ensure}; + use std::path::Path; + + #[test] + fn checked_command_schema() { + command().debug_assert(); + } + + #[test] + fn listener_defaults_and_maintenance_conflicts() -> Result<()> { + let defaults = Arguments::try_parse_from(["stack-chat-server"])?; + anyhow::ensure!((defaults.root) == (Path::new(".")), "Test invariant failed"); + anyhow::ensure!( + !defaults.listener.proxy && defaults.listener.bind.is_none(), + "Test invariant failed" + ); + let parsed = Arguments::try_parse_from([ + "stack-chat-server", + "--root", + "/tmp/stack", + "--bind", + "127.0.0.1:9180", + "--proxy", + "--healthcheck", + ])?; + anyhow::ensure!( + (parsed.root) == (Path::new("/tmp/stack")), + "Test invariant failed" + ); + ensure!( + parsed + .listener + .bind + .is_some_and(|address| address.port() == 9180), + "Listener port differs" + ); + anyhow::ensure!( + parsed.listener.proxy && parsed.healthcheck, + "Test invariant failed" + ); + for args in [ + vec!["stack-chat-server", "--bind", "garbage"], + vec!["stack-chat-server", "--init", "--snapshot", "snapshot"], + vec![ + "stack-chat-server", + "--reset-owner", + "abc", + "--owner-fingerprint", + ], + vec![ + "stack-chat-server", + "--owner-fingerprint", + "--snapshot", + "snapshot", + ], + ] { + anyhow::ensure!( + Arguments::try_parse_from(args).is_err(), + "Test invariant failed" + ); + } + Ok(()) + } +} diff --git a/crates/chat-server/src/app/http.rs b/crates/chat-server/src/app/http.rs new file mode 100644 index 0000000..5b1941a --- /dev/null +++ b/crates/chat-server/src/app/http.rs @@ -0,0 +1,243 @@ +// Copyright (C) 2026 Doan Luu Duc Tai <doan@luuduct.ai> +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <https://www.gnu.org/licenses/>. + +use crate::app::service::Service; +use anyhow::Result; +use axum::{ + Json, Router, + body::Body, + extract::{DefaultBodyLimit, State}, + http::{HeaderMap, HeaderValue, StatusCode}, + response::{IntoResponse, Redirect, Response}, + routing::{get, post}, +}; +use serde_json::{Value, json}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use tokio::sync::Semaphore; + +#[derive(Clone)] +struct App { + service: Arc<Mutex<Service>>, + permits: Arc<Semaphore>, + root: PathBuf, + origin: String, + public: std::collections::BTreeMap<String, String>, +} + +fn secured(mut response: Response) -> Response { + for (name, value) in [ + ("cache-control", "no-store, max-age=0"), + ( + "content-security-policy", + "default-src 'none'; script-src 'self' 'wasm-unsafe-eval'; worker-src 'self'; connect-src 'self'; style-src 'self'; img-src 'none'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'; object-src 'none'", + ), + ("referrer-policy", "no-referrer"), + ("x-content-type-options", "nosniff"), + ("cross-origin-opener-policy", "same-origin"), + ("cross-origin-resource-policy", "same-origin"), + ( + "permissions-policy", + "camera=(), microphone=(), geolocation=(), payment=(), usb=()", + ), + ] { + response + .headers_mut() + .insert(name, HeaderValue::from_static(value)); + } + response +} + +async fn api(State(app): State<App>, headers: HeaderMap, Json(request): Json<Value>) -> Response { + if headers.get("origin").and_then(|value| value.to_str().ok()) != Some(app.origin.as_str()) { + return secured( + ( + StatusCode::FORBIDDEN, + Json(json!({"error":"Origin rejected"})), + ) + .into_response(), + ); + } + let Ok(permit) = Arc::clone(&app.permits).try_acquire_owned() else { + return secured( + ( + StatusCode::TOO_MANY_REQUESTS, + Json(json!({"error":"Server busy"})), + ) + .into_response(), + ); + }; + let result = tokio::task::spawn_blocking(move || { + let _permit = permit; + app.service + .lock() + .map_err(|_| anyhow::anyhow!("Service unavailable"))? + .call(&request) + }) + .await; + match result { + Ok(Ok(value)) => secured(Json(value).into_response()), + _ => secured((StatusCode::BAD_REQUEST,Json(json!({"error":"Request rejected. Unlock again if your session expired; check credentials, message limits, and conversation access."}))).into_response()), + } +} + +fn packaged_asset<'a>(root: &Path, packaged: &'a str, development: &'a str) -> &'a str { + if root.join(packaged).is_file() { + packaged + } else { + development + } +} + +fn asset_location(root: &Path, path: &str) -> Option<(&'static str, &'static str)> { + let assets = [ + ( + "/", + "bin/chat/web/index.html", + "crates/chat-server/web/index.html", + "text/html; charset=utf-8", + ), + ( + "/app.js", + "bin/chat/web/app.js", + "var/build/typescript/crates/chat-server/web/app.js", + "text/javascript; charset=utf-8", + ), + ( + "/worker.js", + "bin/chat/web/worker.js", + "var/build/typescript/crates/chat-server/web/worker.js", + "text/javascript; charset=utf-8", + ), + ( + "/wasi.mjs", + "bin/chat/wasi.mjs", + "var/build/typescript/crates/chat-crypto/browser/wasi.js", + "text/javascript; charset=utf-8", + ), + ( + "/style.css", + "bin/chat/web/style.css", + "crates/chat-server/web/style.css", + "text/css; charset=utf-8", + ), + ( + "/crypto.wasm", + "bin/chat/crypto.wasm", + "target/wasm32-wasip1/release/stack-chat-crypto.wasm", + "application/wasm", + ), + ( + "/javascript.html", + "bin/chat/web/javascript.html", + "crates/chat-server/web/javascript.html", + "text/html; charset=utf-8", + ), + ( + "/SOURCE.tar.gz", + "bin/chat/SOURCE.tar.gz", + "var/build/chat-source.tar.gz", + "application/gzip", + ), + ( + "/LIMITATIONS.txt", + "docs/CHAT-APPLICATION.txt", + "docs/CHAT-APPLICATION.txt", + "text/plain; charset=utf-8", + ), + ( + "/LICENSE", + "LICENSE", + "LICENSE", + "text/plain; charset=utf-8", + ), + ]; + assets + .into_iter() + .find(|(route, _, _, _)| *route == path) + .map(|(_, packaged, development, mime)| (packaged_asset(root, packaged, development), mime)) +} + +async fn asset(State(app): State<App>, uri: axum::http::Uri) -> Response { + if uri.path() == "/THIRD-PARTY-LICENSES.html" { + let Some(origin) = app.public.get("SITE_ORIGIN") else { + return secured(StatusCode::INTERNAL_SERVER_ERROR.into_response()); + }; + return secured( + Redirect::permanent(&format!("{origin}/THIRD-PARTY-LICENSES.html")).into_response(), + ); + } + let Some((path, mime)) = asset_location(&app.root, uri.path()) else { + return secured(StatusCode::NOT_FOUND.into_response()); + }; + if matches!(uri.path(), "/" | "/javascript.html") { + let Ok(bytes) = tokio::fs::read(app.root.join(path)).await else { + return secured(StatusCode::NOT_FOUND.into_response()); + }; + let rendered = String::from_utf8(bytes) + .map_err(anyhow::Error::from) + .and_then(|source| stack_config::render(&source, &app.public)); + return rendered.map_or_else( + |_| secured(StatusCode::INTERNAL_SERVER_ERROR.into_response()), + |html| secured(([("content-type", mime)], html).into_response()), + ); + } + let Ok(file) = tokio::fs::File::open(app.root.join(path)).await else { + return secured(StatusCode::NOT_FOUND.into_response()); + }; + let length = file.metadata().await.ok().map(|metadata| metadata.len()); + let mut response = Response::new(Body::from_stream(tokio_util::io::ReaderStream::new(file))); + response + .headers_mut() + .insert("content-type", HeaderValue::from_static(mime)); + if let Some(header_length) = + length.and_then(|value| HeaderValue::from_str(&value.to_string()).ok()) + { + response + .headers_mut() + .insert("content-length", header_length); + } + secured(response) +} + +pub(super) async fn serve( + listener: tokio::net::TcpListener, + service: Service, + root: &Path, + public: std::collections::BTreeMap<String, String>, +) -> Result<()> { + let app = App { + origin: service.origin.clone(), + service: Arc::new(Mutex::new(service)), + permits: Arc::new(Semaphore::new(4)), + root: root.to_path_buf(), + public, + }; + let routes = Router::new() + .route( + "/healthz", + get(|| async { secured("ok\n".into_response()) }), + ) + .route("/api", post(api)) + .fallback(get(asset)) + .layer(DefaultBodyLimit::max(262_144)) + .with_state(app); + axum::serve(listener, routes) + .with_graceful_shutdown(async { + drop(tokio::signal::ctrl_c().await); + }) + .await?; + Ok(()) +} diff --git a/crates/chat-server/src/app/maintenance.rs b/crates/chat-server/src/app/maintenance.rs new file mode 100644 index 0000000..647dff9 --- /dev/null +++ b/crates/chat-server/src/app/maintenance.rs @@ -0,0 +1,72 @@ +// Copyright (C) 2026 Doan Luu Duc Tai <doan@luuduct.ai> +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <https://www.gnu.org/licenses/>. + +use crate::app::store::Store; +use anyhow::{Context, Result, ensure}; +use std::fs::{File, OpenOptions}; +use std::io::Write; +use std::net::{SocketAddr, TcpListener}; +use std::os::unix::fs::OpenOptionsExt; +use std::path::Path; + +pub(super) fn lock(directory: &Path) -> Result<File> { + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .mode(0o600) + .open(directory.join("server.lock"))?; + file.try_lock().context( + "Chat is already running. Stop it before starting another server or resetting the owner.", + )?; + Ok(file) +} + +pub(super) fn fingerprint(store: &Store) -> Result<()> { + let certificate = store.owner()?.context("No owner identity is registered")?; + writeln!( + std::io::stdout().lock(), + "{}", + stack_chat_crypto::public_identity(&certificate)? + )?; + Ok(()) +} + +pub(super) fn reset( + store: &mut Store, + expected: &str, + address: SocketAddr, + proxy: bool, +) -> Result<()> { + let _port = if proxy { + None + } else { + ensure!( + address.ip().is_loopback(), + "Reset must use the local chat address" + ); + Some( + TcpListener::bind(address) + .context("Stop the chat server before resetting its owner")?, + ) + }; + store.reset_owner(expected)?; + writeln!( + std::io::stdout().lock(), + "Owner registration and chat history cleared. Restart chat and use OWNER SETUP to create a new owner file. Existing backup archives were not changed." + )?; + Ok(()) +} diff --git a/crates/chat-server/src/app/service.rs b/crates/chat-server/src/app/service.rs new file mode 100644 index 0000000..d5433e7 --- /dev/null +++ b/crates/chat-server/src/app/service.rs @@ -0,0 +1,438 @@ +// Copyright (C) 2026 Doan Luu Duc Tai <doan@luuduct.ai> +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <https://www.gnu.org/licenses/>. + +use crate::app::store::Store; +use anyhow::{Context, Result, ensure}; +use base64::{Engine, prelude::BASE64_STANDARD}; +use serde_json::{Value, json}; +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +struct Challenge { + certificate: String, + text: String, + expiry: Instant, + setup: bool, +} + +struct Session { + identity: String, + expiry: Instant, +} + +pub(super) struct Service { + pub(super) store: Store, + pub(super) origin: String, + setup_token: String, + challenges: HashMap<String, Challenge>, + sessions: HashMap<String, Session>, + window: Instant, + attempts: u16, +} + +pub(super) fn token() -> Result<String> { + let mut bytes = [0; 32]; + openssl::rand::rand_bytes(&mut bytes)?; + Ok(hex::encode(bytes)) +} + +pub(super) fn field<'a>(request: &'a Value, key: &str) -> Result<&'a str> { + request + .get(key) + .and_then(Value::as_str) + .context("Missing string field") +} + +fn ciphertext(request: &Value) -> Result<&str> { + let encoded = field(request, "ciphertext")?; + ensure!( + encoded.len() <= 65_536 && BASE64_STANDARD.decode(encoded)?.len() >= 64, + "Invalid ciphertext length" + ); + Ok(encoded) +} + +fn identifier(value: &str) -> Result<()> { + ensure!( + value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()), + "Invalid identifier" + ); + Ok(()) +} + +impl Service { + pub(super) fn new(store: Store, origin: String, setup_token: String) -> Self { + Self { + store, + origin, + setup_token, + challenges: HashMap::new(), + sessions: HashMap::new(), + window: Instant::now(), + attempts: 0, + } + } + + fn owner_id(&self) -> Result<String> { + stack_chat_crypto::public_identity(&self.store.owner()?.context("Owner not initialized")?) + } + + fn challenge(&mut self, request: &Value) -> Result<Value> { + if self.window.elapsed() >= Duration::from_secs(60) { + self.window = Instant::now(); + self.attempts = 0; + } + ensure!(self.attempts < 60, "Authentication rate exceeded"); + self.attempts = self + .attempts + .checked_add(1) + .context("Rate counter overflow")?; + self.challenges + .retain(|_, challenge| challenge.expiry > Instant::now()); + ensure!(self.challenges.len() < 128, "Too many pending challenges"); + let certificate = field(request, "public")?; + let fingerprint = stack_chat_crypto::public_identity(certificate)?; + let setup = self.store.owner()?.is_none(); + if setup { + let supplied = field(request, "setup_token")?; + ensure!( + self.setup_token.len() == 64 + && supplied.len() == 64 + && openssl::memcmp::eq(supplied.as_bytes(), self.setup_token.as_bytes()), + "Invalid setup credential" + ); + } + let nonce = token()?; + let expires = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_secs() + .checked_add(60) + .context("Clock overflow")?; + let owner = if setup { + fingerprint.clone() + } else { + self.owner_id()? + }; + let text = format!( + "stack-chat-auth/v1\n{}\n{fingerprint}\n{nonce}\n{expires}\n{owner}", + self.origin + ); + self.challenges.insert( + nonce.clone(), + Challenge { + certificate: certificate.into(), + text: text.clone(), + expiry: Instant::now() + .checked_add(Duration::from_secs(60)) + .context("Clock overflow")?, + setup, + }, + ); + Ok(json!({"id":nonce,"challenge":text})) + } + + fn authenticate(&mut self, request: &Value) -> Result<Value> { + let challenge = self + .challenges + .remove(field(request, "id")?) + .context("Missing or consumed challenge")?; + ensure!(challenge.expiry > Instant::now(), "Expired challenge"); + stack_chat_crypto::verify_proof( + &challenge.certificate, + &challenge.text, + field(request, "signature")?, + )?; + let identity = stack_chat_crypto::public_identity(&challenge.certificate)?; + if challenge.setup { + self.store.install_owner(&challenge.certificate)?; + self.setup_token.clear(); + } + let owner = identity == self.owner_id()?; + if !owner { + self.store.register(&identity, &challenge.certificate)?; + } + self.sessions + .retain(|_, session| session.expiry > Instant::now()); + ensure!(self.sessions.len() < 256, "Too many sessions"); + let credential = token()?; + self.sessions.insert( + credential.clone(), + Session { + identity: identity.clone(), + expiry: Instant::now() + .checked_add(Duration::from_mins(30)) + .context("Clock overflow")?, + }, + ); + Ok(json!({"token":credential,"owner":owner,"id":identity,"expires_seconds":1800})) + } + + fn principal(&mut self, request: &Value) -> Result<String> { + self.sessions + .retain(|_, session| session.expiry > Instant::now()); + Ok(self + .sessions + .get(field(request, "token")?) + .context("Missing or expired session")? + .identity + .clone()) + } + + fn authorized(&self, request: &Value, identity: &str) -> Result<String> { + let conversation = field(request, "conversation")?; + identifier(conversation)?; + ensure!( + identity == conversation || identity == self.owner_id()?, + "Forbidden conversation" + ); + self.store.conversation(conversation)?; + Ok(conversation.into()) + } + + pub(super) fn call(&mut self, request: &Value) -> Result<Value> { + match field(request, "op")? { + "config" => Ok( + json!({"owner":self.store.owner()?,"origin":self.origin,"protocol":"stack-chat/v1"}), + ), + "challenge" => self.challenge(request), + "authenticate" => self.authenticate(request), + _ => self.authenticated(request), + } + } + + fn authenticated(&mut self, request: &Value) -> Result<Value> { + let identity = self.principal(request)?; + let operation = field(request, "op")?; + if operation == "lock" { + self.sessions.remove(field(request, "token")?); + return Ok(json!({"locked":true})); + } + if operation == "inbox" { + ensure!(identity == self.owner_id()?, "Owner only"); + return self.store.inbox(); + } + let conversation = self.authorized(request, &identity)?; + match operation { + "history" => self.store.messages( + &conversation, + request + .get("count") + .and_then(Value::as_u64) + .context("Missing count")?, + request.get("before").and_then(Value::as_u64).unwrap_or(0), + ), + "profile" => { + ensure!(identity == conversation, "Visitor only"); + self.store.profile(&conversation, ciphertext(request)?)?; + Ok(json!({"stored":true})) + } + "send" => { + let id = field(request, "id")?; + identifier(id)?; + self.store + .send(&conversation, id, &identity, ciphertext(request)?) + } + _ => anyhow::bail!("Unknown operation"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + struct Identity { + file: String, + public: String, + fingerprint: String, + } + + fn crypto(request: &Value) -> Result<Value> { + let response = + stack_chat_crypto::dispatch(&request.to_string()).map_err(anyhow::Error::msg)?; + Ok(serde_json::from_str(&response)?) + } + + fn identity(name: &str, owner: &str) -> Result<Identity> { + let created = crypto( + &json!({"op":"create","origin":"http://127.0.0.1:9080","name":name,"password":"test-only-strong-password","owner":owner}), + )?; + let file = field(&created, "file")?; + let opened = crypto( + &json!({"op":"open","origin":"http://127.0.0.1:9080","file":file,"password":"test-only-strong-password"}), + )?; + Ok(Identity { + file: file.into(), + public: field(&opened, "public")?.into(), + fingerprint: field(&opened, "fingerprint")?.into(), + }) + } + + fn authenticate(service: &mut Service, identity: &Identity, setup: &str) -> Result<Value> { + let challenge = service + .call(&json!({"op":"challenge","public":identity.public,"setup_token":setup}))?; + let proof = crypto( + &json!({"op":"proof","file":identity.file,"password":"test-only-strong-password","origin":service.origin,"challenge":field(&challenge,"challenge")?}), + )?; + let request = json!({"op":"authenticate","id":field(&challenge,"id")?,"signature":field(&proof,"signature")?}); + let result = service.call(&request)?; + ensure!(service.call(&request).is_err(), "Challenge was replayed"); + Ok(result) + } + + fn reject_expired_and_failed_proofs(service: &mut Service, identity: &Identity) -> Result<()> { + let challenge = service.call(&json!({"op":"challenge","public":identity.public}))?; + let challenge_id = field(&challenge, "id")?; + service + .challenges + .get_mut(challenge_id) + .context("Challenge")? + .expiry = Instant::now(); + let expired = json!({"op":"authenticate","id":challenge_id,"signature":""}); + ensure!( + service.call(&expired).is_err(), + "Expired challenge accepted" + ); + ensure!( + !service.challenges.contains_key(challenge_id), + "Expired challenge retained" + ); + let invalid = service.call(&json!({"op":"challenge","public":identity.public}))?; + let invalid_id = field(&invalid, "id")?; + ensure!( + service + .call(&json!({"op":"authenticate","id":invalid_id,"signature":""})) + .is_err(), + "Invalid proof accepted" + ); + ensure!( + !service.challenges.contains_key(invalid_id), + "Failed challenge reusable" + ); + service.attempts = 60; + ensure!( + service + .call(&json!({"op":"challenge","public":identity.public})) + .is_err(), + "Challenge rate bound" + ); + ensure!( + service.store.install_owner(&identity.public).is_err(), + "Owner overwritten" + ); + Ok(()) + } + + #[test] + fn authorization_replays_history_and_snapshot_recovery() -> Result<()> { + let directory = TempDir::new()?; + let path = directory.path().join("chat.sqlite"); + let setup = token()?; + let mut service = Service::new( + Store::open(&path)?, + "http://127.0.0.1:9080".into(), + setup.clone(), + ); + let owner = identity("Owner", "")?; + ensure!( + service + .call(&json!({"op":"challenge","public":owner.public,"setup_token":"wrong"})) + .is_err(), + "Setup unauthenticated" + ); + let owner_session = authenticate(&mut service, &owner, &setup)?; + ensure!( + owner_session.get("owner") == Some(&json!(true)), + "Owner role" + ); + let visitor = identity("Visitor", &owner.public)?; + let visitor_session = authenticate(&mut service, &visitor, "")?; + let visitor_token = field(&visitor_session, "token")?; + let owner_token = field(&owner_session, "token")?; + let ciphertext = BASE64_STANDARD.encode([42; 128]); + let message_id = token()?; + let send = json!({"op":"send","conversation":visitor.fingerprint,"token":visitor_token,"id":message_id,"ciphertext":ciphertext}); + service.call(&send)?; + service.call(&send)?; + let conflicting = json!({"op":"send","conversation":visitor.fingerprint,"token":visitor_token,"id":message_id,"ciphertext":BASE64_STANDARD.encode([43;128])}); + ensure!( + service.call(&conflicting).is_err(), + "Conflicting ID accepted" + ); + let empty = service.call(&json!({"op":"history","conversation":visitor.fingerprint,"token":visitor_token,"count":0}))?; + ensure!( + empty + .get("messages") + .and_then(Value::as_array) + .is_some_and(Vec::is_empty), + "N=0" + ); + ensure!( + empty + .get("conversation") + .and_then(|value| value.get("total")) + == Some(&json!(1)), + "Retry duplicated message" + ); + ensure!( + service + .call(&json!({"op":"inbox","token":visitor_token})) + .is_err(), + "Visitor read inbox" + ); + ensure!(service.call(&json!({"op":"history","conversation":owner.fingerprint,"token":visitor_token,"count":1})).is_err(), "Wrong conversation accepted"); + service.call(&json!({"op":"send","conversation":visitor.fingerprint,"token":owner_token,"id":token()?,"ciphertext":ciphertext}))?; + let latest = service.call(&json!({"op":"history","conversation":visitor.fingerprint,"token":visitor_token,"count":1}))?; + let sequence = latest + .get("messages") + .and_then(Value::as_array) + .and_then(|messages| messages.first()) + .and_then(|message| message.get("seq")) + .context("Sequence")?; + let older = service.call(&json!({"op":"history","conversation":visitor.fingerprint,"token":visitor_token,"count":1,"before":sequence}))?; + ensure!( + older + .get("messages") + .and_then(Value::as_array) + .is_some_and(|messages| messages.len() == 1), + "Pagination" + ); + ensure!(service.call(&json!({"op":"history","conversation":visitor.fingerprint,"token":visitor_token,"count":51})).is_err(), "History quota"); + reject_expired_and_failed_proofs(&mut service, &visitor)?; + let snapshot = directory.path().join("snapshot.sqlite"); + service.store.snapshot(&snapshot)?; + let restored = Store::open(&snapshot)?; + ensure!( + restored.owner()? == Some(owner.public), + "Owner missing after restore" + ); + ensure!( + restored.conversation(&visitor.fingerprint)?.get("total") == Some(&json!(2)), + "Restored count" + ); + service.call(&json!({"op":"lock","token":visitor_token}))?; + ensure!(service.call(&json!({"op":"history","conversation":visitor.fingerprint,"token":visitor_token,"count":0})).is_err(), "Locked session accepted"); + let session = service.sessions.get_mut(owner_token).context("Session")?; + session.expiry = Instant::now(); + ensure!( + service + .call(&json!({"op":"inbox","token":owner_token})) + .is_err(), + "Expired session accepted" + ); + Ok(()) + } +} diff --git a/crates/chat-server/src/app/store.rs b/crates/chat-server/src/app/store.rs new file mode 100644 index 0000000..23e8ba1 --- /dev/null +++ b/crates/chat-server/src/app/store.rs @@ -0,0 +1,182 @@ +// Copyright (C) 2026 Doan Luu Duc Tai <doan@luuduct.ai> +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see <https://www.gnu.org/licenses/>. + +use anyhow::{Context, Result, ensure}; +use rusqlite::{Connection, OptionalExtension, params}; +use serde_json::{Value, json}; +use std::path::Path; + +pub(super) struct Store { + connection: Connection, +} + +impl Store { + pub(super) fn open(path: &Path) -> Result<Self> { + let connection = Connection::open(path)?; + connection.execute_batch("PRAGMA foreign_keys=ON; PRAGMA journal_mode=DELETE; PRAGMA synchronous=FULL; PRAGMA secure_delete=ON; CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS conversations (id TEXT PRIMARY KEY, cert TEXT NOT NULL, profile TEXT); CREATE TABLE IF NOT EXISTS messages (seq INTEGER PRIMARY KEY AUTOINCREMENT, conversation TEXT NOT NULL REFERENCES conversations(id), id TEXT NOT NULL UNIQUE, sender TEXT NOT NULL, ciphertext TEXT NOT NULL);")?; + Ok(Self { connection }) + } + + pub(super) fn owner(&self) -> Result<Option<String>> { + Ok(self + .connection + .query_row("SELECT value FROM metadata WHERE key='owner'", [], |row| { + row.get(0) + }) + .optional()?) + } + + pub(super) fn install_owner(&self, cert: &str) -> Result<()> { + ensure!(self.owner()?.is_none(), "Already initialized"); + self.connection + .execute("INSERT INTO metadata VALUES ('owner', ?1)", [cert])?; + Ok(()) + } + + pub(super) fn reset_owner(&mut self, expected: &str) -> Result<()> { + let owner = self.owner()?.context("No owner identity is registered")?; + let fingerprint = stack_chat_crypto::public_identity(&owner)?; + ensure!( + expected.len() == 64 && expected.eq_ignore_ascii_case(&fingerprint), + "Owner fingerprint does not match; no data was cleared" + ); + let transaction = self.connection.transaction()?; + transaction.execute_batch("DELETE FROM messages; DELETE FROM conversations; DELETE FROM metadata WHERE key='owner'; DELETE FROM sqlite_sequence WHERE name='messages';")?; + transaction.commit()?; + Ok(()) + } + + pub(super) fn register(&self, id: &str, cert: &str) -> Result<()> { + if self.connection.query_row( + "SELECT EXISTS(SELECT 1 FROM conversations WHERE id=?1)", + [id], + |row| row.get::<_, bool>(0), + )? { + return Ok(()); + } + let count: i64 = + self.connection + .query_row("SELECT COUNT(*) FROM conversations", [], |row| row.get(0))?; + ensure!(count < 1000, "Conversation limit reached"); + self.connection.execute( + "INSERT INTO conversations (id,cert) VALUES (?1,?2)", + params![id, cert], + )?; + Ok(()) + } + + pub(super) fn conversation(&self, id: &str) -> Result<Value> { + let (cert, profile): (String, Option<String>) = self.connection.query_row( + "SELECT cert,profile FROM conversations WHERE id=?1", + [id], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + let (total, visitor): (i64, i64) = self.connection.query_row( + "SELECT COUNT(*),COALESCE(SUM(sender=?1),0) FROM messages WHERE conversation=?1", + [id], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + Ok( + json!({"id":id,"public":cert,"profile":profile,"total":total,"visitor_sent":visitor,"owner_sent":total.checked_sub(visitor).context("Counter invariant")?}), + ) + } + + pub(super) fn inbox(&self) -> Result<Value> { + let mut statement = self + .connection + .prepare("SELECT id FROM conversations ORDER BY id")?; + let ids = statement + .query_map([], |row| row.get::<_, String>(0))? + .collect::<rusqlite::Result<Vec<_>>>()?; + let conversations = ids + .iter() + .map(|id| self.conversation(id)) + .collect::<Result<Vec<_>>>()?; + Ok(json!({"conversations":conversations})) + } + + pub(super) fn profile(&self, id: &str, ciphertext: &str) -> Result<()> { + self.connection.execute( + "UPDATE conversations SET profile=?2 WHERE id=?1 AND profile IS NULL", + params![id, ciphertext], + )?; + Ok(()) + } + + pub(super) fn messages(&self, id: &str, count: u64, before: u64) -> Result<Value> { + ensure!( + count <= 50 && before <= i64::MAX.unsigned_abs(), + "Invalid history window" + ); + let mut statement = self.connection.prepare("SELECT seq,id,sender,ciphertext FROM messages WHERE conversation=?1 AND (?3=0 OR seq<?3) ORDER BY seq DESC LIMIT ?2")?; + let messages = statement.query_map(params![id,i64::try_from(count)?,i64::try_from(before)?], |row| Ok(json!({"seq":row.get::<_,i64>(0)?,"id":row.get::<_,String>(1)?,"sender":row.get::<_,String>(2)?,"ciphertext":row.get::<_,String>(3)?})))?.collect::<rusqlite::Result<Vec<_>>>()?; + Ok(json!({"conversation":self.conversation(id)?,"messages":messages})) + } + + pub(super) fn send( + &mut self, + conversation: &str, + id: &str, + sender: &str, + ciphertext: &str, + ) -> Result<Value> { + let transaction = self.connection.transaction()?; + let existing: Option<(i64, String, String, String)> = transaction + .query_row( + "SELECT seq,conversation,sender,ciphertext FROM messages WHERE id=?1", + [id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + if let Some((seq, previous_conversation, previous_sender, previous_ciphertext)) = existing { + ensure!( + previous_conversation == conversation + && previous_sender == sender + && previous_ciphertext == ciphertext, + "Conflicting message identifier" + ); + transaction.commit()?; + return Ok(json!({"seq":seq,"conversation":self.conversation(conversation)?})); + } + let count: i64 = transaction.query_row( + "SELECT COUNT(*) FROM messages WHERE conversation=?1", + [conversation], + |row| row.get(0), + )?; + let used: i64 = transaction.query_row( + "SELECT COALESCE(SUM(LENGTH(ciphertext)),0) FROM messages", + [], + |row| row.get(0), + )?; + ensure!( + count < 10_000 && used < 1_000_000_000, + "Storage quota reached" + ); + transaction.execute( + "INSERT INTO messages (conversation,id,sender,ciphertext) VALUES (?1,?2,?3,?4)", + params![conversation, id, sender, ciphertext], + )?; + let sequence = transaction.last_insert_rowid(); + transaction.commit()?; + Ok(json!({"seq":sequence,"conversation":self.conversation(conversation)?})) + } + + pub(super) fn snapshot(&self, destination: &Path) -> Result<()> { + ensure!(!destination.exists(), "Snapshot destination already exists"); + self.connection + .backup("main", destination, None::<fn(rusqlite::backup::Progress)>)?; + Ok(()) + } +} |