Merge branch 'feataggmode/add-setters-and-subscriptions' into feataggmode/add-subscription-limit-and-arbitrary-expiracy-lists

Author: maximopalopoli

Makefile

@@ -330,14 +330,10 @@ agg_mode_batcher_send_payment:
 		--private-key 0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d
 
 agg_mode_batcher_send_sp1_proof:
-	@NONCE=$$(curl -s http://127.0.0.1:8089/nonce/0x70997970C51812dc3A010C7d01b50e0d17dc79C8 | jq -r '.data.nonce'); \
-	curl -X POST \
-		-H "Content-Type: multipart/form-data" \
-		-F "nonce=$${NONCE}" \
-		-F "proof=@scripts/test_files/sp1/sp1_fibonacci_5_0_0.proof" \
-		-F "program_vk=@scripts/test_files/sp1/sp1_fibonacci_5_0_0_vk.bin" \
-		-F "_signature_hex=0x0" \
-		http://127.0.0.1:8089/proof/sp1
+	@cargo run --manifest-path aggregation_mode/cli/Cargo.toml -- submit sp1 \
+		--proof scripts/test_files/sp1/sp1_fibonacci_5_0_0.proof \
+		--vk scripts/test_files/sp1/sp1_fibonacci_5_0_0_vk.bin \
+		--private-key "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"
 
 __AGGREGATOR__: ## ____
 

aggregation_mode/Cargo.lock

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["./batcher", "./proof_aggregator", "./db"]
+members = ["./batcher", "./proof_aggregator", "./db", "./sdk", "./cli"]
 
 [workspace.package]
 version = "0.1.0"
@@ -13,6 +13,7 @@ serde_yaml = "0.9"
 alloy = { version = "1.1.1", features = ["default", "signer-keystore", "kzg"] }
 bincode = "1.3.3"
 aligned-sdk = { path = "../crates/sdk/" }
+db = { path = "./db" }
 sp1-sdk = "5.0.0"
 risc0-zkvm = { version = "3.0.3" }
 

aggregation_mode/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 serde =  { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
+agg_mode_sdk = { path = "../sdk"}
 aligned-sdk = { workspace = true }
 sp1-sdk = { workspace = true }
 tracing = { version = "0.1", features = ["log"] }

aggregation_mode/batcher/Cargo.toml

@@ -8,6 +8,8 @@ pub struct Config {
     pub db_connection_url: String,
     pub eth_rpc_url: String,
     pub payment_service_address: String,
+    pub network: String,
+    pub max_daily_proofs_per_user: i64,
 }
 
 impl Config {

aggregation_mode/batcher/src/config.rs

@@ -96,6 +96,18 @@ impl Db {
         .await
     }
 
+    pub async fn get_daily_tasks_by_address(&self, address: &str) -> Result<i64, sqlx::Error> {
+        sqlx::query_scalar::<_, i64>(
+            "SELECT COUNT(*)
+            FROM tasks
+            WHERE address = $1
+            AND inserted_at::date = CURRENT_DATE",
+        )
+        .bind(address.to_lowercase())
+        .fetch_one(&self.pool)
+        .await
+    }
+
     pub async fn insert_task(
         &self,
         address: &str,

aggregation_mode/batcher/src/db.rs

@@ -8,7 +8,9 @@ use actix_web::{
     web::{self, Data},
     App, HttpRequest, HttpResponse, HttpServer, Responder,
 };
+use agg_mode_sdk::types::Network;
 use aligned_sdk::aggregation_layer::AggregationModeProvingSystem;
+use alloy::signers::Signature;
 use sp1_sdk::{SP1ProofWithPublicValues, SP1VerifyingKey};
 use sqlx::types::BigDecimal;
 
@@ -28,11 +30,17 @@ use crate::{
 pub struct BatcherServer {
     db: Db,
     config: Config,
+    network: Network,
 }
 
 impl BatcherServer {
     pub fn new(db: Db, config: Config) -> Self {
-        Self { db, config }
+        let network = Network::from_str(&config.network).expect("A valid network in config file");
+        Self {
+            db,
+            config,
+            network,
+        }
     }
 
     pub async fn start(&self) {
@@ -93,13 +101,57 @@ impl BatcherServer {
         req: HttpRequest,
         MultipartForm(data): MultipartForm<SubmitProofRequestSP1>,
     ) -> impl Responder {
-        let recovered_address = "0x70997970C51812dc3A010C7d01b50e0d17dc79C8".to_lowercase();
-
         let Some(state) = req.app_data::<Data<BatcherServer>>() else {
             return HttpResponse::InternalServerError()
                 .json(AppResponse::new_unsucessfull("Internal server error", 500));
         };
+
         let state = state.get_ref();
+        let Ok(signature) = Signature::from_str(&data.signature_hex.0) else {
+            return HttpResponse::InternalServerError()
+                .json(AppResponse::new_unsucessfull("Invalid signature", 500));
+        };
+
+        let Ok(proof_content) = tokio::fs::read(data.proof.file.path()).await else {
+            return HttpResponse::InternalServerError()
+                .json(AppResponse::new_unsucessfull("Internal server error", 500));
+        };
+
+        let Ok(vk_content) = tokio::fs::read(data.program_vk.file.path()).await else {
+            return HttpResponse::InternalServerError()
+                .json(AppResponse::new_unsucessfull("Internal server error", 500));
+        };
+
+        // reconstruct message and recover address
+        let msg = agg_mode_sdk::gateway::types::SubmitSP1ProofMessage::new(
+            data.nonce.0,
+            proof_content.clone(),
+            vk_content.clone(),
+        );
+        let Ok(recovered_address) =
+            signature.recover_address_from_prehash(&msg.eip712_hash(&state.network).into())
+        else {
+            return HttpResponse::InternalServerError()
+                .json(AppResponse::new_unsucessfull("Internal server error", 500));
+        };
+        let recovered_address = recovered_address.to_string().to_lowercase();
+
+        // Checking if this address has submited more proofs than the ones allowed per day
+        let Ok(daily_tasks_by_address) = state
+            .db
+            .get_daily_tasks_by_address(&recovered_address)
+            .await
+        else {
+            return HttpResponse::InternalServerError()
+                .json(AppResponse::new_unsucessfull("Internal server error", 500));
+        };
+
+        if daily_tasks_by_address >= state.config.max_daily_proofs_per_user {
+            return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
+                "Request denied: Query limit exceeded.",
+                400,
+            ));
+        }
 
         let Ok(count) = state.db.count_tasks_by_address(&recovered_address).await else {
             return HttpResponse::InternalServerError()
@@ -143,22 +195,11 @@ impl BatcherServer {
                 400,
             ));
         }
-
-        let Ok(proof_content) = tokio::fs::read(data.proof.file.path()).await else {
-            return HttpResponse::InternalServerError()
-                .json(AppResponse::new_unsucessfull("Internal server error", 500));
-        };
-
         let Ok(proof) = bincode::deserialize::<SP1ProofWithPublicValues>(&proof_content) else {
             return HttpResponse::BadRequest()
                 .json(AppResponse::new_unsucessfull("Invalid SP1 proof", 400));
         };
 
-        let Ok(vk_content) = tokio::fs::read(data.program_vk.file.path()).await else {
-            return HttpResponse::InternalServerError()
-                .json(AppResponse::new_unsucessfull("Internal server error", 500));
-        };
-
         let Ok(vk) = bincode::deserialize::<SP1VerifyingKey>(&vk_content) else {
             return HttpResponse::BadRequest()
                 .json(AppResponse::new_unsucessfull("Invalid vk", 400));

aggregation_mode/batcher/src/server/http.rs

@@ -44,7 +44,7 @@ pub(super) struct SubmitProofRequestSP1 {
     pub nonce: Text<u64>,
     pub proof: TempFile,
     pub program_vk: TempFile,
-    pub _signature_hex: Text<String>,
+    pub signature_hex: Text<String>,
 }
 
 #[derive(Debug, MultipartForm)]

aggregation_mode/batcher/src/server/types.rs

@@ -0,0 +1,15 @@
+[package]
+name = "agg_mode_cli"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+serde =  { workspace = true }
+tracing = { version = "0.1", features = ["log"] }
+tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
+bincode = "1.3.3"
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] }
+alloy = { workspace = true }
+agg_mode_sdk = { path = "../sdk"}
+sp1-sdk = "5.0.0"
+clap = { version = "4.5.4", features = ["derive"] }

aggregation_mode/cli/Cargo.toml

@@ -0,0 +1 @@
+pub mod submit;