diff --git a/Cargo.lock b/Cargo.lock index 1b35ad2d5..36926386e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5480,14 +5480,15 @@ dependencies = [ "blake2", "clap", "hex", + "logos-blockchain-codec", "logos-blockchain-groth16", "logos-blockchain-key-management-system-service", "logos-blockchain-log-targets", + "logos-blockchain-utils", "logos-blockchain-zone-sdk", "rand 0.8.6", "reqwest 0.12.28", "rusqlite", - "serde", "tempfile", "thiserror 2.0.18", "tokio", diff --git a/logos_sql/Cargo.toml b/logos_sql/Cargo.toml index 8e30fc245..5cd70a89b 100644 --- a/logos_sql/Cargo.toml +++ b/logos_sql/Cargo.toml @@ -16,13 +16,14 @@ workspace = true [dependencies] bincode = { workspace = true } blake2 = { workspace = true } +lb-codec = { workspace = true } lb-key-management-system-service = { workspace = true } lb-log-targets = { workspace = true } +lb-utils = { workspace = true } lb-zone-sdk = { workspace = true } rand = { features = ["getrandom"], workspace = true } reqwest = { workspace = true } rusqlite = { features = ["bundled", "functions", "hooks"], workspace = true } -serde = { features = ["derive"], workspace = true } thiserror = { workspace = true } tokio = { features = ["macros", "rt", "sync", "time"], workspace = true } tracing = { workspace = true } diff --git a/logos_sql/src/applier.rs b/logos_sql/src/applier.rs index 4ce0550fc..4496f3712 100644 --- a/logos_sql/src/applier.rs +++ b/logos_sql/src/applier.rs @@ -215,12 +215,13 @@ mod tests { SequencerCheckpoint, }, }; + use rusqlite::types::Value; use tempfile::TempDir; use super::on_event; use crate::{ db::Databases, - protocol::{ChannelInscription, EncodedWrite, Statement, Transaction, Value}, + protocol::{ChannelInscription, EncodedWrite, PAYLOAD_MARKER, Statement, Transaction}, }; const CHANNEL_ID: [u8; 32] = [9; 32]; @@ -518,7 +519,7 @@ mod tests { Vec::new(), )) .payload; - malformed.truncate(b"LOGOS_SQL".len()); + malformed.pop(); let create = encoded_write(&transaction( "CREATE TABLE items(value INTEGER NOT NULL)", @@ -614,8 +615,9 @@ mod tests { Vec::new(), )) .payload; - let version_offset = b"LOGOS_SQL".len(); - unsupported[version_offset..version_offset + 2].copy_from_slice(&2u16.to_le_bytes()); + let version_offset = PAYLOAD_MARKER.len(); + unsupported[version_offset..version_offset + size_of::()] + .copy_from_slice(&2u16.to_le_bytes()); let following = encoded_write(&transaction( "CREATE TABLE following_write(value INTEGER)", diff --git a/logos_sql/src/db.rs b/logos_sql/src/db.rs index edee6877a..bf6296da1 100644 --- a/logos_sql/src/db.rs +++ b/logos_sql/src/db.rs @@ -251,7 +251,7 @@ impl Databases { // the transaction published to other participants. apply_statements(&db_transaction, transaction)?; - let transaction_digest = transaction.digest()?; + let transaction_digest = transaction.digest(); db_transaction.execute( INSERT_APPLIED_WRITE, @@ -281,7 +281,7 @@ impl Databases { &mut self, write: &ChannelInscription, ) -> Result<(), Error> { - let transaction_digest = write.transaction.digest()?; + let transaction_digest = write.transaction.digest(); is_write_applied(&self.lib, write.tx_id, &transaction_digest)?; is_write_applied(&self.live, write.tx_id, &transaction_digest)?; @@ -363,7 +363,7 @@ fn apply_channel_write( connection: &mut Connection, write: &ChannelInscription, ) -> Result<(), Error> { - let transaction_digest = write.transaction.digest()?; + let transaction_digest = write.transaction.digest(); let db_transaction = connection.transaction()?; if is_write_applied(&db_transaction, write.tx_id, &transaction_digest)? { @@ -567,13 +567,14 @@ mod tests { node_types::{HeaderId, MsgId, Slot}, sequencer::SequencerCheckpoint, }; + use rusqlite::types::Value; use tempfile::TempDir; use super::Databases; use crate::{ error::Error, local_write, - protocol::{ChannelInscription, EncodedWrite, Statement, Transaction, TxId, Value}, + protocol::{ChannelInscription, EncodedWrite, Statement, Transaction, TxId}, }; fn checkpoint(byte: u8, slot: u64) -> SequencerCheckpoint { diff --git a/logos_sql/src/error.rs b/logos_sql/src/error.rs index 48c98d669..c093c0812 100644 --- a/logos_sql/src/error.rs +++ b/logos_sql/src/error.rs @@ -11,7 +11,7 @@ pub enum Error { #[error("database error: {0}")] Database(#[from] rusqlite::Error), - /// Local draft serialization failed. + /// The participant-local `ZoneSDK` checkpoint could not be encoded. #[error("encoding error: {0}")] Encoding(#[from] bincode::Error), diff --git a/logos_sql/src/local_write.rs b/logos_sql/src/local_write.rs index 94b26330d..c0ba191c9 100644 --- a/logos_sql/src/local_write.rs +++ b/logos_sql/src/local_write.rs @@ -2,8 +2,6 @@ //! //! Transactions arriving through channel history are handled by the applier. -use lb_zone_sdk::node_types::Inscription; - use crate::{ db::Databases, error::Error, @@ -13,9 +11,5 @@ use crate::{ pub fn commit(db: &mut Databases, transaction: &Transaction) -> Result { let encoded = EncodedWrite::new(transaction)?; - if encoded.payload.len() > Inscription::MAX { - return Err(Error::InscriptionTooLarge); - } - db.commit_local_write(transaction, &encoded) } diff --git a/logos_sql/src/protocol/codec.rs b/logos_sql/src/protocol/codec.rs new file mode 100644 index 000000000..dc289f80f --- /dev/null +++ b/logos_sql/src/protocol/codec.rs @@ -0,0 +1,134 @@ +//! Binary encoding for protocol leaves backed by external types. + +use lb_codec::{BinaryDecode, BinaryEncode, DecodeError}; +use lb_utils::bounded::UpperBoundedVec; +use rusqlite::types::Value; + +use super::{MAX_PAYLOAD_BYTES, SqlParameter, SqlText}; + +const NULL: u8 = 0; +const INTEGER: u8 = 1; +const REAL: u8 = 2; +const TEXT: u8 = 3; +const BLOB: u8 = 4; + +type BoundedBytes = UpperBoundedVec; + +// Every variable-length field uses the same fixed-width prefix. Keep the two +// leaf encoders in lockstep with the bounded collection decoder. +const _: () = assert!(MAX_PAYLOAD_BYTES > u16::MAX as usize); +const _: () = assert!(MAX_PAYLOAD_BYTES <= u32::MAX as usize); + +impl BinaryEncode for SqlText { + fn encoded_length(&self) -> usize { + size_of::() + self.as_str().len() + } + + fn encode_into(&self, out: &mut Vec) { + u32::try_from(self.as_str().len()) + .expect("validated SQL length fits in u32") + .encode_into(out); + out.extend_from_slice(self.as_str().as_bytes()); + } +} + +impl BinaryDecode for SqlText { + type Context = (); + + fn decode<'input>( + input: &'input [u8], + (): &Self::Context, + ) -> Result<(&'input [u8], Self), DecodeError> { + let (input, sql) = ::decode(input, &())?; + let sql = String::from_utf8(sql.into_inner()) + .map_err(|_| DecodeError::invalid_value::("statement SQL is not UTF-8"))?; + let sql = Self::new(sql) + .map_err(|_| DecodeError::invalid_value::("statement SQL is invalid"))?; + + Ok((input, sql)) + } +} + +impl BinaryEncode for SqlParameter { + fn encoded_length(&self) -> usize { + 1 + match &self.0 { + Value::Null => 0, + Value::Integer(_) | Value::Real(_) => size_of::(), + Value::Text(value) => size_of::() + value.len(), + Value::Blob(value) => size_of::() + value.len(), + } + } + + fn encode_into(&self, out: &mut Vec) { + match &self.0 { + Value::Null => NULL.encode_into(out), + Value::Integer(value) => { + INTEGER.encode_into(out); + u64::from_le_bytes(value.to_le_bytes()).encode_into(out); + } + Value::Real(value) => { + REAL.encode_into(out); + value.to_bits().encode_into(out); + } + Value::Text(value) => { + TEXT.encode_into(out); + u32::try_from(value.len()) + .expect("validated text length fits in u32") + .encode_into(out); + out.extend_from_slice(value.as_bytes()); + } + Value::Blob(value) => { + BLOB.encode_into(out); + u32::try_from(value.len()) + .expect("validated blob length fits in u32") + .encode_into(out); + out.extend_from_slice(value); + } + } + } +} + +impl BinaryDecode for SqlParameter { + type Context = (); + + fn decode<'input>( + input: &'input [u8], + (): &Self::Context, + ) -> Result<(&'input [u8], Self), DecodeError> { + let (input, tag) = ::decode(input, &())?; + + let (input, value) = match tag { + NULL => (input, Value::Null), + INTEGER => { + let (input, value) = ::decode(input, &())?; + let value = i64::from_le_bytes(value.to_le_bytes()); + + (input, Value::Integer(value)) + } + REAL => { + let (input, bits) = ::decode(input, &())?; + + (input, Value::Real(f64::from_bits(bits))) + } + TEXT => { + let (input, value) = ::decode(input, &())?; + let value = String::from_utf8(value.into_inner()).map_err(|_| { + DecodeError::invalid_value::("text parameter is not UTF-8") + })?; + + (input, Value::Text(value)) + } + BLOB => { + let (input, value) = ::decode(input, &())?; + + (input, Value::Blob(value.into_inner())) + } + _ => return Err(DecodeError::unknown_discriminant::(u64::from(tag))), + }; + + let value = Self::try_from(value) + .map_err(|_| DecodeError::invalid_value::("SQL parameter is invalid"))?; + + Ok((input, value)) + } +} diff --git a/logos_sql/src/protocol/fixtures.rs b/logos_sql/src/protocol/fixtures.rs new file mode 100644 index 000000000..711e06805 --- /dev/null +++ b/logos_sql/src/protocol/fixtures.rs @@ -0,0 +1,86 @@ +//! Well-known examples that pin the protocol's binary representation. + +use lb_codec::codec_fixtures; +use rusqlite::types::Value; + +use super::{ChannelInscription, SqlParameter, SqlText, Statement, Transaction, TxId}; + +codec_fixtures!( + TxId, + TxId::from([3; 32]) => + "0303030303030303030303030303030303030303030303030303030303030303" +); + +codec_fixtures!( + SqlText, + SqlText::new("SELECT 1".to_owned()).expect("fixture should be valid") => + "0800000053454c4543542031" +); + +codec_fixtures!( + SqlParameter, + SqlParameter::try_from(Value::Null).expect("fixture should be valid") => "00", + SqlParameter::try_from(Value::Integer(42)).expect("fixture should be valid") => + "012a00000000000000", + SqlParameter::try_from(Value::Real(1.5)).expect("fixture should be valid") => + "02000000000000f83f", + SqlParameter::try_from(Value::Text("hi".to_owned())).expect("fixture should be valid") => + "03020000006869", + SqlParameter::try_from(Value::Blob(vec![0, 255])).expect("fixture should be valid") => + "040200000000ff" +); + +fn statement_fixture() -> Statement { + Statement::new("SELECT 1".to_owned(), Vec::new()).expect("fixture should be valid") +} + +fn statement_with_values_fixture() -> Statement { + Statement::new( + "VALUES".to_owned(), + vec![ + Value::Null, + Value::Integer(42), + Value::Real(1.5), + Value::Text("hi".to_owned()), + Value::Blob(vec![0, 255]), + ], + ) + .expect("fixture should be valid") +} + +codec_fixtures!( + Statement, + statement_fixture() => "0800000053454c454354203100000000", + statement_with_values_fixture() => concat!( + "0600000056414c55455305000000", + "00", + "012a00000000000000", + "02000000000000f83f", + "03020000006869", + "040200000000ff" + ) +); + +fn transaction_fixture() -> Transaction { + Transaction::new(vec![statement_fixture()]).expect("fixture should be valid") +} + +codec_fixtures!( + Transaction, + transaction_fixture() => "010000000800000053454c454354203100000000" +); + +fn channel_inscription_fixture() -> ChannelInscription { + ChannelInscription { + tx_id: TxId::from([3; 32]), + transaction: transaction_fixture(), + } +} + +codec_fixtures!( + ChannelInscription, + channel_inscription_fixture() => concat!( + "0303030303030303030303030303030303030303030303030303030303030303", + "010000000800000053454c454354203100000000" + ) +); diff --git a/logos_sql/src/protocol/mod.rs b/logos_sql/src/protocol/mod.rs index 263841062..0f7dec08e 100644 --- a/logos_sql/src/protocol/mod.rs +++ b/logos_sql/src/protocol/mod.rs @@ -2,20 +2,25 @@ use std::fmt::{self, Display, Formatter}; -use bincode::Options as _; use blake2::{Blake2b, Digest as _, digest::consts::U32}; +use lb_codec::{BinaryCodec, BinaryDecode, BinaryEncode as _}; +use lb_utils::bounded::{NonEmptyBoundedVec, UpperBoundedVec}; +use lb_zone_sdk::node_types::Inscription; use rand::RngCore as _; -pub use rusqlite::types::Value; -use serde::{Deserialize, Serialize}; +use rusqlite::types::{ToSql, ToSqlOutput, Value}; use crate::error::Error; -const PAYLOAD_MARKER: [u8; 9] = *b"LOGOS_SQL"; +mod codec; +mod fixtures; + +pub const PAYLOAD_MARKER: [u8; 9] = *b"LOGOS_SQL"; const PAYLOAD_VERSION: u16 = 1; const PAYLOAD_HEADER_LEN: usize = PAYLOAD_MARKER.len() + size_of::(); +const MAX_PAYLOAD_BYTES: usize = Inscription::MAX; /// Stable identity of one application write. -#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)] +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, BinaryCodec)] pub struct TxId([u8; 32]); impl Display for TxId { @@ -55,112 +60,102 @@ impl From for [u8; 32] { } } -/// One parameterized SQL statement. -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] -pub struct Statement { - sql: String, - #[serde(with = "serialized_values")] - params: Vec, -} +/// Validated SQL text carried by one statement. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SqlText(String); -impl Statement { - /// Creates one non-empty statement. - /// - /// # Errors - /// - /// Returns an error if `sql` is empty. - pub fn new(sql: String, params: Vec) -> Result { +impl SqlText { + fn new(sql: String) -> Result { if sql.trim().is_empty() { return Err(Error::InvalidTransaction("statement SQL must not be empty")); } - Ok(Self { sql, params }) + if sql.len() > MAX_PAYLOAD_BYTES { + return Err(Error::InvalidTransaction("statement SQL is too large")); + } + + Ok(Self(sql)) } - /// Returns the SQL text. - #[must_use] - pub fn sql(&self) -> &str { - &self.sql - } - - /// Returns parameters in `SQLite` binding order. - #[must_use] - pub fn params(&self) -> &[Value] { - &self.params + fn as_str(&self) -> &str { + &self.0 } } -mod serialized_values { - use serde::{Deserialize, Deserializer, Serialize, Serializer}; +/// One `SQLite` parameter with `λSQL`'s protocol validation. +#[derive(Clone, Debug, PartialEq)] +pub struct SqlParameter(Value); - use super::Value; +impl TryFrom for SqlParameter { + type Error = Error; - /// Serializable representation of a `rusqlite` parameter value. - #[derive(Deserialize, Serialize)] - enum SqlValue { - Null, - Integer(i64), - Real(f64), - Text(String), - Blob(Vec), - } - - impl From<&Value> for SqlValue { - fn from(value: &Value) -> Self { - match value { - Value::Null => Self::Null, - Value::Integer(value) => Self::Integer(*value), - Value::Real(value) => Self::Real(*value), - Value::Text(value) => Self::Text(value.clone()), - Value::Blob(value) => Self::Blob(value.clone()), + fn try_from(value: Value) -> Result { + match &value { + Value::Real(value) if !value.is_finite() => { + return Err(Error::InvalidTransaction("real parameters must be finite")); } - } - } - - impl From for Value { - fn from(value: SqlValue) -> Self { - match value { - SqlValue::Null => Self::Null, - SqlValue::Integer(value) => Self::Integer(value), - SqlValue::Real(value) => Self::Real(value), - SqlValue::Text(value) => Self::Text(value), - SqlValue::Blob(value) => Self::Blob(value), + Value::Text(value) if value.len() > MAX_PAYLOAD_BYTES => { + return Err(Error::InvalidTransaction("text parameter is too large")); } + Value::Blob(value) if value.len() > MAX_PAYLOAD_BYTES => { + return Err(Error::InvalidTransaction("blob parameter is too large")); + } + _ => {} } + + Ok(Self(value)) + } +} + +impl ToSql for SqlParameter { + fn to_sql(&self) -> rusqlite::Result> { + self.0.to_sql() + } +} + +/// One parameterized SQL statement. +#[derive(Clone, Debug, PartialEq, BinaryCodec)] +pub struct Statement { + sql: SqlText, + params: UpperBoundedVec, +} + +impl Statement { + /// Creates one non-empty statement within the protocol limits. + pub fn new(sql: String, params: Vec) -> Result { + if params.len() > MAX_PAYLOAD_BYTES { + return Err(Error::InvalidTransaction( + "statement has too many parameters", + )); + } + + let sql = SqlText::new(sql)?; + let params = params + .into_iter() + .map(SqlParameter::try_from) + .collect::, _>>()?; + let params = UpperBoundedVec::new_unchecked(params); + + Ok(Self { sql, params }) } - pub fn serialize(values: &[Value], serializer: S) -> Result - where - S: Serializer, - { - values - .iter() - .map(SqlValue::from) - .collect::>() - .serialize(serializer) + pub fn sql(&self) -> &str { + self.sql.as_str() } - pub fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> - where - D: Deserializer<'de>, - { - Vec::::deserialize(deserializer) - .map(|values| values.into_iter().map(Value::from).collect()) + pub fn params(&self) -> &[SqlParameter] { + self.params.as_slice() } } /// Statements applied atomically at one channel position. -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[derive(Clone, Debug, PartialEq, BinaryCodec)] pub struct Transaction { - statements: Vec, + statements: NonEmptyBoundedVec, } impl Transaction { - /// Creates a non-empty transaction. - /// - /// # Errors - /// - /// Returns an error if no statements are provided. + /// Creates a non-empty transaction within the protocol limits. pub fn new(statements: Vec) -> Result { if statements.is_empty() { return Err(Error::InvalidTransaction( @@ -168,39 +163,50 @@ impl Transaction { )); } - Ok(Self { statements }) + if statements.len() > MAX_PAYLOAD_BYTES { + return Err(Error::InvalidTransaction( + "transaction contains too many statements", + )); + } + + Ok(Self { + statements: NonEmptyBoundedVec::new_unchecked(statements), + }) } - /// Returns statements in execution order. - #[must_use] pub fn statements(&self) -> &[Statement] { - &self.statements + self.statements.as_slice() } - pub(crate) fn digest(&self) -> Result<[u8; 32], Error> { - let encoded = codec().serialize(self)?; - - Ok(Blake2b::::digest(encoded).into()) + pub(crate) fn digest(&self) -> [u8; 32] { + Blake2b::::digest(self.encode_to_vec()).into() } } /// Transaction payload carried by a `λSQL` channel inscription. -#[derive(Serialize, Deserialize)] +#[derive(Clone, Debug, PartialEq, BinaryCodec)] pub struct ChannelInscription { pub tx_id: TxId, pub transaction: Transaction, } impl ChannelInscription { - pub(crate) fn encode(&self) -> Result, Error> { - let mut payload = Vec::from(PAYLOAD_MARKER); + pub fn encode(&self) -> Result, Error> { + let payload_len = payload_len(self.encoded_length())?; + let mut payload = Vec::with_capacity(payload_len); + + payload.extend_from_slice(&PAYLOAD_MARKER); payload.extend_from_slice(&PAYLOAD_VERSION.to_le_bytes()); - payload.extend(codec().serialize(self)?); + self.encode_into(&mut payload); Ok(payload) } pub fn decode(payload: &[u8]) -> Result { + if payload.len() > MAX_PAYLOAD_BYTES { + return Err(Error::InvalidPayload("payload exceeds the protocol limit")); + } + let (header, body) = payload .split_at_checked(PAYLOAD_HEADER_LEN) .ok_or(Error::InvalidPayload("header is missing"))?; @@ -216,8 +222,7 @@ impl ChannelInscription { return Err(Error::InvalidPayload("protocol version is not supported")); } - codec() - .deserialize(body) + ::decode_all(body, &()) .map_err(|_| Error::InvalidPayload("body cannot be decoded")) } } @@ -243,22 +248,31 @@ impl EncodedWrite { } } +fn payload_len(body_len: usize) -> Result { + let payload_len = PAYLOAD_HEADER_LEN + .checked_add(body_len) + .ok_or(Error::InscriptionTooLarge)?; + + if payload_len > MAX_PAYLOAD_BYTES { + return Err(Error::InscriptionTooLarge); + } + + Ok(payload_len) +} + /// Returns whether an inscription belongs to `λSQL`. #[must_use] pub fn is_logos_sql_payload(payload: &[u8]) -> bool { payload.starts_with(&PAYLOAD_MARKER) } -fn codec() -> impl bincode::Options { - bincode::DefaultOptions::new() - .with_little_endian() - .with_fixint_encoding() - .reject_trailing_bytes() -} - #[cfg(test)] mod tests { - use super::{ChannelInscription, EncodedWrite, Statement, Transaction, TxId, Value}; + use rusqlite::types::Value; + + use super::{ + ChannelInscription, EncodedWrite, MAX_PAYLOAD_BYTES, Statement, Transaction, TxId, + }; #[test] fn transaction_id_is_displayed_as_hex() { @@ -285,4 +299,56 @@ mod tests { assert_eq!(decoded.tx_id, encoded.tx_id); assert_eq!(decoded.transaction, transaction); } + + #[test] + fn payload_rejects_trailing_bytes() { + let transaction = Transaction::new(vec![ + Statement::new("SELECT 1".to_owned(), Vec::new()).expect("statement should be valid"), + ]) + .expect("transaction should be valid"); + let write = ChannelInscription { + tx_id: TxId::from([3; 32]), + transaction, + }; + let mut payload = write.encode().expect("payload should encode"); + payload.push(0); + + assert!(ChannelInscription::decode(&payload).is_err()); + } + + #[test] + fn payload_bytes_are_pinned() { + let transaction = Transaction::new(vec![ + Statement::new("SELECT 1".to_owned(), Vec::new()).expect("statement should be valid"), + ]) + .expect("transaction should be valid"); + let write = ChannelInscription { + tx_id: TxId::from([3; 32]), + transaction, + }; + + let expected = hex::decode(concat!( + "4c4f474f535f53514c0100", + "0303030303030303030303030303030303030303030303030303030303030303", + "010000000800000053454c454354203100000000" + )) + .expect("fixture should be valid hex"); + + assert_eq!(write.encode().expect("payload should encode"), expected); + } + + #[test] + fn complete_payload_must_fit_one_inscription() { + let transaction = Transaction::new(vec![ + Statement::new( + "SELECT ?1".to_owned(), + vec![Value::Blob(vec![0; MAX_PAYLOAD_BYTES])], + ) + .expect("statement should be valid"), + ]) + .expect("transaction should be valid"); + let result = EncodedWrite::new(&transaction); + + assert!(matches!(result, Err(crate::Error::InscriptionTooLarge))); + } }