Merge pull request #458 from Cerebellum-Network/feature/dac-challenge…

…-protobuf

Signature verification during aggregate challenge

pallets/ddc-verification/Cargo.toml

@@ -24,6 +24,7 @@ hex = { workspace = true }
 itertools = { workspace = true }
 log = { workspace = true }
 polkadot-ckb-merkle-mountain-range = { workspace = true }
+prost = { version = "0.13", default-features = false, features = ["prost-derive"] }
 rand = { workspace = true, features = ["small_rng", "alloc"], default-features = false }
 scale-info = { workspace = true }
 serde = { workspace = true }
@@ -44,6 +45,9 @@ pallet-timestamp = { workspace = true }
 sp-core = { workspace = true, default-features = true }
 sp-keystore = { workspace = true }
 
+[build-dependencies]
+prost-build = "0.13.3"
+
 [features]
 default = ["std"]
 std = [

pallets/ddc-verification/build.rs

@@ -0,0 +1,8 @@
+use std::io::Result;
+
+fn main() -> Result<()> {
+	let mut prost_build = prost_build::Config::new();
+	prost_build.protoc_arg("--experimental_allow_proto3_optional");
+	prost_build.compile_protos(&["src/protos/activity.proto"], &["src/"])?;
+	Ok(())
+}

pallets/ddc-verification/src/aggregator_client.rs

@@ -0,0 +1,215 @@
+#![allow(dead_code)]
+
+use ddc_primitives::{BucketId, DdcEra};
+use prost::Message;
+use sp_io::offchain::timestamp;
+use sp_runtime::offchain::{http, Duration};
+
+use super::*;
+
+pub struct AggregatorClient<'a> {
+	pub base_url: &'a str,
+	timeout: Duration,
+	retries: u32,
+}
+
+impl<'a> AggregatorClient<'a> {
+	pub fn new(base_url: &'a str, timeout: Duration, retries: u32) -> Self {
+		Self { base_url, timeout, retries }
+	}
+
+	pub fn buckets_aggregates(
+		&self,
+		era_id: DdcEra,
+		limit: Option<u32>,
+		prev_token: Option<BucketId>,
+	) -> Result<Vec<pallet::BucketAggregateResponse>, http::Error> {
+		let mut url = format!("{}/activity/buckets?eraId={}", self.base_url, era_id);
+		if let Some(limit) = limit {
+			url = format!("{}&limit={}", url, limit);
+		}
+		if let Some(prev_token) = prev_token {
+			url = format!("{}&prevToken={}", url, prev_token);
+		}
+		let response = self.get(&url, Accept::Any)?;
+		let body = response.body().collect::<Vec<u8>>();
+		let json_response = serde_json::from_slice(&body).map_err(|_| http::Error::Unknown)?;
+
+		Ok(json_response)
+	}
+
+	pub fn nodes_aggregates(
+		&self,
+		era_id: DdcEra,
+		limit: Option<u32>,
+		prev_token: Option<String>, // node_id hex string
+	) -> Result<Vec<pallet::NodeAggregateResponse>, http::Error> {
+		let mut url = format!("{}/activity/nodes?eraId={}", self.base_url, era_id);
+		if let Some(limit) = limit {
+			url = format!("{}&limit={}", url, limit);
+		}
+		if let Some(prev_token) = prev_token {
+			url = format!("{}&prevToken={}", url, prev_token);
+		}
+		let response = self.get(&url, Accept::Any)?;
+		let body = response.body().collect::<Vec<u8>>();
+		let json_response = serde_json::from_slice(&body).map_err(|_| http::Error::Unknown)?;
+
+		Ok(json_response)
+	}
+
+	pub fn challenge_bucket_sub_aggregate(
+		&self,
+		era_id: DdcEra,
+		bucket_id: BucketId,
+		node_id: &str,
+		merkle_tree_node_id: Vec<u32>,
+	) -> Result<proto::ChallengeResponse, http::Error> {
+		let url = format!(
+			"{}/activity/buckets/{}/challenge?eraId={}&nodeId={}&merkleTreeNodeId={}",
+			self.base_url,
+			bucket_id,
+			era_id,
+			node_id,
+			Self::merkle_tree_node_id_param(merkle_tree_node_id.as_slice()),
+		);
+		let response = self.get(&url, Accept::Protobuf)?;
+		let body = response.body().collect::<Vec<u8>>();
+		let proto_response =
+			proto::ChallengeResponse::decode(body.as_slice()).map_err(|_| http::Error::Unknown)?;
+
+		Ok(proto_response)
+	}
+
+	pub fn challenge_node_aggregate(
+		&self,
+		era_id: DdcEra,
+		node_id: &str,
+		merkle_tree_node_id: Vec<u32>,
+	) -> Result<proto::ChallengeResponse, http::Error> {
+		let url = format!(
+			"{}/activity/nodes/{}/challenge?eraId={}&merkleTreeNodeId={}",
+			self.base_url,
+			node_id,
+			era_id,
+			Self::merkle_tree_node_id_param(merkle_tree_node_id.as_slice()),
+		);
+		let response = self.get(&url, Accept::Protobuf)?;
+		let body = response.body().collect::<Vec<u8>>();
+		let proto_response =
+			proto::ChallengeResponse::decode(body.as_slice()).map_err(|_| http::Error::Unknown)?;
+
+		Ok(proto_response)
+	}
+
+	pub fn eras(&self) -> Result<Vec<pallet::AggregationEraResponse>, http::Error> {
+		let url = format!("{}/activity/eras", self.base_url);
+		let response = self.get(&url, Accept::Any)?;
+		let body = response.body().collect::<Vec<u8>>();
+		let json_response = serde_json::from_slice(&body).map_err(|_| http::Error::Unknown)?;
+
+		Ok(json_response)
+	}
+
+	pub fn traverse_bucket_sub_aggregate(
+		&self,
+		era_id: DdcEra,
+		bucket_id: BucketId,
+		node_id: &str,
+		merkle_tree_node_id: u32,
+		levels: u16,
+	) -> Result<pallet::MerkleTreeNodeResponse, http::Error> {
+		let url = format!(
+			"{}/activity/buckets/{}/traverse?eraId={}&nodeId={}&merkleTreeNodeId={}&levels={}",
+			self.base_url, bucket_id, era_id, node_id, merkle_tree_node_id, levels,
+		);
+
+		let response = self.get(&url, Accept::Any)?;
+		let body = response.body().collect::<Vec<u8>>();
+		let json_response = serde_json::from_slice(&body).map_err(|_| http::Error::Unknown)?;
+
+		Ok(json_response)
+	}
+
+	pub fn traverse_node_aggregate(
+		&self,
+		era_id: DdcEra,
+		node_id: &str,
+		merkle_tree_node_id: u32,
+		levels: u16,
+	) -> Result<pallet::MerkleTreeNodeResponse, http::Error> {
+		let url = format!(
+			"{}/activity/nodes/{}/traverse?eraId={}&merkleTreeNodeId={}&levels={}",
+			self.base_url, node_id, era_id, merkle_tree_node_id, levels,
+		);
+
+		let response = self.get(&url, Accept::Any)?;
+		let body = response.body().collect::<Vec<u8>>();
+		let json_response = serde_json::from_slice(&body).map_err(|_| http::Error::Unknown)?;
+
+		Ok(json_response)
+	}
+
+	fn merkle_tree_node_id_param(merkle_tree_node_id: &[u32]) -> String {
+		merkle_tree_node_id
+			.iter()
+			.map(|x| format!("{}", x.clone()))
+			.collect::<Vec<_>>()
+			.join(",")
+	}
+
+	fn get(&self, url: &str, accept: Accept) -> Result<http::Response, http::Error> {
+		let mut maybe_response = None;
+
+		let deadline = timestamp().add(self.timeout);
+		let mut error = None;
+
+		for _ in 0..self.retries {
+			let mut request = http::Request::get(url).deadline(deadline);
+			request = match accept {
+				Accept::Any => request,
+				Accept::Protobuf => request.add_header("Accept", "application/protobuf"),
+			};
+
+			let pending = match request.send() {
+				Ok(p) => p,
+				Err(_) => {
+					error = Some(http::Error::IoError);
+					continue;
+				},
+			};
+
+			match pending.try_wait(deadline) {
+				Ok(Ok(r)) => {
+					maybe_response = Some(r);
+					error = None;
+					break;
+				},
+				Ok(Err(_)) | Err(_) => {
+					error = Some(http::Error::DeadlineReached);
+					continue;
+				},
+			}
+		}
+
+		if let Some(e) = error {
+			return Err(e);
+		}
+
+		let response = match maybe_response {
+			Some(r) => r,
+			None => return Err(http::Error::Unknown),
+		};
+
+		if response.code != 200 {
+			return Err(http::Error::Unknown);
+		}
+
+		Ok(response)
+	}
+}
+
+enum Accept {
+	Any,
+	Protobuf,
+}