commit 105ad89aceca1c40fd753e107d3cb81c21e649c3 Author: LIm1tZz7 <36764786+TheBadRoger@users.noreply.github.com> Date: Thu Oct 15 17:58:54 2026 +0800 Add lightweight async ChuSQL Rust driver and blog query support diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ea8c4bf --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/target diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..d064b63 --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,194 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "chusql-driver" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "tokio", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.190" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.154" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "syn" +version = "3.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tokio" +version = "1.53.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "unicode-ident" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..49b47d8 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "chusql-driver" +version = "0.1.0" +edition = "2024" +description = "Lightweight asynchronous Rust driver for ChuSQL" + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = { version = "1", features = ["arbitrary_precision"] } +tokio = { version = "1", features = ["net", "io-util", "sync", "time"] } + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread"] } diff --git a/README.md b/README.md new file mode 100644 index 0000000..954eefb --- /dev/null +++ b/README.md @@ -0,0 +1,108 @@ +# ChuSQL Rust Driver + +轻量异步 ChuSQL 数据库驱动，使用 Tokio，通过 TCP 直接连接 `chusql-server`。可作为 Axum、Actix Web 等 Tokio 后端框架的应用状态使用。 + +实现依据是相邻 ChuSQL 项目的 `chusql-interface/src/ChuSQL/Interface/Protocol.hs`、`Link.hs` 和 `chusql-core/model/src/ChuSQL/Core/Protocol.hs`，使用版本 1 的 UTF-8 JSON 行协议。依赖仅有 Tokio、Serde 和 serde_json，没有 ORM 或连接池层。 + +## 接入 + +在后端项目的 `Cargo.toml` 中添加： + +```toml +[dependencies] +chusql-driver = { path = "../ChuSQLDriver" } +tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +``` + +```rust,no_run +use chusql_driver::{Client, Error}; + +#[tokio::main] +async fn main() -> Result<(), Error> { + let db = Client::connect("127.0.0.1:7777", "app", "your-password").await?; + db.ping().await?; + let result = db.query("SHOW DATABASES").await?; + println!("{:?}", result.columns); + for row in &result.rows { + println!("{row:?}"); + } + if result.truncated { + println!("结果已被服务端截断，请缩小查询范围"); + } + db.close().await; + Ok(()) +} +``` + +需要已初始化并启动 ChuSQL TCP 服务，且登录账号具备相应 SQL 权限。默认地址为 `127.0.0.1:7777`。 + +`Client` 可 `clone()`，支持 `Send + Sync`。将它传给 Axum 的 `Router::with_state(db)` 或 Actix Web 的 `web::Data::new(db)`，handler 中直接调用 `db.query(sql).await`。所有克隆共享同一个登录会话，同一时刻只执行一个请求；需要并行执行时建立多个独立连接。不要在多个 handler 间交错执行依赖会话状态的一组 SQL（例如切换数据库或事务），这类操作应使用独立连接。 + +## API 与结果 + +- `Client::connect(address, user, password)`：连接、协议握手、账号登录。 +- `Client::connect_with_options(address, user, password, options)`：自定义连接/请求超时和响应字节上限。 +- `client.query(sql)`：执行单条 SQL，DDL、DML 和 SELECT 使用相同入口。 +- `client.ping()`：探活。 +- `client.close()`：关闭所有克隆共享的连接；也可在最后一个客户端释放时自动断开。 + +`QueryResult` 保留服务端的 `columns`、`rows`、`row_count`、`truncated` 和 `database`。行内值使用 `serde_json::Value`，可以使用 `as_str()`、`as_i64()`、`as_bool()`、`is_null()` 等方法读取。数值启用任意精度 JSON 表示，运行时自定义类型对象原样保留。`row_count` 是服务端返回的行数，不承诺等同于写操作影响行数；务必检查 `truncated`。 + +```rust +use chusql_driver::Options; +use std::time::Duration; + +let options = Options { + timeout: Duration::from_secs(5), + max_response_bytes: 8 * 1024 * 1024, +}; +``` + +默认连接及每次请求超时为 30 秒，单条响应上限为 16 MiB（含换行）；等待其他请求释放连接的时间不计入超时。可匹配 `Error::Server { code, message }` 处理服务端错误，普通 SQL 错误不会关闭连接。I/O、JSON、超时或取消会使连接失效，后续返回 `Error::Closed`，需要重新建立连接。驱动不自动重试写入：断线或超时发生时，SQL 可能已执行。 + +## 边界 + +当前 ChuSQL 协议只接收 SQL 文本，不提供服务端参数绑定。新增的 `Query` 按 ChuSQL 解析规则编码客户端字面量；不要将用户输入直接拼接到 SQL，也不要将其传给 `Query::new` 或 `push`。 + +```rust +use chusql_driver::Query; + +let title = "It's my blog"; +let statement = Query::new("INSERT INTO posts (id, title) VALUES (") + .bind(1_i64) + .push(", ") + .bind(title) + .push(")"); +// db.query(statement.as_str()).await?; +``` + +`bind` 支持 `&str`、`i64`、`bool` 和 `Param::Null`，仅用于 SQL 值。表名和列名必须由代码固定或使用白名单；LIKE 通配符仍有其原本语义。这不是服务端预编译语句，也不模拟问号占位符。 + +## 轻量博客 + +当前驱动适合低并发个人博客的文章、评论和分类读写。新增 `examples/blog.rs` 展示建表、含单引号/中文/换行的正文写入、更新、列表和删除。对一个空的测试数据库设置 `CHUSQL_DATABASE` 后运行 `cargo run --example blog`；示例会创建并最终删除 `driver_blog_posts`，中途失败时可能保留表。数据库需要事先创建，账号需要建表和读写权限。 + +查询可使用 `ORDER BY id DESC LIMIT 10`，翻页使用 `WHERE id < 上一页最后的ID`，通过 `bind` 传入 ID；当前解析器没有 OFFSET。大正文应按需查询，避免列表加载全部正文。ChuSQL 默认请求上限为 1 MiB、结果行数上限为 1000，图片建议独立存储。 + +一个共享连接可以作为低流量起点；事务和 `USE` 需要独立连接。超时后重新连接，但不要盲目重试发文；使用唯一 ID 或唯一 slug 避免重复提交。生产使用前仍需对真实 ChuSQL 验证权限、备份恢复和预期并发，驱动本身不保证数据库的成熟度。 + +服务器协议通过明文 TCP 传输账号密码及查询数据，跨不可信网络时请使用受保护的网络或加密隧道。本驱动不内置 TLS、ORM 适配、连接池、迁移或自动重连，先保持核心连接功能简单。 + +## 验证与示例 + +```sh +cargo test --all-targets +cargo fmt --check +cargo clippy --all-targets -- -D warnings +``` + +协议测试使用本地 TCP 测试服务覆盖握手、登录、Unicode/换行、结果映射、错误恢复、响应分片、损坏/超长/未终止响应、超时和请求取消。它们不替代真实数据库联调。 + +对运行中的真实数据库执行查询： + +```sh +export CHUSQL_ADDRESS=127.0.0.1:7777 +export CHUSQL_USER=app +# 在运行环境中设置 CHUSQL_PASSWORD +cargo run --example query -- 'SHOW DATABASES' +``` diff --git a/examples/blog.rs b/examples/blog.rs new file mode 100644 index 0000000..be86fb9 --- /dev/null +++ b/examples/blog.rs @@ -0,0 +1,54 @@ +//! Run against a disposable database: creates and deletes a demonstration post. +use chusql_driver::{Client, Query}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let address = std::env::var("CHUSQL_ADDRESS").unwrap_or_else(|_| "127.0.0.1:7777".into()); + let db = Client::connect( + address, + &std::env::var("CHUSQL_USER")?, + &std::env::var("CHUSQL_PASSWORD")?, + ) + .await?; + // Use a dedicated session so USE and subsequent operations cannot interleave with other handlers. + let database = std::env::var("CHUSQL_DATABASE")?; + if database.is_empty() + || !database + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '_') + || database.starts_with(|c: char| c.is_ascii_digit()) + { + return Err("CHUSQL_DATABASE must be a plain identifier".into()); + } + db.query(&format!("USE {database}")).await?; + // Run once in a disposable database; table must not already exist. + db.query("CREATE TABLE driver_blog_posts (id INT PRIMARY KEY, title TEXT, body TEXT, published BOOL)").await?; + let insert = Query::new("INSERT INTO driver_blog_posts (id, title, body, published) VALUES (") + .bind(1_i64) + .push(", ") + .bind("ChuSQL 博客") + .push(", ") + .bind("第一篇文章：It's simple.\n正文支持换行。") + .push(", ") + .bind(true) + .push(")"); + db.query(insert.as_str()).await?; + let update = Query::new("UPDATE driver_blog_posts SET title = ") + .bind("更新后的标题") + .push(" WHERE id = ") + .bind(1_i64); + db.query(update.as_str()).await?; + let select = Query::new("SELECT id, title, body FROM driver_blog_posts WHERE published = ") + .bind(true) + .push(" ORDER BY id DESC LIMIT 10"); + let result = db.query(select.as_str()).await?; + assert!(!result.truncated); + assert_eq!(result.rows.len(), 1); + assert_eq!(result.rows[0][1].as_str(), Some("更新后的标题")); + println!("{:?}", result.rows); + let delete = Query::new("DELETE FROM driver_blog_posts WHERE id = ").bind(1_i64); + db.query(delete.as_str()).await?; + db.query("DROP TABLE driver_blog_posts").await?; + db.close().await; + Ok(()) +} diff --git a/examples/query.rs b/examples/query.rs new file mode 100644 index 0000000..7e74f62 --- /dev/null +++ b/examples/query.rs @@ -0,0 +1,16 @@ +use chusql_driver::Client; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let address = std::env::var("CHUSQL_ADDRESS").unwrap_or_else(|_| "127.0.0.1:7777".into()); + let user = std::env::var("CHUSQL_USER")?; + let password = std::env::var("CHUSQL_PASSWORD")?; + let sql = std::env::args() + .nth(1) + .unwrap_or_else(|| "SHOW DATABASES".into()); + let client = Client::connect(address, &user, &password).await?; + let result = client.query(&sql).await?; + println!("{:?}", result); + client.close().await; + Ok(()) +} diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..252aa67 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,50 @@ +use std::{fmt, io}; + +/// Errors preserve server codes so applications can distinguish authorization and SQL failures. +#[derive(Debug)] +#[non_exhaustive] +pub enum Error { + Io(io::Error), + Json(serde_json::Error), + Server { code: String, message: String }, + Protocol(&'static str), + Timeout, + Closed, + ResponseTooLarge, +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Io(error) => write!(f, "ChuSQL I/O: {error}"), + Self::Json(error) => write!(f, "ChuSQL JSON: {error}"), + Self::Server { code, message } => write!(f, "ChuSQL {code}: {message}"), + Self::Protocol(message) => write!(f, "ChuSQL protocol: {message}"), + Self::Timeout => f.write_str("ChuSQL request timed out"), + Self::Closed => f.write_str("ChuSQL connection is closed; reconnect before retrying"), + Self::ResponseTooLarge => f.write_str("ChuSQL response exceeds configured byte limit"), + } + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(error) => Some(error), + Self::Json(error) => Some(error), + _ => None, + } + } +} + +impl From for Error { + fn from(error: io::Error) -> Self { + Self::Io(error) + } +} + +impl From for Error { + fn from(error: serde_json::Error) -> Self { + Self::Json(error) + } +} diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..989f84a --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,184 @@ +//! An asynchronous client for ChuSQL's version 1 newline-delimited JSON protocol. +//! Clones share one authenticated session and serialize requests. +mod error; +mod query; + +pub use error::Error; +pub use query::{Param, Query}; +pub use serde_json::Value; + +use serde::Deserialize; +use serde_json::json; +use std::{sync::Arc, time::Duration}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + net::{TcpStream, ToSocketAddrs}, + sync::Mutex, + time::timeout, +}; + +/// Limits for connecting and exchanging each request (excluding time queued behind other requests). +#[derive(Debug, Clone)] +pub struct Options { + pub timeout: Duration, + pub max_response_bytes: usize, +} + +impl Default for Options { + fn default() -> Self { + Self { + timeout: Duration::from_secs(30), + max_response_bytes: 16 * 1024 * 1024, + } + } +} + +/// The original wire values are preserved, including NULL and runtime type objects. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct QueryResult { + pub columns: Vec, + pub rows: Vec>, + pub row_count: u64, + pub truncated: bool, + pub database: Option, +} + +#[derive(Deserialize)] +#[serde(tag = "status")] +enum Response { + #[serde(rename = "hello")] + Hello { protocol: u32 }, + #[serde(rename = "ok")] + Login {}, + #[serde(rename = "result")] + Result(QueryResult), + #[serde(rename = "pong")] + Pong, + #[serde(rename = "error")] + Error { code: String, message: String }, +} + +/// Cloneable backend application state. Each clone uses the same connection and login. +#[derive(Clone)] +pub struct Client { + connection: Arc>>>, + options: Options, +} + +impl Client { + /// Connect, negotiate protocol 1 and authenticate using the default limits. + pub async fn connect( + address: impl ToSocketAddrs, + user: &str, + password: &str, + ) -> Result { + Self::connect_with_options(address, user, password, Options::default()).await + } + + /// Connect with explicit limits. Passwords are sent over plain TCP by the server protocol. + pub async fn connect_with_options( + address: impl ToSocketAddrs, + user: &str, + password: &str, + options: Options, + ) -> Result { + let stream = timeout(options.timeout, TcpStream::connect(address)) + .await + .map_err(|_| Error::Timeout)??; + stream.set_nodelay(true)?; + let client = Self { + connection: Arc::new(Mutex::new(Some(BufReader::new(stream)))), + options, + }; + match client + .request(json!({"method": "hello", "protocol": 1})) + .await? + { + Response::Hello { protocol: 1 } => {} + _ => return Err(Error::Protocol("expected hello with protocol 1")), + } + match client + .request(json!({"method": "login", "user": user, "password": password})) + .await? + { + Response::Login {} => {} + _ => return Err(Error::Protocol("expected login acknowledgement")), + } + Ok(client) + } + + /// Execute one SQL statement. The server does not support bound parameters. + /// Never interpolate untrusted input into SQL. + pub async fn query(&self, sql: &str) -> Result { + match self.request(json!({"method": "query", "sql": sql})).await? { + Response::Result(result) => { + if result + .rows + .iter() + .any(|row| row.len() != result.columns.len()) + { + self.close().await; + return Err(Error::Protocol("row width differs from columns")); + } + Ok(result) + } + _ => { + self.close().await; + Err(Error::Protocol("expected query result")) + } + } + } + + /// Check whether the server is reachable on this connection. + pub async fn ping(&self) -> Result<(), Error> { + match self.request(json!({"method": "ping"})).await? { + Response::Pong => Ok(()), + _ => { + self.close().await; + Err(Error::Protocol("expected pong")) + } + } + } + + /// Close the shared connection for all clones; repeated calls are harmless. + pub async fn close(&self) { + self.connection.lock().await.take(); + } + + async fn request(&self, request: Value) -> Result { + let mut bytes = serde_json::to_vec(&request)?; + bytes.push(b'\n'); + let mut guard = self.connection.lock().await; + // Taking ownership makes cancellation discard the stream instead of reusing a stale response. + let mut connection = guard.take().ok_or(Error::Closed)?; + let response = timeout(self.options.timeout, async { + connection.get_mut().write_all(&bytes).await?; + let mut line = Vec::new(); + loop { + let chunk = connection.fill_buf().await?; + if chunk.is_empty() { + return Err(Error::Closed); + } + let end = chunk.iter().position(|byte| *byte == b'\n'); + let used = end.map_or(chunk.len(), |index| index + 1); + if used > self.options.max_response_bytes.saturating_sub(line.len()) { + return Err(Error::ResponseTooLarge); + } + line.extend_from_slice(&chunk[..used]); + connection.consume(used); + if end.is_some() { + break; + } + } + Ok::(serde_json::from_slice(&line)?) + }) + .await + .map_err(|_| Error::Timeout)??; + *guard = Some(connection); + match response { + Response::Error { code, message } => Err(Error::Server { code, message }), + other => Ok(other), + } + } +} diff --git a/src/query.rs b/src/query.rs new file mode 100644 index 0000000..ae82abf --- /dev/null +++ b/src/query.rs @@ -0,0 +1,99 @@ +/// A scalar SQL value, encoded for ChuSQL's literal syntax. +#[derive(Debug, Clone, Copy)] +pub enum Param<'a> { + Text(&'a str), + Int(i64), + Bool(bool), + Null, +} + +impl<'a> From<&'a str> for Param<'a> { + fn from(value: &'a str) -> Self { + Self::Text(value) + } +} + +impl From for Param<'_> { + fn from(value: i64) -> Self { + Self::Int(value) + } +} + +impl From for Param<'_> { + fn from(value: bool) -> Self { + Self::Bool(value) + } +} + +/// Builds SQL from trusted syntax and separately encoded values. +/// This is client-side literal encoding, not a server prepared statement. +/// Never pass user input to `new` or `push`; use `bind` for values. +#[derive(Debug, Clone)] +pub struct Query(String); + +impl Query { + /// Start with trusted SQL syntax. + pub fn new(sql: &str) -> Self { + Self(sql.to_owned()) + } + + /// Append trusted SQL syntax, including whitespace as needed. + pub fn push(mut self, sql: &str) -> Self { + self.0.push_str(sql); + self + } + + /// Append a scalar value. Strings escape quotes using ChuSQL's `''` rule. + /// Does not encode table/column names, SQL expressions, or LIKE patterns. + pub fn bind<'a>(mut self, value: impl Into>) -> Self { + match value.into() { + Param::Text(text) => { + self.0.push('\''); + for ch in text.chars() { + self.0.push(ch); + if ch == '\'' { + self.0.push('\''); + } + } + self.0.push('\''); + } + Param::Int(number) => self.0.push_str(&number.to_string()), + Param::Bool(value) => self.0.push_str(if value { "TRUE" } else { "FALSE" }), + Param::Null => self.0.push_str("NULL"), + } + self + } + + /// The encoded SQL; execute it using `client.query(query.as_str()).await`. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn user_text_stays_inside_one_literal() { + let query = Query::new("INSERT INTO posts VALUES (") + .bind("楚的博客'); DROP TABLE posts; --\n\\正文\0") + .push(", ") + .bind(i64::MIN) + .push(", ") + .bind(true) + .push(", ") + .bind(Param::Null) + .push(")"); + assert_eq!( + query.as_str(), + "INSERT INTO posts VALUES ('楚的博客''); DROP TABLE posts; --\n\\正文\0', -9223372036854775808, TRUE, NULL)" + ); + } + + #[test] + fn empty_and_consecutive_quotes_are_preserved() { + assert_eq!(Query::new("").bind("").as_str(), "''"); + assert_eq!(Query::new("").bind("''").as_str(), "''''''"); + } +} diff --git a/tests/protocol.rs b/tests/protocol.rs new file mode 100644 index 0000000..7fe8324 --- /dev/null +++ b/tests/protocol.rs @@ -0,0 +1,173 @@ +use chusql_driver::{Client, Error, Options}; +use serde_json::{Value, json}; +use std::time::Duration; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + net::TcpListener, +}; + +async fn server(script: Vec<(Value, &'static str)>) -> (String, tokio::task::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap().to_string(); + let task = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut stream = BufReader::new(stream); + for (expected, response) in script { + let mut line = String::new(); + stream.read_line(&mut line).await.unwrap(); + assert_eq!(serde_json::from_str::(&line).unwrap(), expected); + // Split writes also exercise TCP framing across packet boundaries. + for chunk in response.as_bytes().chunks(7) { + stream.get_mut().write_all(chunk).await.unwrap(); + } + } + }); + (address, task) +} + +fn handshake() -> Vec<(Value, &'static str)> { + vec![ + ( + json!({"method":"hello","protocol":1}), + "{\"status\":\"hello\",\"protocol\":1,\"server\":\"chusql-server\"}\n", + ), + ( + json!({"method":"login","user":"app","password":"秘密\npass"}), + "{\"status\":\"ok\",\"user\":\"app\",\"admin\":false}\n", + ), + ] +} + +#[tokio::test] +async fn authenticated_session_preserves_results_and_recovers_from_sql_errors() { + let mut script = handshake(); + script.extend([ + (json!({"method":"query","sql":"SELECT * FROM users"}), "{\"status\":\"result\",\"columns\":[\"id\",\"name\",\"active\",\"extra\"],\"rows\":[[9223372036854775808,\"小楚\",true,null]],\"rowCount\":1,\"truncated\":true,\"database\":\"app\"}\n"), + (json!({"method":"query","sql":"bad SQL"}), "{\"status\":\"error\",\"code\":\"syntax_error\",\"message\":\"invalid SQL\"}\n"), + (json!({"method":"ping"}), "{\"status\":\"pong\"}\n"), + ]); + let (address, task) = server(script).await; + let client = Client::connect(&address, "app", "秘密\npass") + .await + .unwrap(); + let result = client.query("SELECT * FROM users").await.unwrap(); + assert_eq!( + result.rows[0], + vec![ + json!(9223372036854775808_u64), + json!("小楚"), + json!(true), + Value::Null + ] + ); + assert!(result.truncated); + assert_eq!(result.row_count, 1); + assert_eq!(result.database.as_deref(), Some("app")); + assert!( + matches!(client.query("bad SQL").await, Err(Error::Server { code, .. }) if code == "syntax_error") + ); + client.clone().ping().await.unwrap(); + client.close().await; + assert!(matches!(client.ping().await, Err(Error::Closed))); + task.await.unwrap(); +} + +#[tokio::test] +async fn rejects_wrong_protocol_and_failed_login() { + let (address, task) = server(vec![( + json!({"method":"hello","protocol":1}), + "{\"status\":\"hello\",\"protocol\":2}\n", + )]) + .await; + assert!(matches!( + Client::connect(address, "app", "pw").await, + Err(Error::Protocol(_)) + )); + task.await.unwrap(); + let mut script = handshake(); + script[1].1 = + "{\"status\":\"error\",\"code\":\"unauthorized\",\"message\":\"invalid credentials\"}\n"; + let (address, task) = server(script).await; + assert!( + matches!(Client::connect(address, "app", "秘密\npass").await, Err(Error::Server { code, .. }) if code == "unauthorized") + ); + task.await.unwrap(); +} + +#[tokio::test] +async fn malformed_oversized_and_unterminated_responses_discard_connection() { + for response in [ + "not JSON\n", + "{\"status\":\"pong\"}", + "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\n", + ] { + let mut script = handshake(); + script.push((json!({"method":"ping"}), response)); + let (address, task) = server(script).await; + let client = Client::connect_with_options( + address, + "app", + "秘密\npass", + Options { + max_response_bytes: 128, + ..Options::default() + }, + ) + .await + .unwrap(); + assert!(client.ping().await.is_err()); + assert!(matches!(client.ping().await, Err(Error::Closed))); + task.await.unwrap(); + } +} + +#[tokio::test] +async fn timeout_and_cancellation_do_not_reuse_pending_responses() { + for cancel in [false, true] { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (sent, received) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut stream = BufReader::new(stream); + for (_, response) in handshake() { + let mut line = String::new(); + stream.read_line(&mut line).await.unwrap(); + stream + .get_mut() + .write_all(response.as_bytes()) + .await + .unwrap(); + } + let mut line = String::new(); + stream.read_line(&mut line).await.unwrap(); + sent.send(()).unwrap(); + line.clear(); + assert_eq!(stream.read_line(&mut line).await.unwrap(), 0); + }); + let client = Client::connect_with_options( + address, + "app", + "秘密\npass", + Options { + timeout: Duration::from_millis(500), + ..Options::default() + }, + ) + .await + .unwrap(); + let pending = tokio::spawn({ + let client = client.clone(); + async move { client.ping().await } + }); + received.await.unwrap(); + if cancel { + pending.abort(); + assert!(pending.await.unwrap_err().is_cancelled()); + } else { + assert!(matches!(pending.await.unwrap(), Err(Error::Timeout))); + } + assert!(matches!(client.ping().await, Err(Error::Closed))); + task.await.unwrap(); + } +}