From 12d278d719c4443ad4768bfea9d3dec0ff09c56e Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 25 Feb 2026 10:37:36 -0500 Subject: [PATCH] feat: add ember-client async RESP3 client library standalone async client crate for connecting to an ember server over TCP or TLS, sending RESP3 commands, and reading responses. provides a clean public API (Client, ClientError, Frame) without coupling to CLI internals. TLS support is an optional feature (enabled by default) so embedded or server-side users can opt out of the rustls dependency. --- Cargo.lock | 29 ++- Cargo.toml | 1 + crates/ember-client/Cargo.toml | 32 ++++ crates/ember-client/src/connection.rs | 265 ++++++++++++++++++++++++++ crates/ember-client/src/lib.rs | 25 +++ crates/ember-client/src/tls.rs | 177 +++++++++++++++++ 6 files changed, 522 insertions(+), 7 deletions(-) create mode 100644 crates/ember-client/Cargo.toml create mode 100644 crates/ember-client/src/connection.rs create mode 100644 crates/ember-client/src/lib.rs create mode 100644 crates/ember-client/src/tls.rs diff --git a/Cargo.lock b/Cargo.lock index 891a0736..ace67d3a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -812,9 +812,24 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +[[package]] +name = "ember-client" +version = "0.4.8" +dependencies = [ + "bytes", + "ember-protocol", + "rustls", + "rustls-native-certs", + "rustls-pki-types", + "thiserror 2.0.18", + "tokio", + "tokio-rustls", + "tracing", +] + [[package]] name = "ember-cluster" -version = "0.4.7" +version = "0.4.8" dependencies = [ "bincode", "bytes", @@ -835,7 +850,7 @@ dependencies = [ [[package]] name = "ember-integration-tests" -version = "0.4.7" +version = "0.4.8" dependencies = [ "bytes", "ember-protocol", @@ -847,7 +862,7 @@ dependencies = [ [[package]] name = "ember-persistence" -version = "0.4.7" +version = "0.4.8" dependencies = [ "aes-gcm", "bytes", @@ -860,7 +875,7 @@ dependencies = [ [[package]] name = "ember-protocol" -version = "0.4.7" +version = "0.4.8" dependencies = [ "bytes", "criterion", @@ -871,7 +886,7 @@ dependencies = [ [[package]] name = "ember-server" -version = "0.4.7" +version = "0.4.8" dependencies = [ "bytes", "clap", @@ -911,7 +926,7 @@ dependencies = [ [[package]] name = "emberkv-cli" -version = "0.4.7" +version = "0.4.8" dependencies = [ "bytes", "clap", @@ -930,7 +945,7 @@ dependencies = [ [[package]] name = "emberkv-core" -version = "0.4.7" +version = "0.4.8" dependencies = [ "ahash 0.8.12", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 7da91153..9b7e7c3d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,7 @@ members = [ "crates/ember-persistence", "crates/ember-cluster", "crates/ember-cli", + "crates/ember-client", "tests/integration", ] resolver = "2" diff --git a/crates/ember-client/Cargo.toml b/crates/ember-client/Cargo.toml new file mode 100644 index 00000000..17ae56ce --- /dev/null +++ b/crates/ember-client/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "ember-client" +description = "async RESP3 client library for ember" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +keywords.workspace = true +categories.workspace = true +readme = "README.md" + +[dependencies] +bytes = { workspace = true } +tokio = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +ember-protocol = { workspace = true } + +# TLS support (optional feature) +tokio-rustls = { workspace = true, optional = true } +rustls = { workspace = true, optional = true } +rustls-pki-types = { workspace = true, optional = true } +rustls-native-certs = { workspace = true, optional = true } + +[features] +default = ["tls"] +tls = [ + "dep:tokio-rustls", + "dep:rustls", + "dep:rustls-pki-types", + "dep:rustls-native-certs", +] diff --git a/crates/ember-client/src/connection.rs b/crates/ember-client/src/connection.rs new file mode 100644 index 00000000..712425e1 --- /dev/null +++ b/crates/ember-client/src/connection.rs @@ -0,0 +1,265 @@ +//! Async client connection to an ember server. +//! +//! Handles connecting, sending commands as RESP3 arrays, and reading back +//! parsed frames. Works transparently over plain TCP or TLS. + +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::Duration; + +use bytes::BytesMut; +use ember_protocol::parse::parse_frame; +use ember_protocol::types::Frame; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}; +use tokio::net::TcpStream; + +#[cfg(feature = "tls")] +use crate::tls::TlsClientConfig; + +/// Maximum read buffer size (64 KiB). Prevents unbounded memory growth if the +/// server sends a response that never completes. +const MAX_READ_BUF: usize = 64 * 1024; + +/// Default timeout for establishing the TCP connection. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); + +/// Default timeout for reading a response from the server. +const READ_TIMEOUT: Duration = Duration::from_secs(10); + +/// Errors that can occur during client operations. +#[derive(Debug, thiserror::Error)] +pub enum ClientError { + #[error("connection failed: {0}")] + Io(#[from] io::Error), + + #[error("protocol error: {0}")] + Protocol(String), + + #[error("server disconnected")] + Disconnected, + + #[error("authentication failed: {0}")] + AuthFailed(String), + + #[error("connection timed out")] + Timeout, + + #[error("response too large (exceeded {MAX_READ_BUF} bytes)")] + ResponseTooLarge, +} + +/// Underlying transport — plain TCP or (optionally) TLS. +/// +/// Centralising the dispatch here keeps the `Client` logic clean regardless +/// of which features are compiled in. +pub(crate) enum Transport { + Tcp(TcpStream), + #[cfg(feature = "tls")] + Tls(Box>), +} + +impl AsyncRead for Transport { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + match self.get_mut() { + Transport::Tcp(s) => Pin::new(s).poll_read(cx, buf), + #[cfg(feature = "tls")] + Transport::Tls(s) => Pin::new(s.as_mut()).poll_read(cx, buf), + } + } +} + +impl AsyncWrite for Transport { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + match self.get_mut() { + Transport::Tcp(s) => Pin::new(s).poll_write(cx, buf), + #[cfg(feature = "tls")] + Transport::Tls(s) => Pin::new(s.as_mut()).poll_write(cx, buf), + } + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.get_mut() { + Transport::Tcp(s) => Pin::new(s).poll_flush(cx), + #[cfg(feature = "tls")] + Transport::Tls(s) => Pin::new(s.as_mut()).poll_flush(cx), + } + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.get_mut() { + Transport::Tcp(s) => Pin::new(s).poll_shutdown(cx), + #[cfg(feature = "tls")] + Transport::Tls(s) => Pin::new(s.as_mut()).poll_shutdown(cx), + } + } +} + +/// An async client connected to a single ember server. +/// +/// Buffers reads and writes internally. Not thread-safe — use one `Client` +/// per task, or wrap in an `Arc>` if sharing is needed. +pub struct Client { + transport: Transport, + read_buf: BytesMut, + write_buf: BytesMut, +} + +impl Client { + /// Connects to an ember server over plain TCP. + /// + /// Times out after 5 seconds if the server is unreachable. + pub async fn connect(host: &str, port: u16) -> Result { + let tcp = tokio::time::timeout( + CONNECT_TIMEOUT, + TcpStream::connect((host, port)), + ) + .await + .map_err(|_| ClientError::Timeout)? + .map_err(ClientError::Io)?; + + Ok(Self::from_transport(Transport::Tcp(tcp))) + } + + /// Connects to an ember server with TLS. + /// + /// Performs the TCP connection and TLS handshake within the 5-second + /// connect timeout. + #[cfg(feature = "tls")] + pub async fn connect_tls( + host: &str, + port: u16, + tls: &TlsClientConfig, + ) -> Result { + let stream = tokio::time::timeout( + CONNECT_TIMEOUT, + crate::tls::connect(host, port, tls), + ) + .await + .map_err(|_| ClientError::Timeout)? + .map_err(ClientError::Io)?; + + Ok(Self::from_transport(stream)) + } + + fn from_transport(transport: Transport) -> Self { + Self { + transport, + read_buf: BytesMut::with_capacity(4096), + write_buf: BytesMut::with_capacity(4096), + } + } + + /// Sends a command and returns the server's RESP3 response. + /// + /// Arguments are serialized as a RESP3 array of bulk strings, which is + /// the standard client-to-server wire format. + /// + /// # Example + /// + /// ```no_run + /// # use ember_client::Client; + /// # async fn example() -> Result<(), ember_client::ClientError> { + /// let mut client = Client::connect("127.0.0.1", 6379).await?; + /// let pong = client.send(&["PING"]).await?; + /// let value = client.send(&["GET", "mykey"]).await?; + /// # Ok(()) + /// # } + /// ``` + pub async fn send(&mut self, args: &[&str]) -> Result { + let parts: Vec = args + .iter() + .map(|t| Frame::Bulk(bytes::Bytes::from(t.to_string()))) + .collect(); + let frame = Frame::Array(parts); + + self.write_buf.clear(); + frame.serialize(&mut self.write_buf); + self.transport.write_all(&self.write_buf).await?; + self.transport.flush().await?; + + self.read_response().await + } + + /// Authenticates with the server using the `AUTH` command. + /// + /// Returns `Ok(())` on success, or `ClientError::AuthFailed` if the server + /// rejects the password. + pub async fn auth(&mut self, password: &str) -> Result<(), ClientError> { + let frame = Frame::Array(vec![ + Frame::Bulk(bytes::Bytes::from_static(b"AUTH")), + Frame::Bulk(bytes::Bytes::from(password.to_string())), + ]); + + self.write_buf.clear(); + frame.serialize(&mut self.write_buf); + self.transport.write_all(&self.write_buf).await?; + self.transport.flush().await?; + + match self.read_response().await? { + Frame::Simple(s) if s == "OK" => Ok(()), + Frame::Error(e) => Err(ClientError::AuthFailed(e)), + _ => Err(ClientError::AuthFailed( + "unexpected response to AUTH".into(), + )), + } + } + + /// Gracefully disconnects from the server. + /// + /// Sends `QUIT` so the server can clean up, then shuts down the transport. + /// Errors are ignored — this is best-effort cleanup. + pub async fn disconnect(&mut self) { + let quit = Frame::Array(vec![Frame::Bulk(bytes::Bytes::from_static(b"QUIT"))]); + self.write_buf.clear(); + quit.serialize(&mut self.write_buf); + let _ = self.transport.write_all(&self.write_buf).await; + let _ = self.transport.flush().await; + let _ = self.transport.shutdown().await; + } + + /// Reads a complete RESP3 frame from the server. + async fn read_response(&mut self) -> Result { + loop { + if !self.read_buf.is_empty() { + match parse_frame(&self.read_buf) { + Ok(Some((frame, consumed))) => { + let _ = self.read_buf.split_to(consumed); + return Ok(frame); + } + Ok(None) => { + // incomplete frame — need more data + } + Err(e) => { + return Err(ClientError::Protocol(e.to_string())); + } + } + } + + if self.read_buf.len() >= MAX_READ_BUF { + return Err(ClientError::ResponseTooLarge); + } + + let read_result = tokio::time::timeout( + READ_TIMEOUT, + self.transport.read_buf(&mut self.read_buf), + ) + .await; + + match read_result { + Ok(Ok(0)) => return Err(ClientError::Disconnected), + Ok(Ok(_)) => {} // got data, loop back to try parsing + Ok(Err(e)) => return Err(ClientError::Io(e)), + Err(_) => return Err(ClientError::Timeout), + } + } + } +} diff --git a/crates/ember-client/src/lib.rs b/crates/ember-client/src/lib.rs new file mode 100644 index 00000000..2081eedc --- /dev/null +++ b/crates/ember-client/src/lib.rs @@ -0,0 +1,25 @@ +//! ember-client: async RESP3 client for ember. +//! +//! Provides a simple async client for connecting to an ember server over +//! TCP (or TLS), sending commands as RESP3 frames, and reading responses. +//! +//! # Example +//! +//! ```no_run +//! use ember_client::Client; +//! +//! #[tokio::main] +//! async fn main() -> Result<(), ember_client::ClientError> { +//! let mut client = Client::connect("127.0.0.1", 6379).await?; +//! let response = client.send(&["PING"]).await?; +//! println!("{response:?}"); +//! Ok(()) +//! } +//! ``` + +mod connection; +#[cfg(feature = "tls")] +pub mod tls; + +pub use connection::{Client, ClientError}; +pub use ember_protocol::types::Frame; diff --git a/crates/ember-client/src/tls.rs b/crates/ember-client/src/tls.rs new file mode 100644 index 00000000..af899458 --- /dev/null +++ b/crates/ember-client/src/tls.rs @@ -0,0 +1,177 @@ +//! TLS configuration and connection helpers for ember-client. +//! +//! All items in this module are only compiled when the `tls` feature is +//! enabled (which is the default). Downstream code that disables the feature +//! can use `Client::connect` for plain TCP without pulling in rustls. + +use std::io; +use std::sync::Arc; + +use rustls::pki_types::pem::PemObject; +use rustls::pki_types::ServerName; +use tokio::net::TcpStream; +use tokio_rustls::TlsConnector; + +use crate::connection::Transport; + +/// TLS configuration for client connections. +/// +/// Pass this to `Client::connect_tls` to establish a TLS-encrypted connection +/// to the server. +#[derive(Clone, Debug, Default)] +pub struct TlsClientConfig { + /// Path to a custom CA certificate (PEM format) for verifying the server. + /// + /// When `None`, the platform's native root certificate store is used. + pub ca_cert: Option, + + /// Skip server certificate verification entirely. + /// + /// Emits a warning to stderr when set. Only use in development with + /// self-signed certificates — never in production. + pub insecure: bool, +} + +/// Establishes a TCP connection and upgrades it to TLS. +/// +/// Used by `Client::connect_tls`; not typically called directly. +pub(crate) async fn connect( + host: &str, + port: u16, + config: &TlsClientConfig, +) -> io::Result { + let tcp = TcpStream::connect((host, port)).await?; + let client_config = build_client_config(config)?; + let connector = TlsConnector::from(Arc::new(client_config)); + + let server_name = ServerName::try_from(host.to_string()).map_err(|e| { + io::Error::new( + io::ErrorKind::InvalidInput, + format!("invalid server name '{host}': {e}"), + ) + })?; + + let tls_stream = connector.connect(server_name, tcp).await?; + Ok(Transport::Tls(Box::new(tls_stream))) +} + +/// Builds a `rustls::ClientConfig` from the provided options. +fn build_client_config(config: &TlsClientConfig) -> io::Result { + if config.insecure { + eprintln!("warning: TLS certificate verification is disabled"); + return build_insecure_config(); + } + + let roots = load_root_certs(config.ca_cert.as_deref())?; + + rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth() + .pipe_ok() +} + +/// Loads root certificates from a custom CA file or the system trust store. +fn load_root_certs(ca_cert_path: Option<&str>) -> io::Result { + let mut roots = rustls::RootCertStore::empty(); + + if let Some(path) = ca_cert_path { + let pem = std::fs::read(path).map_err(|e| { + io::Error::new( + io::ErrorKind::NotFound, + format!("failed to read CA cert '{path}': {e}"), + ) + })?; + let certs = rustls::pki_types::CertificateDer::pem_slice_iter(&pem) + .collect::, _>>() + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; + + if certs.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("no certificates found in '{path}'"), + )); + } + + for cert in certs { + roots.add(cert).map_err(|e| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid CA certificate: {e}"), + ) + })?; + } + } else { + let native_certs = rustls_native_certs::load_native_certs(); + for cert in native_certs.certs { + roots.add(cert).map_err(|e| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid native CA certificate: {e}"), + ) + })?; + } + } + + Ok(roots) +} + +/// Builds a client config that skips all certificate verification. +/// +/// Only safe for development/testing with self-signed certificates. +fn build_insecure_config() -> io::Result { + let config = rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(NoVerifier)) + .with_no_client_auth(); + Ok(config) +} + +/// A certificate verifier that accepts anything. Used with `insecure: true`. +#[derive(Debug)] +struct NoVerifier; + +impl rustls::client::danger::ServerCertVerifier for NoVerifier { + fn verify_server_cert( + &self, + _end_entity: &rustls::pki_types::CertificateDer<'_>, + _intermediates: &[rustls::pki_types::CertificateDer<'_>], + _server_name: &ServerName<'_>, + _ocsp_response: &[u8], + _now: rustls::pki_types::UnixTime, + ) -> Result { + Ok(rustls::client::danger::ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _message: &[u8], + _cert: &rustls::pki_types::CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn verify_tls13_signature( + &self, + _message: &[u8], + _cert: &rustls::pki_types::CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn supported_verify_schemes(&self) -> Vec { + rustls::crypto::aws_lc_rs::default_provider() + .signature_verification_algorithms + .supported_schemes() + } +} + +/// Convenience helper to wrap a value in `Ok`. +trait PipeOk: Sized { + fn pipe_ok(self) -> io::Result { + Ok(self) + } +} + +impl PipeOk for rustls::ClientConfig {}