Merge from main

.github/workflows/check.yml

@@ -201,7 +201,7 @@ jobs:
       - name: Add Rust sources
         run: rustup component add rust-src
       - name: Run tests with sanitizer
-        run: RUSTFLAGS="-Z sanitizer=${{ matrix.sanitizer }} -Z sanitizer-memory-track-origins" cargo test -Z build-std -p ipa-core --target $TARGET --no-default-features --features "cli web-app real-world-infra test-fixture compact-gate ${{ matrix.features }}"
+        run: RUSTFLAGS="-Z sanitizer=${{ matrix.sanitizer }} -Z sanitizer-memory-track-origins" cargo test -Z build-std -p ipa-core --all-targets --target $TARGET --no-default-features --features "cli web-app real-world-infra test-fixture compact-gate ${{ matrix.features }}"
 
   miri:
     runs-on: ubuntu-latest

ipa-core/Cargo.toml

@@ -143,6 +143,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sha2 = "0.10"
 shuttle-crate = { package = "shuttle", version = "0.6.1", optional = true }
+subtle = "2.6"
 thiserror = "1.0"
 time = { version = "0.3", optional = true }
 tokio = { version = "1.35", features = ["fs", "rt", "rt-multi-thread", "macros"] }

ipa-core/benches/dzkp_convert_prover.rs

@@ -1,41 +1,15 @@
-//! Benchmark for the convert_prover function in dzkp_field.rs.
+//! Benchmark for the table_indices_prover function in dzkp_field.rs.
 
 use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
