enable env to set FulfilmentStrategy for network provers (#32)

nuke-web3 · web-flow · commit 717a56e5be83 · 2025-06-23T18:11:51.000-06:00
* enable env to set FulfilmentStrategy for network provers
* rm network zk prover timeout
* actually submit encrypted blob lol
* defensive take to ensure no unencrypted data remains in intercepted body
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/example.env b/example.env
@@ -73,6 +73,10 @@ SP1_PROVER=cuda
 # TODO: This feature is NOT fully implemented! We need a TEE wrapper around proving remote for this.
 NETWORK_RPC_URL=https://rpc.production.succinct.xyz
 NETWORK_PRIVATE_KEY=
+# Strategy one of "UNSPECIFIED_FULFILLMENT_STRATEGY", "HOSTED", "RESERVED", "AUCTION"
+# You likely want HOSTED (the hardcoded default for network provers)
+# https://github.com/succinctlabs/sp1/blob/11ab6b783cfce295b6f1113af088cc5f0a8caa5b/crates/sdk/src/network/prover.rs#L138
+SP1_FULFILLMENT_STRATEGY=HOSTED
 
 #### Dependent & Provider Settings
 
@@ -90,7 +94,3 @@ CELESTIA_NODE_NAME=celestia-light-mocha-4
 CELESTIA_DATA_DIR=/home/you/.celestia-light-mocha-4
 CELESTIA_NODE_WRITE_TOKEN=
 
-
-
-
-
diff --git a/justfile b/justfile
@@ -139,12 +139,23 @@ celestia-node-balance:
       --data '{"jsonrpc":"2.0","id":1,"method":"state.Balance","params":[]}' \
       http://127.0.0.1:26658 | jq
 
+# https://mocha-4.celenium.io/tx/28fa01d026ac5a229e5d5472a204d290beda02ea229f6b3f42da520b00154e58?tab=messages
+
 # Test blob.Get for PDA Proxy
 curl-blob-get:
-    curl -H "Content-Type: application/json" -H "Authorization: Bearer $CELESTIA_NODE_WRITE_TOKEN" --data '{"id": 1,"jsonrpc": "2.0", "method": "blob.Get", "params": [ 42, "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAMJ/xGlNMdE=", "aHlbp+J9yub6hw/uhK6dP8hBLR2mFy78XNRRdLf2794=" ] }' \
+    curl -H "Content-Type: application/json" -H "Authorization: Bearer $CELESTIA_NODE_WRITE_TOKEN" --data '{"id": 1,"jsonrpc": "2.0", "method": "blob.Get", "params": [ 6629478, "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAMJ/xGlNMdE=", "yS3XX33mc1uXkGinkTCvS9oqE0k9mtHMWTz0mwZccOc=" ] }' \
     https://127.0.0.1:26657 \
     --insecure | jq
 
+# https://mocha-4.celenium.io/tx/28fa01d026ac5a229e5d5472a204d290beda02ea229f6b3f42da520b00154e58?tab=messages
+
+# Test blob.Get for local light node
+curl-blob-get-passthrough:
+    curl -H "Content-Type: application/json" -H "Authorization: Bearer $CELESTIA_NODE_WRITE_TOKEN" --data '{"id": 1,"jsonrpc": "2.0", "method": "blob.Get", "params": [ 6629478, "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAMJ/xGlNMdE=", "yS3XX33mc1uXkGinkTCvS9oqE0k9mtHMWTz0mwZccOc=" ] }' \
+    http://127.0.0.1:26658 | jq
+
+# https://mocha.celenium.io/tx/436f223bfa8c4adf1e1b79dde43a84918f3a50809583c57c33c1c079568b47cb?tab=messages
+
 # Test blob.Submit for PDA proxy
 curl-blob-submit:
     curl -H "Content-Type: application/json" -H "Authorization: Bearer $CELESTIA_NODE_WRITE_TOKEN" \
diff --git a/service/Cargo.toml b/service/Cargo.toml
@@ -4,6 +4,8 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
+zkvm-common = { workspace = true, features = ["std"] }
+
 anyhow.workspace = true
 serde.workspace = true
 serde_json.workspace = true
diff --git a/service/src/internal/runner.rs b/service/src/internal/runner.rs
@@ -6,7 +6,8 @@ use sha2::Digest;
 use sled::{Transactional, Tree as SledTree};
 use sp1_sdk::{
     EnvProver as SP1EnvProver, NetworkProver as SP1NetworkProver, Prover, SP1ProofWithPublicValues,
-    SP1Stdin, network::Error as SP1NetworkError,
+    SP1Stdin,
+    network::{Error as SP1NetworkError, FulfillmentStrategy},
 };
 use std::sync::Arc;
 use tokio::sync::{OnceCell, mpsc};
@@ -521,6 +522,12 @@ impl PdaRunner {
             .get_proof_setup_remote(program_id, zk_client_handle.clone())
             .await?;
 
+        let fs_string = std::env::var("SP1_FULFILLMENT_STRATEGY")
+            .expect("Missing SP1_FULFILLMENT_STRATEGY env var");
+        let strategy = FulfillmentStrategy::from_str_name(fs_string.as_str()).ok_or(
+            PdaRunnerError::InternalError("SP1_FULFILLMENT_STRATEGY env var malformed".to_string()),
+        )?;
+
         let mut stdin = SP1Stdin::new();
         // Setup the inputs:
         // - key = 32 bytes
@@ -544,9 +551,10 @@ impl PdaRunner {
         debug!("0x{} - Starting proof", hex::encode(job_key));
         let request_id: util::SuccNetJobId = zk_client_handle
             .prove(&proof_setup.pk, &stdin)
+            .strategy(strategy)
             .groth16()
             .skip_simulation(false)
-            .timeout(std::time::Duration::from_secs(5)) // Don't hang too long on this. If it's gonna fail, fail fast.
+            // .timeout(std::time::Duration::from_secs(60))
             .request_async()
             .await
             // TODO: how to handle errors without a concrete type? Anyhow is not the right thing for us...
diff --git a/service/src/main.rs b/service/src/main.rs
@@ -16,6 +16,7 @@ use log::{debug, error, info, warn};
 use rustls::ServerConfig;
 use serde_json::json;
 use sha2::{Digest, Sha256};
+use sp1_sdk::SP1ProofWithPublicValues;
 use std::{net::SocketAddr, sync::Arc};
 use tokio::{
     net::TcpListener,
@@ -29,6 +30,8 @@ use internal::job::*;
 use internal::runner::*;
 use internal::util::*;
 
+use zkvm_common::{chacha, std_only::ZkvmOutput};
+
 type GenericError = Box<dyn std::error::Error + Send + Sync>;
 type BoxBody = http_body_util::combinators::BoxBody<Bytes, GenericError>;
 
@@ -198,40 +201,47 @@ async fn main() -> Result<()> {
                             }
 
                             let mut request_method: String = Default::default();
-                            let maybe_wrapped_req =
-                                match inbound_handler(plaintext_req, &mut request_method, runner)
-                                    .await
-                                {
-                                    Ok(req) => req,
-                                    Err(e) => return Ok(internal_error_response(e.to_string())),
-                                };
+                            let maybe_wrapped_req = match inbound_handler(
+                                plaintext_req,
+                                &mut request_method,
+                                runner.clone(),
+                            )
+                            .await
+                            {
+                                Ok(req) => req,
+                                Err(e) => return Ok(internal_error_response(e.to_string())),
+                            };
 
                             match maybe_wrapped_req {
                                 Some(wrapped_req) => {
-                                    debug!(
-                                        "Forwarding (maybe modified) `{}` --> DA: {:?}",
-                                        request_method, wrapped_req
-                                    );
+                                    debug!("Forwarding (maybe modified) --> DA",);
+                                    // debug!(
+                                    //     "Forwarding (maybe modified) `{}` --> DA: {:?}",
+                                    //     request_method, wrapped_req
+                                    // );
                                     let returned = match celestia_client.request(wrapped_req).await
                                     {
-                                        Ok(resp) => {
-                                            outbound_handler(resp, request_method.to_owned())
-                                                .await
-                                                .unwrap_or_else(|e| {
-                                                    internal_error_response(format!(
-                                                        "Outbound Handler: {}",
-                                                        e
-                                                    ))
-                                                })
-                                        }
+                                        Ok(resp) => outbound_handler(
+                                            resp,
+                                            request_method.to_owned(),
+                                            runner,
+                                        )
+                                        .await
+                                        .unwrap_or_else(|e| {
+                                            internal_error_response(format!(
+                                                "Outbound Handler: {}",
+                                                e
+                                            ))
+                                        }),
                                         Err(e) => {
                                             internal_error_response(format!("DA Client: {}", e))
                                         }
                                     };
-                                    debug!(
-                                        "Responding (maybe modified) `{}` <-- DA: {:?}",
-                                        request_method, returned
-                                    );
+                                    debug!("Responding (maybe modified)  <-- DA",);
+                                    // debug!(
+                                    //     "Responding (maybe modified) `{}` <-- DA: {:?}",
+                                    //     request_method, returned
+                                    // );
                                     anyhow::Ok(returned)
                                 }
                                 None => Ok(pending_response()),
@@ -284,11 +294,10 @@ pub async fn inbound_handler(
                             .get_mut(0)
                             .ok_or_else(|| anyhow::anyhow!("Expected first 'params' entry"))?;
 
-                        let blobs: Vec<Blob> = serde_json::from_value(blobs_value.clone())?;
+                        let blobs: Vec<Blob> = serde_json::from_value(std::mem::take(blobs_value))?;
 
                         let mut encrypted_blobs = Vec::with_capacity(blobs.len());
 
-                        // TODO: consider only allowing one blob and one job on the queue
                         for blob in blobs {
                             let pda_runner = pda_runner.clone();
                             let data = blob.data.clone();
@@ -304,10 +313,11 @@ pub async fn inbound_handler(
                             if let Some(proof_with_values) =
                                 pda_runner.get_verifiable_encryption(job).await?
                             {
-                                debug!(
-                                    "Replacing blob.data with <Sp1ProofWithPublicValues>.proof = {:?}",
-                                    proof_with_values.proof
-                                );
+                                debug!("Replacing blob.data with <Sp1ProofWithPublicValues>.proof");
+                                // debug!(
+                                //     "Replacing blob.data with <Sp1ProofWithPublicValues>.proof = {:?}",
+                                //     proof_with_values.proof
+                                // );
 
                                 let encrypted_data = bincode::serialize(&proof_with_values)?;
                                 let encrypted_blob = Blob::new(
@@ -318,9 +328,12 @@ pub async fn inbound_handler(
 
                                 encrypted_blobs.push(encrypted_blob);
                             } else {
-                                return Ok(None);
+                                return Ok(None); // Bail out if any blob can't be encrypted
                             }
                         }
+
+                        // Overwrite the original blob array in the params with encrypted blobs
+                        *blobs_value = serde_json::to_value(encrypted_blobs)?;
                     }
                 } else {
                     debug!("Forwarding `blob.Submit` error: missing params");
@@ -355,48 +368,93 @@ pub async fn inbound_handler(
 async fn outbound_handler(
     resp: Response<IncomingBody>,
     request_method: String,
+    pda_runner: Arc<PdaRunner>,
 ) -> Result<Response<BoxBody>> {
     let (mut parts, body_stream) = resp.into_parts();
     let mut body_buf = body_stream.collect().await?.aggregate();
     let body_bytes = body_buf.copy_to_bytes(body_buf.remaining());
 
-    debug!("Raw upstream response: {:?}", body_bytes);
-
     let status = parts.status;
-
     if status == StatusCode::UNAUTHORIZED {
         return Ok(bad_auth_response());
     }
 
-    let body_json: serde_json::Value = serde_json::from_slice(&body_bytes)?;
-    // (Optional) handle blob.Get/All for decryption etc.
-    match request_method.as_str() {
-        "blob.Get" => {
-            if let Some(result_raw) = body_json.get("result") {
+    let mut body_json: serde_json::Value = serde_json::from_slice(&body_bytes)?;
+
+    let try_mutate_response = async {
+        let result_raw = body_json
+            .get_mut("result")
+            .ok_or_else(|| anyhow::anyhow!("Missing 'result' field"))?;
+
+        let key = <[u8; 32]>::from_hex(
+            std::env::var("ENCRYPTION_KEY").expect("Missing ENCRYPTION_KEY env var"),
+        )
+        .expect("ENCRYPTION_KEY must be 32 bytes, hex encoded (ex: `1234...abcd`)");
+
+        match request_method.as_str() {
+            "blob.Get" => {
                 let blob: Blob = serde_json::from_value(result_raw.clone())?;
-                debug!("{blob:?}");
+                let plaintext_only_blob =
+                    verify_decrypt_blob(blob, key, pda_runner.clone()).await?;
+                *result_raw = serde_json::to_value(plaintext_only_blob)?;
             }
-        }
-        "blob.GetAll" => {
-            if let Some(result_raw) = body_json.get("result") {
-                let blobs: Vec<Blob> = serde_json::from_value(result_raw.clone())?;
-                for blob in blobs {
-                    debug!("{blob:?}");
+
+            "blob.GetAll" => {
+                let original_array = result_raw
+                    .as_array()
+                    .ok_or_else(|| anyhow::anyhow!("Expected 'result' to be an array"))?
+                    .clone();
+
+                let mut plaintext_blobs = Vec::with_capacity(original_array.len());
+                for blob_val in original_array {
+                    let blob: Blob = serde_json::from_value(blob_val)?;
+                    let plaintext_only_blob =
+                        verify_decrypt_blob(blob, key, pda_runner.clone()).await?;
+                    plaintext_blobs.push(serde_json::to_value(plaintext_only_blob)?);
                 }
+
+                *result_raw = serde_json::Value::Array(plaintext_blobs);
             }
+
+            _ => {}
         }
-        _ => {}
+
+        Ok::<_, anyhow::Error>(())
+    }
+    .await;
+
+    if let Err(err) = try_mutate_response {
+        warn!("Failed to decrypt response: {:?}", err);
+        let orig_body = Full::new(body_bytes)
+            .map_err(|err: std::convert::Infallible| match err {})
+            .boxed();
+        return Ok(Response::from_parts(parts, orig_body));
     }
 
     let json = serde_json::to_string(&body_json)?;
     parts.headers.remove("content-length");
+
     let new_body = Full::new(Bytes::from(json))
         .map_err(|err: std::convert::Infallible| match err {})
         .boxed();
 
     Ok(Response::from_parts(parts, new_body))
 }
 
+async fn verify_decrypt_blob(
+    blob: Blob,
+    key: [u8; 32],
+    pda_runner: Arc<PdaRunner>,
+) -> Result<Blob, anyhow::Error> {
+    let proof: SP1ProofWithPublicValues = bincode::deserialize(&blob.data)?;
+    let output = extract_verified_proof_output(&proof, pda_runner).await?;
+    let mut buffer = output.ciphertext.to_owned();
+    chacha(&key, &output.nonce, &mut buffer);
+    let mut decrypted_plaintext_blob = blob.clone();
+    decrypted_plaintext_blob.data = buffer.to_vec();
+    Ok(decrypted_plaintext_blob)
+}
+
 /// Job is in queue, we are waiting on it to finish
 fn pending_response() -> Response<BoxBody> {
     let raw_json = r#"{ "id": 1, "jsonrpc": "2.0", "status": "[pda-proxy] Verifiable encryption processing... Call back for result" }"#;
@@ -438,3 +496,18 @@ fn new_response_from(raw_json: &str, status: StatusCode) -> Response<BoxBody> {
     *response.status_mut() = status;
     response
 }
+
+/// Verify a proof before returning it's attested output
+async fn extract_verified_proof_output<'a>(
+    proof: &'a SP1ProofWithPublicValues,
+    runner: Arc<PdaRunner>,
+) -> Result<ZkvmOutput<'a>> {
+    let zk_client_local = runner.get_zk_client_local().await;
+    let vk = &runner
+        .get_proof_setup_local(&get_program_id().await, zk_client_local.clone())
+        .await?
+        .vk;
+    zk_client_local.verify(proof, vk)?;
+
+    ZkvmOutput::from_bytes(proof.public_values.as_slice()).map_err(anyhow::Error::msg)
+}
diff --git a/zkVM/common/src/lib.rs b/zkVM/common/src/lib.rs
@@ -18,7 +18,7 @@ pub const HASH_LEN: usize = 32;
 pub const NONCE_LEN: usize = 12;
 pub const HEADER_LEN: usize = HASH_LEN + NONCE_LEN + HASH_LEN;
 
-/// Encrypt a buffer in-place using [ChaCha20](https://en.wikipedia.org/wiki/Salsa20#ChaCha_variant).
+/// Encrypt or Decrypt a buffer in-place using [ChaCha20](https://en.wikipedia.org/wiki/Salsa20#ChaCha_variant).
 ///
 /// ## Important Notice
 ///
@@ -32,7 +32,6 @@ pub fn chacha(key: &[u8; KEY_LEN], nonce: &[u8; NONCE_LEN], buffer: &mut [u8]) {
     cipher.apply_keystream(buffer);
 }
 
-// Only compile this when the standard library is available
 #[cfg(feature = "std")]
 pub mod std_only {
     use super::*;
@@ -93,14 +92,14 @@ pub mod std_only {
         }
     }
 
-    // Helper to get a OsRng nonce of correct length
+    /// Helper to get a OsRng nonce of correct length
     pub fn random_nonce() -> [u8; NONCE_LEN] {
         let mut nonce = [0u8; NONCE_LEN];
         OsRng.try_fill_bytes(&mut nonce).expect("Rng->buffer");
         nonce
     }
 
-    // Helper to format bytes as hex for pretty printing
+    /// Helper to format bytes as hex for pretty printing
     pub fn bytes_to_hex(bytes: &[u8]) -> String {
         let digest_hex: String = bytes.iter().map(|b| format!("{:02x}", b)).collect();
         digest_hex
