From 8f84a9e38f1fa430d531915c504fa2739c9f4388 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 5 Aug 2025 14:49:58 +1000 Subject: [PATCH] prototype of kit --- Cargo.lock | 14 +++++++ Cargo.toml | 4 +- crates/increment/Cargo.toml | 1 + crates/increment/src/lib.rs | 40 +++++++------------ crates/increment/src/proof.rs | 46 +--------------------- crates/increment/src/sequence.rs | 9 ----- crates/increment/src/signing.rs | 37 ++---------------- crates/increment/src/state.rs | 11 +++++- crates/kit/Cargo.toml | 17 ++++++++ crates/kit/src/lib.rs | 3 ++ crates/kit/src/poa.rs | 37 ++++++++++++++++++ crates/kit/src/state.rs | 9 +++++ crates/kit/src/verify.rs | 66 ++++++++++++++++++++++++++++++++ 13 files changed, 178 insertions(+), 116 deletions(-) delete mode 100644 crates/increment/src/sequence.rs create mode 100644 crates/kit/Cargo.toml create mode 100644 crates/kit/src/lib.rs create mode 100644 crates/kit/src/poa.rs create mode 100644 crates/kit/src/state.rs create mode 100644 crates/kit/src/verify.rs diff --git a/Cargo.lock b/Cargo.lock index f2a4c6a..2f9a7fe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2231,6 +2231,7 @@ dependencies = [ "clap", "futures", "http", + "kit", "reqwest", "serde", "serde_json", @@ -2365,6 +2366,19 @@ dependencies = [ "sha3-asm", ] +[[package]] +name = "kit" +version = "0.1.0" +dependencies = [ + "alloy", + "anyhow", + "futures", + "reqwest", + "sigma-stream", + "sigma-types", + "tokio", +] + [[package]] name = "lazy_static" version = "1.5.0" diff --git a/Cargo.toml b/Cargo.toml index 91cdd17..fe7390b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,4 +24,6 @@ sigma-stream = { git = "ssh://git@github.com/essential-contributions/sigma-tools tempfile = "3.20.0" tokio = { version = "1.45.1", features = ["full"] } tower-http = { version = "0.6.6", features = ["cors"] } -tokio-util = "0.7.15" \ No newline at end of file +tokio-util = "0.7.15" + +kit = { path = "crates/kit", version = "0.1.0" } \ No newline at end of file diff --git a/crates/increment/Cargo.toml b/crates/increment/Cargo.toml index 82cc2d7..39d9e6a 100644 --- a/crates/increment/Cargo.toml +++ b/crates/increment/Cargo.toml @@ -14,6 +14,7 @@ axum.workspace = true clap.workspace = true futures.workspace = true http.workspace = true +kit.workspace = true reqwest.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/increment/src/lib.rs b/crates/increment/src/lib.rs index 503d14e..e5573dd 100644 --- a/crates/increment/src/lib.rs +++ b/crates/increment/src/lib.rs @@ -2,11 +2,11 @@ use std::collections::HashMap; use alloy::{primitives::Address, signers::local::PrivateKeySigner, sol}; use futures::StreamExt; -use sigma_stream::stream_events; -use sigma_types::{Event, Height, Signed}; +use sigma_types::{Height, Signed}; use tokio::sync::mpsc; use crate::{ + Increment::Incremented, server::{GetCounts, GetProof, Req}, signing::sign, }; @@ -21,7 +21,6 @@ sol!( ); pub mod proof; -pub mod sequence; pub mod server; pub mod signing; pub mod state; @@ -41,7 +40,7 @@ pub struct Channels { enum Message { Req(Req), - Event(Result, std::io::Error>), + Event(Result), } impl App { @@ -113,11 +112,11 @@ pub async fn run( ) -> anyhow::Result<()> { let mut app = App::new(signer)?; - let event_stream = stream_events(&sigma_url, 0) - .await - .unwrap() - .map(Message::Event) - .boxed(); + let event_stream = + kit::verify::stream_events(&sigma_url, 0, Height::MAX, sigma_poa, state::map_input) + .await? + .map(Message::Event) + .boxed(); let req_stream = futures::stream::unfold(channels.state_rx, |mut rx| async move { rx.recv().await.map(|req| (req, rx)) }) @@ -127,23 +126,12 @@ pub async fn run( .for_each(|msg| { match msg { Message::Event(result) => match result { - Ok(signed_event) => { - match proof::update( - &mut app.state, - &mut app.last_height, - sigma_poa, - signed_event, - ) { - Ok(_) => { - update_app(&mut app); - println!("New count: {}", app.state.count); - if channels.current_count.send(app.state.count).is_err() { - eprintln!("Failed to send state update"); - } - } - Err(e) => { - eprintln!("Failed to update state: {e}"); - } + Ok(input) => { + state::update(&mut app.state, input); + update_app(&mut app); + println!("New count: {}", app.state.count); + if channels.current_count.send(app.state.count).is_err() { + eprintln!("Failed to send state update"); } } Err(e) => { diff --git a/crates/increment/src/proof.rs b/crates/increment/src/proof.rs index 7b638b6..d4fac2a 100644 --- a/crates/increment/src/proof.rs +++ b/crates/increment/src/proof.rs @@ -1,18 +1,9 @@ //! The proof module contains all the logic that is required to //! be verified before signing a proof of new state. use alloy::{ - primitives::{Address, Log}, - sol_types::SolEvent, + primitives::{Address}, }; use serde::{Deserialize, Serialize}; -use sigma_types::{Event, Height, Signed}; - -use crate::{ - Increment::Incremented, - sequence, - signing::verify_signature, - state::{self, State}, -}; /// This is the proof that is signed by the apps POA /// and sent to the underlying chains contract @@ -23,38 +14,3 @@ pub struct Proof { /// The sender address that resulted in this count pub sender: Address, } - -/// This is the full function that must be run before -/// a proof of the new state can be signed. -pub fn update( - pre_state: &mut State, - last_height: &mut Height, - sigma_poa: &Address, - signed_event: Signed, -) -> anyhow::Result<()> { - // Verify the sigma POA signature of the event - let event = verify_signature(signed_event, sigma_poa)?; - - // Check the height is sequential - if !sequence::check(last_height, &event.height) { - return Err(anyhow::anyhow!( - "Invalid height: expected {}, got {}", - last_height.saturating_add(1), - event.height - )); - } - - // Update the last height - *last_height = event.height; - - // Decode the log - let log: Log = serde_json::from_slice(&event.event_data.data)?; - let event = Incremented::decode_log(&log) - .map_err(|_| anyhow::anyhow!("Failed to decode log"))? - .data; - - // Once all the above are verified, we can update the state - state::update(pre_state, event); - - Ok(()) -} diff --git a/crates/increment/src/sequence.rs b/crates/increment/src/sequence.rs deleted file mode 100644 index 28814c7..0000000 --- a/crates/increment/src/sequence.rs +++ /dev/null @@ -1,9 +0,0 @@ -//! Simple module to check that the height is sequential. -use sigma_types::Height; - -/// Height must either be the next sequential height or -/// the first height where the last will be set to [`Height::MAX`]. -pub fn check(last_height: &Height, new_height: &Height) -> bool { - *new_height == last_height.saturating_add(1) - || (*last_height == Height::MAX && *new_height == Height::MIN) -} diff --git a/crates/increment/src/signing.rs b/crates/increment/src/signing.rs index bee0263..8311a9b 100644 --- a/crates/increment/src/signing.rs +++ b/crates/increment/src/signing.rs @@ -1,44 +1,13 @@ use alloy::{ hex::FromHex, - primitives::{Address, FixedBytes, keccak256}, - signers::{Signature, SignerSync, local::PrivateKeySigner}, + primitives::{FixedBytes, keccak256}, + signers::{SignerSync, local::PrivateKeySigner}, sol_types::SolValue, }; -use sigma_types::{Event, Signed}; +use sigma_types::Signed; use crate::proof::Proof; -/// Verify the signature of a signed event against the expected POA address. -pub fn verify_signature(signed_event: Signed, sigma_poa: &Address) -> anyhow::Result { - // Decode the signature - let Ok(sig) = Signature::try_from(&signed_event.signature[..]) else { - return Err(anyhow::anyhow!("Invalid signature")); - }; - - // Encode in the same way as the Solidity contract - let encoded = ( - signed_event.data.height, - signed_event.data.event_data.stream_type as u64, - &signed_event.data.event_data.data, - ) - .abi_encode_packed(); - - // Hash the encoded data - let h = keccak256(encoded); - - // Recover the address from the signature - let addr = sig.recover_address_from_prehash(&h)?; - - // Check if the recovered address matches the expected POA address - if addr == *sigma_poa { - Ok(signed_event.data) - } else { - Err(anyhow::anyhow!( - "Signature does not match the expected POA address" - )) - } -} - /// Get a signer from an environment variable or generate a random one. pub fn get_signer(key: Option) -> anyhow::Result { match key { diff --git a/crates/increment/src/state.rs b/crates/increment/src/state.rs index c02fc1f..37136f1 100644 --- a/crates/increment/src/state.rs +++ b/crates/increment/src/state.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; -use alloy::primitives::Address; +use alloy::{primitives::Address, sol_types::SolEvent}; +use sigma_types::Event; use crate::Increment::Incremented; @@ -21,3 +22,11 @@ pub fn update(state: &mut State, event: Incremented) { // Insert the sender address into the owners map state.owners.insert(state.count, event.sender); } + +pub fn map_input(event: Event) -> anyhow::Result { + // Decode the event data into an Incremented struct + let log: alloy::primitives::Log = serde_json::from_slice(&event.event_data.data)?; + Incremented::decode_log(&log) + .map_err(|_| anyhow::anyhow!("Failed to decode log")) + .map(|decoded| decoded.data) +} \ No newline at end of file diff --git a/crates/kit/Cargo.toml b/crates/kit/Cargo.toml new file mode 100644 index 0000000..116d7f3 --- /dev/null +++ b/crates/kit/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "kit" +version = "0.1.0" +edition.workspace = true +authors.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +alloy.workspace = true +anyhow.workspace = true +futures.workspace = true +reqwest.workspace = true +sigma-stream.workspace = true +sigma-types.workspace = true +tokio.workspace = true \ No newline at end of file diff --git a/crates/kit/src/lib.rs b/crates/kit/src/lib.rs new file mode 100644 index 0000000..f556340 --- /dev/null +++ b/crates/kit/src/lib.rs @@ -0,0 +1,3 @@ +pub mod poa; +pub mod verify; +pub mod state; \ No newline at end of file diff --git a/crates/kit/src/poa.rs b/crates/kit/src/poa.rs new file mode 100644 index 0000000..472ec75 --- /dev/null +++ b/crates/kit/src/poa.rs @@ -0,0 +1,37 @@ +use alloy::{ + primitives::{Address, keccak256}, + signers::Signature, + sol_types::SolValue, +}; +use sigma_types::{Event, Signed}; + +/// Verify the signature of a signed event against the expected POA address. +pub fn verify_signature(signed_event: Signed, sigma_poa: &Address) -> anyhow::Result { + // Decode the signature + let Ok(sig) = Signature::try_from(&signed_event.signature[..]) else { + return Err(anyhow::anyhow!("Invalid signature")); + }; + + // Encode in the same way as the Solidity contract + let encoded = ( + signed_event.data.height, + signed_event.data.event_data.stream_type as u64, + &signed_event.data.event_data.data, + ) + .abi_encode_packed(); + + // Hash the encoded data + let h = keccak256(encoded); + + // Recover the address from the signature + let addr = sig.recover_address_from_prehash(&h)?; + + // Check if the recovered address matches the expected POA address + if addr == *sigma_poa { + Ok(signed_event.data) + } else { + Err(anyhow::anyhow!( + "Signature does not match the expected POA address" + )) + } +} diff --git a/crates/kit/src/state.rs b/crates/kit/src/state.rs new file mode 100644 index 0000000..2296e2a --- /dev/null +++ b/crates/kit/src/state.rs @@ -0,0 +1,9 @@ +use sigma_types::Event; + +pub trait StateUpdate { + type Input; + + fn update(&mut self, input: Self::Input); + + fn map_input(event: Event) -> anyhow::Result; +} diff --git a/crates/kit/src/verify.rs b/crates/kit/src/verify.rs new file mode 100644 index 0000000..0096898 --- /dev/null +++ b/crates/kit/src/verify.rs @@ -0,0 +1,66 @@ +use alloy::primitives::Address; +use futures::{StreamExt, TryStreamExt}; +use sigma_types::{Event, Height}; + +use crate::{poa::verify_signature, state::StateUpdate}; + +pub async fn stream_events( + sigma_url: &reqwest::Url, + from_height: usize, + mut last_height: Height, + sigma_poa: &Address, + map_input: F, +) -> anyhow::Result>> +where + F: Fn(Event) -> anyhow::Result, +{ + let s = sigma_stream::stream_events(sigma_url, from_height) + .await? + .map_err(|e| anyhow::anyhow!(e)) + .and_then(async |signed_event| verify_signature(signed_event, sigma_poa)) + .and_then(move |event| { + let height = event.height; + let r = map_input(event).map(|input| (height, input)); + std::future::ready(r) + }) + .and_then(move |(height, input)| { + let r = if check_sequence(&last_height, &height) { + last_height = height; + Ok(input) + } else { + Err(anyhow::anyhow!( + "Invalid height: expected {}, got {}", + last_height.saturating_add(1), + height + )) + }; + std::future::ready(r) + }); + Ok(s) +} + +/// Height must either be the next sequential height or +/// the first height where the last will be set to [`Height::MAX`]. +pub fn check_sequence(last_height: &Height, new_height: &Height) -> bool { + *new_height == last_height.saturating_add(1) + || (*last_height == Height::MAX && *new_height == Height::MIN) +} + +pub async fn run( + sigma_url: &reqwest::Url, + from_height: usize, + last_height: Height, + sigma_poa: &Address, + mut state: S, +) -> anyhow::Result<()> { + stream_events(sigma_url, from_height, last_height, sigma_poa, S::map_input) + .await? + .filter_map(|r| std::future::ready(r.ok())) + .for_each(|input| { + state.update(input); + std::future::ready(()) + }) + .await; + + Ok(()) +}