-use ipa_core::{
-    ff::Fp61BitPrime,
-    protocol::context::{dzkp_field::DZKPBaseField, dzkp_validator::MultiplicationInputsBlock},
-};
+use ipa_core::protocol::context::dzkp_validator::MultiplicationInputsBlock;
 use rand::{thread_rng, Rng};
 
 fn convert_prover_benchmark(c: &mut Criterion) {
     let mut group = c.benchmark_group("dzkp_convert_prover");
     group.bench_function("convert", |b| {
         b.iter_batched_ref(
-            || {
-                // Generate input
-                let mut rng = thread_rng();
-
-                MultiplicationInputsBlock {
-                    x_left: rng.gen::<[u8; 32]>().into(),
-                    x_right: rng.gen::<[u8; 32]>().into(),
-                    y_left: rng.gen::<[u8; 32]>().into(),
-                    y_right: rng.gen::<[u8; 32]>().into(),
-                    prss_left: rng.gen::<[u8; 32]>().into(),
-                    prss_right: rng.gen::<[u8; 32]>().into(),
-                    z_right: rng.gen::<[u8; 32]>().into(),
-                }
-            },
-            |input| {
-                let MultiplicationInputsBlock {
-                    x_left,
-                    x_right,
-                    y_left,
-                    y_right,
-                    prss_right,
-                    ..
-                } = input;
-                Fp61BitPrime::convert_prover(x_left, x_right, y_left, y_right, prss_right);
-            },
+            || thread_rng().gen(),
+            |input: &mut MultiplicationInputsBlock| input.table_indices_prover(),
             BatchSize::SmallInput,
         )
     });

ipa-core/src/app.rs

@@ -5,7 +5,7 @@ use async_trait::async_trait;
 use crate::{
     executor::IpaRuntime,
     helpers::{
-        query::{PrepareQuery, QueryConfig, QueryInput},
+        query::{CompareStatusRequest, PrepareQuery, QueryConfig, QueryInput},
         routing::{Addr, RouteId},
         ApiError, BodyStream, HandlerBox, HandlerRef, HelperIdentity, HelperResponse,
         MpcTransportImpl, RequestHandler, ShardTransportImpl, Transport, TransportIdentity,
@@ -149,8 +149,13 @@ impl HelperApp {
     ///
     /// ## Errors
     /// Propagates errors from the helper.
-    pub fn query_status(&self, query_id: QueryId) -> Result<QueryStatus, ApiError> {
-        Ok(self.inner.query_processor.query_status(query_id)?)
+    pub async fn query_status(&self, query_id: QueryId) -> Result<QueryStatus, ApiError> {
+        let shard_transport = self.inner.shard_transport.clone_ref();
+        Ok(self
+            .inner
+            .query_processor
+            .query_status(shard_transport, query_id)
+            .await?)
     }
 
     /// Waits for a query to complete and returns the result.
@@ -161,7 +166,7 @@ impl HelperApp {
         Ok(self
             .inner
             .query_processor
-            .complete(query_id)
+            .complete(query_id, self.inner.shard_transport.clone_ref())
             .await?
             .to_bytes())
     }
@@ -177,7 +182,7 @@ impl RequestHandler<ShardIndex> for Inner {
     async fn handle(
         &self,
         req: Addr<ShardIndex>,
-        _data: BodyStream,
+        data: BodyStream,
     ) -> Result<HelperResponse, ApiError> {
         let qp = &self.query_processor;
 
@@ -186,6 +191,17 @@ impl RequestHandler<ShardIndex> for Inner {
                 let req = req.into::<PrepareQuery>()?;
                 HelperResponse::from(qp.prepare_shard(&self.shard_transport, req)?)
             }
+            RouteId::QueryStatus => {
+                let req = req.into::<CompareStatusRequest>()?;
+                HelperResponse::from(qp.shard_status(&self.shard_transport, &req)?)
+            }
+            RouteId::CompleteQuery => {
+                // The processing flow for this API is exactly the same, regardless
+                // whether it was received from a peer shard or from report collector.
+                // Authentication is handled on the layer above, so we erase the identity
+                // and pass it down to the MPC handler.
+                RequestHandler::<HelperIdentity>::handle(self, req.erase_origin(), data).await?
+            }
             r => {
                 return Err(ApiError::BadRequest(
                     format!("{r:?} request must not be handled by shard query processing flow")
@@ -247,11 +263,16 @@ impl RequestHandler<HelperIdentity> for Inner {
             }
             RouteId::QueryStatus => {
                 let query_id = ext_query_id(&req)?;
-                HelperResponse::from(qp.query_status(query_id)?)
+                let shard_transport = Transport::clone_ref(&self.shard_transport);
+                let query_status = qp.query_status(shard_transport, query_id).await?;
+                HelperResponse::from(query_status)
             }
             RouteId::CompleteQuery => {
                 let query_id = ext_query_id(&req)?;
-                HelperResponse::from(qp.complete(query_id).await?)
+                HelperResponse::from(
+                    qp.complete(query_id, self.shard_transport.clone_ref())
+                        .await?,
+                )
             }
             RouteId::KillQuery => {
                 let query_id = ext_query_id(&req)?;

ipa-core/src/bin/helper.rs

@@ -183,6 +183,7 @@ async fn server(args: ServerArgs, logging_handle: LoggingHandle) -> Result<(), B
     let shard_index = ShardIndex::from(args.shard_index.expect("enforced by clap"));
     let shard_count = ShardIndex::from(args.shard_count.expect("enforced by clap"));
     assert!(shard_index < shard_count);
+    assert_eq!(args.tls_cert.is_some(), !args.disable_https);
 
     let (identity, server_tls) =
         create_client_identity(my_identity, args.tls_cert.clone(), args.tls_key.clone())?;
@@ -223,8 +224,13 @@ async fn server(args: ServerArgs, logging_handle: LoggingHandle) -> Result<(), B
 
     let network_config_path = args.network.as_deref().unwrap();
     let network_config_string = &fs::read_to_string(network_config_path)?;
-    let (mut mpc_network, mut shard_network) =
-        sharded_server_from_toml_str(network_config_string, my_identity, shard_index, shard_count)?;
+    let (mut mpc_network, mut shard_network) = sharded_server_from_toml_str(
+        network_config_string,
+        my_identity,
+        shard_index,
+        shard_count,
+        args.shard_port,
+    )?;
     mpc_network = mpc_network.override_scheme(&scheme);
     shard_network = shard_network.override_scheme(&scheme);
 

ipa-core/src/config.rs

@@ -3,6 +3,7 @@ use std::{
     fmt::{Debug, Formatter},
     iter::zip,
     path::PathBuf,
+    str::FromStr,
     time::Duration,
 };
 
@@ -280,20 +281,21 @@ fn parse_sharded_network_toml(input: &str) -> Result<ShardedNetworkToml, Error>
 ///
 /// # Errors
 /// if `input` is in an invalid format
+///
+/// # Panics
+/// If you somehow provide an invalid non-sharded network toml
 pub fn sharded_server_from_toml_str(
     input: &str,
     id: HelperIdentity,
     shard_index: ShardIndex,
     shard_count: ShardIndex,
+    shard_port: Option<u16>,
 ) -> Result<(NetworkConfig<Helper>, NetworkConfig<Shard>), Error> {
     let all_network = parse_sharded_network_toml(input)?;
-    let missing_urls = all_network.missing_shard_urls();
-    if !missing_urls.is_empty() {
-        return Err(Error::MissingShardUrls(missing_urls));
-    }
 
     let ix: usize = shard_index.as_index();
     let ix_count: usize = shard_count.as_index();
+    // assert ix < count
     let mpc_id: usize = id.as_index();
 
     let mpc_network = NetworkConfig {
@@ -307,21 +309,45 @@ pub fn sharded_server_from_toml_str(
         client: all_network.client.clone(),
         identities: HelperIdentity::make_three().to_vec(),
     };
-
-    let shard_network = NetworkConfig {
-        peers: all_network
-            .peers
-            .iter()
-            .map(ShardedPeerConfigToml::to_shard_peer)
-            .skip(mpc_id)
-            .step_by(3)
-            .take(ix_count)
-            .collect(),
-        client: all_network.client,
-        identities: shard_count.iter().collect(),
-    };
-
-    Ok((mpc_network, shard_network))
+    let missing_urls = all_network.missing_shard_urls();
+    if missing_urls.is_empty() {
+        let shard_network = NetworkConfig {
+            peers: all_network
+                .peers
+                .iter()
+                .map(ShardedPeerConfigToml::to_shard_peer)
+                .skip(mpc_id)
+                .step_by(3)
+                .take(ix_count)
+                .collect(),
+            client: all_network.client,
+            identities: shard_count.iter().collect(),
+        };
+        Ok((mpc_network, shard_network))
+    } else if missing_urls == [0, 1, 2] && shard_count == ShardIndex(1) {
+        // This is the special case we're dealing with a non-sharded, single ring MPC.
+        // Since the shard network will be of size 1, it can't really communicate with anyone else.
+        // Hence we just create a config where I'm the only shard. We take the MPC configuration
+        // and modify the port.
+        let mut myself = ShardedPeerConfigToml::to_mpc_peer(all_network.peers.get(mpc_id).unwrap());
+        let url = myself.url.to_string();
+        let pos = url.rfind(':');
+        let port = shard_port.expect("Shard port should be set");
+        let new_url = if pos.is_some() {
+            format!("{}{port}", &url[..=pos.unwrap()])
+        } else {
+            format!("{}:{port}", &url)
+        };
+        myself.url = Uri::from_str(&new_url).expect("Problem creating uri with sharded port");
+        let shard_network = NetworkConfig {
+            peers: vec![myself],
+            client: all_network.client,
+            identities: shard_count.iter().collect(),
+        };
+        Ok((mpc_network, shard_network))
+    } else {
+        return Err(Error::MissingShardUrls(missing_urls));
+    }
 }
 
 #[derive(Clone, Debug, Deserialize)]
@@ -788,6 +814,7 @@ mod tests {
             HelperIdentity::TWO,
             ShardIndex::from(1),
             ShardIndex::from(3),
+            None,
         )
         .unwrap();
         assert_eq!(
@@ -856,19 +883,43 @@ mod tests {
         ));
     }
 
-    /// Check that shard urls are given for [`sharded_server_from_toml_str`] or error is returned.
+    /// Check that [`sharded_server_from_toml_str`] can work in the previous format.
     #[test]
     fn parse_sharded_without_shard_urls() {
-        // Second, I test the networkconfig parsing
-        assert!(matches!(
-            sharded_server_from_toml_str(
-                &NON_SHARDED_COMPAT,
-                HelperIdentity::TWO,
-                ShardIndex::FIRST,
-                ShardIndex::from(1)
-            ),
-            Err(Error::MissingShardUrls(_))
-        ));
+        let (mpc, mut shard) = sharded_server_from_toml_str(
+            &NON_SHARDED_COMPAT,
+            HelperIdentity::ONE,
+            ShardIndex::FIRST,
+            ShardIndex::from(1),
+            Some(666),
+        )
+        .unwrap();
+        assert_eq!(1, shard.peers.len());
+        assert_eq!(3, mpc.peers.len());
+        assert_eq!(
+            "helper1.org:666",
+            shard.peers.pop().unwrap().url.to_string()
+        );
+    }
+
+    /// Check that [`sharded_server_from_toml_str`] can work in the previous format, even when the
+    /// given MPC URL doesn't have a port (NOTE: helper 2 doesn't specify it).
+    #[test]
+    fn parse_sharded_without_shard_urls_no_port() {
+        let (mpc, mut shard) = sharded_server_from_toml_str(
+            &NON_SHARDED_COMPAT,
+            HelperIdentity::TWO,
+            ShardIndex::FIRST,
+            ShardIndex::from(1),
+            Some(666),
+        )
+        .unwrap();
+        assert_eq!(1, shard.peers.len());
+        assert_eq!(3, mpc.peers.len());
+        assert_eq!(
+            "helper2.org:666",
+            shard.peers.pop().unwrap().url.to_string()
+        );
     }
 
     /// Testing happy case of a sharded network config
@@ -959,6 +1010,7 @@ a6L0t1Ug8i2RcequSo21x319Tvs5nUbGwzMFSS5wKA==
 url = "helper1.org:443""#;
 
     /// The rest of a configuration
+    /// Note the second helper doesn't provide a port as part of its url
     const REST: &str = r#"
 [peers.hpke]
 public_key = "f458d5e1989b2b8f5dacd4143276aa81eaacf7449744ab1251ff667c43550756"
@@ -977,7 +1029,7 @@ zj0EAwIDSAAwRQIhAI4G5ICVm+v5KK5Y8WVetThtNCXGykUBAM1eE973FBOUAiAS
 XXgJe9q9hAfHf0puZbv0j0tGY3BiqCkJJaLvK7ba+g==
 -----END CERTIFICATE-----
 """
-url = "helper2.org:443"
+url = "helper2.org"
 
 [peers.hpke]
 public_key = "62357179868e5594372b801ddf282c8523806a868a2bff2685f66aa05ffd6c22"