From a80df869a92c330fbad9695057720b11a18af901 Mon Sep 17 00:00:00 2001 From: Doan Luu Duc Tai Date: Sat, 31 Oct 2026 13:08:08 -0500 Subject: Forbid Rust lints and enforce strict typed TypeScript linting --- crates/chat-server/src/app.rs | 199 ++++++++++++++ crates/chat-server/src/app/cli.rs | 167 ++++++++++++ crates/chat-server/src/app/http.rs | 243 +++++++++++++++++ crates/chat-server/src/app/maintenance.rs | 72 +++++ crates/chat-server/src/app/service.rs | 438 ++++++++++++++++++++++++++++++ crates/chat-server/src/app/store.rs | 182 +++++++++++++ crates/chat-server/src/http.rs | 243 ----------------- crates/chat-server/src/main.rs | 202 +------------- crates/chat-server/src/maintenance.rs | 72 ----- crates/chat-server/src/service.rs | 438 ------------------------------ crates/chat-server/src/store.rs | 182 ------------- 11 files changed, 1306 insertions(+), 1132 deletions(-) create mode 100644 crates/chat-server/src/app.rs create mode 100644 crates/chat-server/src/app/cli.rs create mode 100644 crates/chat-server/src/app/http.rs create mode 100644 crates/chat-server/src/app/maintenance.rs create mode 100644 crates/chat-server/src/app/service.rs create mode 100644 crates/chat-server/src/app/store.rs delete mode 100644 crates/chat-server/src/http.rs delete mode 100644 crates/chat-server/src/maintenance.rs delete mode 100644 crates/chat-server/src/service.rs delete mode 100644 crates/chat-server/src/store.rs (limited to 'crates/chat-server/src') diff --git a/crates/chat-server/src/app.rs b/crates/chat-server/src/app.rs new file mode 100644 index 0000000..31efd28 --- /dev/null +++ b/crates/chat-server/src/app.rs @@ -0,0 +1,199 @@ +// Copyright (C) 2026 Doan Luu Duc Tai +// +// 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 . + +#![doc = "Encrypted private chat server and backend maintenance commands."] + +mod cli; +mod http; +mod maintenance; +mod service; +mod store; + +use crate::app::cli::Arguments; +use anyhow::{Context, Result, ensure}; +use std::io::{Read, Write}; +use std::net::SocketAddr; +use std::process::ExitCode; + +fn initialize(root: &std::path::Path) -> Result<()> { + let env = root.join(".env"); + let existing = if env.exists() { + std::fs::read_to_string(&env)? + } else { + String::new() + }; + ensure!( + !existing + .lines() + .any(|line| line.starts_with("CHAT_SETUP_TOKEN=")), + "CHAT_SETUP_TOKEN already exists" + ); + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&env)?; + std::fs::set_permissions(&env, std::os::unix::fs::PermissionsExt::from_mode(0o600))?; + writeln!(file, "\nCHAT_SETUP_TOKEN={}", service::token()?)?; + if !existing + .lines() + .any(|line| line.starts_with("CHAT_ORIGIN=")) + { + writeln!(file, "CHAT_ORIGIN=http://127.0.0.1:9080")?; + } + stack_config::initialize_public(root) +} + +fn healthcheck(mut address: SocketAddr) -> Result<()> { + if address.ip().is_unspecified() { + address.set_ip(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)); + } + let mut socket = + std::net::TcpStream::connect_timeout(&address, std::time::Duration::from_secs(2))?; + socket.set_read_timeout(Some(std::time::Duration::from_secs(2)))?; + socket.set_write_timeout(Some(std::time::Duration::from_secs(2)))?; + socket.write_all(b"GET /healthz HTTP/1.0\r\nHost: localhost\r\n\r\n")?; + let mut response = String::new(); + socket.take(8192).read_to_string(&mut response)?; + ensure!( + response.starts_with("HTTP/1.0 200") || response.starts_with("HTTP/1.1 200"), + "Chat health endpoint failed" + ); + Ok(()) +} + +fn validate_origin( + configuration: &stack_config::Config, + bind: SocketAddr, + proxy: bool, +) -> Result<()> { + let origin = configuration.value("CHAT_ORIGIN")?; + if proxy { + let public = configuration.public_values(None)?; + let canonical = public + .get("CHAT_PUBLIC_ORIGIN") + .context("Missing public chat origin")?; + let local = format!( + "http://{}:{}", + configuration.value("CHAT_LOCAL_HOST")?, + configuration.port("LOCAL_HTTP_PORT")? + ); + ensure!( + origin == canonical || origin == local, + "Gateway chat origin must match the configured HTTPS hostname or dedicated local hostname" + ); + } else { + ensure!( + bind.ip().is_loopback() && origin == format!("http://{bind}"), + "Direct development chat requires an exact loopback HTTP origin" + ); + } + Ok(()) +} + +async fn serve(arguments: Arguments) -> Result<()> { + let root = arguments.root.canonicalize()?; + if arguments.init { + return initialize(&root); + } + let configuration = if arguments.listener.proxy { + stack_config::Config::from_environment()? + } else { + stack_config::Config::load(&root)? + }; + let bind = arguments + .listener + .bind + .unwrap_or(configuration.chat_bind()?); + if arguments.healthcheck { + return healthcheck(bind); + } + let data = root.join("var/data/chat"); + std::fs::create_dir_all(&data)?; + std::fs::set_permissions(&data, std::os::unix::fs::PermissionsExt::from_mode(0o700))?; + let database = data.join("chat.sqlite"); + let _server_lock = if arguments.snapshot.is_none() && !arguments.owner_fingerprint { + Some(maintenance::lock(&data)?) + } else { + None + }; + let mut store = store::Store::open(&database)?; + std::fs::set_permissions( + &database, + std::os::unix::fs::PermissionsExt::from_mode(0o600), + )?; + if arguments.owner_fingerprint { + return maintenance::fingerprint(&store); + } + if let Some(expected) = arguments.reset_owner { + return maintenance::reset(&mut store, &expected, bind, arguments.listener.proxy); + } + if let Some(destination) = arguments.snapshot { + store.snapshot(&destination)?; + std::fs::set_permissions( + destination, + std::os::unix::fs::PermissionsExt::from_mode(0o600), + )?; + return Ok(()); + } + let origin = configuration.value("CHAT_ORIGIN")?; + validate_origin(&configuration, bind, arguments.listener.proxy)?; + let setup = configuration + .value("CHAT_SETUP_TOKEN") + .context("Run --init to configure chat")?; + let state = service::Service::new(store, origin.to_owned(), setup.to_owned()); + let listener = tokio::net::TcpListener::bind(bind).await?; + writeln!( + std::io::stdout().lock(), + "Chat ready: {origin} (owner setup: /#setup)" + )?; + let local = (arguments.listener.proxy && origin.starts_with("http://")) + .then(|| configuration.port("LOCAL_HTTP_PORT")) + .transpose()?; + http::serve(listener, state, &root, configuration.public_values(local)?).await +} + +fn execute(arguments: Arguments) -> Result<()> { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()?; + runtime.block_on(serve(arguments)) +} + +#[doc = "Runs the checked command-line entry point and returns its process status."] +#[must_use] +pub fn run() -> ExitCode { + let arguments = match Arguments::try_parse() { + Ok(arguments) => arguments, + Err(error) => { + let status = if error.use_stderr() { + ExitCode::from(2) + } else { + ExitCode::SUCCESS + }; + drop(error.print()); + return status; + } + }; + match execute(arguments) { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + drop(writeln!( + std::io::stderr().lock(), + "Chat server failed: {error:#}" + )); + ExitCode::FAILURE + } + } +} 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 +// +// 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 . + +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, + + pub(super) owner_fingerprint: bool, + + pub(super) snapshot: Option, +} + +#[derive(Debug)] +pub(super) struct Listener { + pub(super) bind: Option, + + 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::), + 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::try_parse_from(std::env::args_os()) + } + + fn try_parse_from( + arguments: impl IntoIterator + Clone>, + ) -> Result { + 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 +// +// 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 . + +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>, + permits: Arc, + root: PathBuf, + origin: String, + public: std::collections::BTreeMap, +} + +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, headers: HeaderMap, Json(request): Json) -> 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, 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, +) -> 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 +// +// 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 . + +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 { + 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 +// +// 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 . + +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, + sessions: HashMap, + window: Instant, + attempts: u16, +} + +pub(super) fn token() -> Result { + 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 { + stack_chat_crypto::public_identity(&self.store.owner()?.context("Owner not initialized")?) + } + + fn challenge(&mut self, request: &Value) -> Result { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 +// +// 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 . + +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 { + 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> { + 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 { + let (cert, profile): (String, Option) = 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 { + let mut statement = self + .connection + .prepare("SELECT id FROM conversations ORDER BY id")?; + let ids = statement + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + let conversations = ids + .iter() + .map(|id| self.conversation(id)) + .collect::>>()?; + 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 { + 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(0)?,"id":row.get::<_,String>(1)?,"sender":row.get::<_,String>(2)?,"ciphertext":row.get::<_,String>(3)?})))?.collect::>>()?; + Ok(json!({"conversation":self.conversation(id)?,"messages":messages})) + } + + pub(super) fn send( + &mut self, + conversation: &str, + id: &str, + sender: &str, + ciphertext: &str, + ) -> Result { + 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::)?; + Ok(()) + } +} diff --git a/crates/chat-server/src/http.rs b/crates/chat-server/src/http.rs deleted file mode 100644 index cb21b1f..0000000 --- a/crates/chat-server/src/http.rs +++ /dev/null @@ -1,243 +0,0 @@ -// Copyright (C) 2026 Doan Luu Duc Tai -// -// 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 . - -use crate::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>, - permits: Arc, - root: PathBuf, - origin: String, - public: std::collections::BTreeMap, -} - -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, headers: HeaderMap, Json(request): Json) -> 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, 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(crate) async fn serve( - listener: tokio::net::TcpListener, - service: Service, - root: &Path, - public: std::collections::BTreeMap, -) -> 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/main.rs b/crates/chat-server/src/main.rs index 1d600b7..942a379 100644 --- a/crates/chat-server/src/main.rs +++ b/crates/chat-server/src/main.rs @@ -13,204 +13,12 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -#![doc = "Encrypted private chat server and backend maintenance commands."] +#![doc = "Checked command-line entry point for the personal infrastructure stack."] -mod http; -mod maintenance; -mod service; -mod store; +mod app; -use anyhow::{Context, Result, ensure}; -use clap::Parser; -use std::io::{Read, Write}; -use std::net::SocketAddr; -use std::path::PathBuf; -use std::process::ExitCode; +pub use app::run; -#[derive(Debug, Parser)] -#[command(about = "Run private chat directly or behind the stack gateway")] -struct Arguments { - #[arg(long, default_value = ".")] - root: PathBuf, - #[command(flatten)] - listener: Listener, - #[arg( - long, - help = "Check the local HTTP health endpoint without opening the database" - )] - healthcheck: bool, - #[arg(long, conflicts_with_all = ["reset_owner", "owner_fingerprint", "snapshot"])] - init: bool, - #[arg(long, value_name = "FINGERPRINT", conflicts_with_all = ["owner_fingerprint", "snapshot"], help = "Permanently clear the matching owner registration and all chat history after key loss")] - reset_owner: Option, - #[arg( - long, - conflicts_with = "snapshot", - help = "Print the registered owner's public fingerprint" - )] - owner_fingerprint: bool, - #[arg(long)] - snapshot: Option, -} - -#[derive(Debug, clap::Args)] -struct Listener { - #[arg(long)] - bind: Option, - #[arg( - long, - help = "Run behind the stack gateway using its explicit container environment" - )] - proxy: bool, -} - -fn initialize(root: &std::path::Path) -> Result<()> { - let env = root.join(".env"); - let existing = if env.exists() { - std::fs::read_to_string(&env)? - } else { - String::new() - }; - ensure!( - !existing - .lines() - .any(|line| line.starts_with("CHAT_SETUP_TOKEN=")), - "CHAT_SETUP_TOKEN already exists" - ); - let mut file = std::fs::OpenOptions::new() - .create(true) - .append(true) - .open(&env)?; - std::fs::set_permissions(&env, std::os::unix::fs::PermissionsExt::from_mode(0o600))?; - writeln!(file, "\nCHAT_SETUP_TOKEN={}", service::token()?)?; - if !existing - .lines() - .any(|line| line.starts_with("CHAT_ORIGIN=")) - { - writeln!(file, "CHAT_ORIGIN=http://127.0.0.1:9080")?; - } - stack_config::initialize_public(root) -} - -fn healthcheck(mut address: SocketAddr) -> Result<()> { - if address.ip().is_unspecified() { - address.set_ip(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)); - } - let mut socket = - std::net::TcpStream::connect_timeout(&address, std::time::Duration::from_secs(2))?; - socket.set_read_timeout(Some(std::time::Duration::from_secs(2)))?; - socket.set_write_timeout(Some(std::time::Duration::from_secs(2)))?; - socket.write_all(b"GET /healthz HTTP/1.0\r\nHost: localhost\r\n\r\n")?; - let mut response = String::new(); - socket.take(8192).read_to_string(&mut response)?; - ensure!( - response.starts_with("HTTP/1.0 200") || response.starts_with("HTTP/1.1 200"), - "Chat health endpoint failed" - ); - Ok(()) -} - -fn validate_origin( - configuration: &stack_config::Config, - bind: SocketAddr, - proxy: bool, -) -> Result<()> { - let origin = configuration.value("CHAT_ORIGIN")?; - if proxy { - let public = configuration.public_values(None)?; - let canonical = public - .get("CHAT_PUBLIC_ORIGIN") - .context("Missing public chat origin")?; - let local = format!( - "http://{}:{}", - configuration.value("CHAT_LOCAL_HOST")?, - configuration.port("LOCAL_HTTP_PORT")? - ); - ensure!( - origin == canonical || origin == local, - "Gateway chat origin must match the configured HTTPS hostname or dedicated local hostname" - ); - } else { - ensure!( - bind.ip().is_loopback() && origin == format!("http://{bind}"), - "Direct development chat requires an exact loopback HTTP origin" - ); - } - Ok(()) -} - -async fn run(arguments: Arguments) -> Result<()> { - let root = arguments.root.canonicalize()?; - if arguments.init { - return initialize(&root); - } - let configuration = if arguments.listener.proxy { - stack_config::Config::from_environment()? - } else { - stack_config::Config::load(&root)? - }; - let bind = arguments - .listener - .bind - .unwrap_or(configuration.chat_bind()?); - if arguments.healthcheck { - return healthcheck(bind); - } - let data = root.join("var/data/chat"); - std::fs::create_dir_all(&data)?; - std::fs::set_permissions(&data, std::os::unix::fs::PermissionsExt::from_mode(0o700))?; - let database = data.join("chat.sqlite"); - let _server_lock = if arguments.snapshot.is_none() && !arguments.owner_fingerprint { - Some(maintenance::lock(&data)?) - } else { - None - }; - let mut store = store::Store::open(&database)?; - std::fs::set_permissions( - &database, - std::os::unix::fs::PermissionsExt::from_mode(0o600), - )?; - if arguments.owner_fingerprint { - return maintenance::fingerprint(&store); - } - if let Some(expected) = arguments.reset_owner { - return maintenance::reset(&mut store, &expected, bind, arguments.listener.proxy); - } - if let Some(destination) = arguments.snapshot { - store.snapshot(&destination)?; - std::fs::set_permissions( - destination, - std::os::unix::fs::PermissionsExt::from_mode(0o600), - )?; - return Ok(()); - } - let origin = configuration.value("CHAT_ORIGIN")?; - validate_origin(&configuration, bind, arguments.listener.proxy)?; - let setup = configuration - .value("CHAT_SETUP_TOKEN") - .context("Run --init to configure chat")?; - let state = service::Service::new(store, origin.to_owned(), setup.to_owned()); - let listener = tokio::net::TcpListener::bind(bind).await?; - writeln!( - std::io::stdout().lock(), - "Chat ready: {origin} (owner setup: /#setup)" - )?; - let local = (arguments.listener.proxy && origin.starts_with("http://")) - .then(|| configuration.port("LOCAL_HTTP_PORT")) - .transpose()?; - http::serve(listener, state, &root, configuration.public_values(local)?).await -} - -#[tokio::main] -async fn main() -> ExitCode { - match run(Arguments::parse()).await { - Ok(()) => ExitCode::SUCCESS, - Err(error) => { - drop(writeln!( - std::io::stderr().lock(), - "Chat server failed: {error:#}" - )); - ExitCode::FAILURE - } - } +fn main() -> std::process::ExitCode { + run() } diff --git a/crates/chat-server/src/maintenance.rs b/crates/chat-server/src/maintenance.rs deleted file mode 100644 index d416d73..0000000 --- a/crates/chat-server/src/maintenance.rs +++ /dev/null @@ -1,72 +0,0 @@ -// Copyright (C) 2026 Doan Luu Duc Tai -// -// 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 . - -use crate::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(crate) fn lock(directory: &Path) -> Result { - 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(crate) 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(crate) 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/service.rs b/crates/chat-server/src/service.rs deleted file mode 100644 index 3f9b6ee..0000000 --- a/crates/chat-server/src/service.rs +++ /dev/null @@ -1,438 +0,0 @@ -// Copyright (C) 2026 Doan Luu Duc Tai -// -// 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 . - -use crate::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(crate) struct Service { - pub(crate) store: Store, - pub(crate) origin: String, - setup_token: String, - challenges: HashMap, - sessions: HashMap, - window: Instant, - attempts: u16, -} - -pub(crate) fn token() -> Result { - let mut bytes = [0; 32]; - openssl::rand::rand_bytes(&mut bytes)?; - Ok(hex::encode(bytes)) -} - -pub(crate) 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(crate) 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 { - stack_chat_crypto::public_identity(&self.store.owner()?.context("Owner not initialized")?) - } - - fn challenge(&mut self, request: &Value) -> Result { - 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 { - 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 { - 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 { - let conversation = field(request, "conversation")?; - identifier(conversation)?; - ensure!( - identity == conversation || identity == self.owner_id()?, - "Forbidden conversation" - ); - self.store.conversation(conversation)?; - Ok(conversation.into()) - } - - pub(crate) fn call(&mut self, request: &Value) -> Result { - 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 { - 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 { - 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 { - 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 { - 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/store.rs b/crates/chat-server/src/store.rs deleted file mode 100644 index 0e09de7..0000000 --- a/crates/chat-server/src/store.rs +++ /dev/null @@ -1,182 +0,0 @@ -// Copyright (C) 2026 Doan Luu Duc Tai -// -// 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 . - -use anyhow::{Context, Result, ensure}; -use rusqlite::{Connection, OptionalExtension, params}; -use serde_json::{Value, json}; -use std::path::Path; - -pub(crate) struct Store { - connection: Connection, -} - -impl Store { - pub(crate) fn open(path: &Path) -> Result { - 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(crate) fn owner(&self) -> Result> { - Ok(self - .connection - .query_row("SELECT value FROM metadata WHERE key='owner'", [], |row| { - row.get(0) - }) - .optional()?) - } - - pub(crate) 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(crate) 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(crate) 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(crate) fn conversation(&self, id: &str) -> Result { - let (cert, profile): (String, Option) = 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(crate) fn inbox(&self) -> Result { - let mut statement = self - .connection - .prepare("SELECT id FROM conversations ORDER BY id")?; - let ids = statement - .query_map([], |row| row.get::<_, String>(0))? - .collect::>>()?; - let conversations = ids - .iter() - .map(|id| self.conversation(id)) - .collect::>>()?; - Ok(json!({"conversations":conversations})) - } - - pub(crate) 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(crate) fn messages(&self, id: &str, count: u64, before: u64) -> Result { - 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(0)?,"id":row.get::<_,String>(1)?,"sender":row.get::<_,String>(2)?,"ciphertext":row.get::<_,String>(3)?})))?.collect::>>()?; - Ok(json!({"conversation":self.conversation(id)?,"messages":messages})) - } - - pub(crate) fn send( - &mut self, - conversation: &str, - id: &str, - sender: &str, - ciphertext: &str, - ) -> Result { - 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(crate) fn snapshot(&self, destination: &Path) -> Result<()> { - ensure!(!destination.exists(), "Snapshot destination already exists"); - self.connection - .backup("main", destination, None::)?; - Ok(()) - } -} -- cgit v1.2.3