fix: don't ser/de the binary message, leave that to the client. only b64 encode. add parse and verify to the example.

Author: tejasbadadare

lazer/Cargo.lock

@@ -16,3 +16,11 @@ base64 = "0.22.1"
 anyhow = "1.0"
 tracing = "0.1"
 url = "2.4"
+
+[dev_dependencies]
+bincode = "1.3.3"
+ed25519-dalek = { version = "2.1.1", features = ["rand_core"] }
+hex = "0.4.3"
+libsecp256k1 = "0.7.1"
+bs58 = "0.5.1"
+alloy-primitives = "0.8.19"

lazer/sdk/rust/client/Cargo.toml

@@ -1,10 +1,13 @@
+use base64::Engine;
 use futures_util::StreamExt;
 use pyth_lazer_client::LazerClient;
+use pyth_lazer_protocol::message::{EvmMessage, SolanaMessage};
+use pyth_lazer_protocol::payload::PayloadData;
 use pyth_lazer_protocol::router::{
     Chain, Channel, DeliveryFormat, FixedRate, JsonBinaryEncoding, PriceFeedId, PriceFeedProperty,
     SubscriptionParams, SubscriptionParamsRepr,
 };
-use pyth_lazer_protocol::subscription::{Request, SubscribeRequest, SubscriptionId};
+use pyth_lazer_protocol::subscription::{Request, Response, SubscribeRequest, SubscriptionId};
 
 fn get_lazer_access_token() -> String {
     // Place your access token in your env at LAZER_ACCESS_TOKEN or set it here
@@ -75,7 +78,44 @@ async fn main() -> anyhow::Result<()> {
     // Process the first few updates
     let mut count = 0;
     while let Some(msg) = stream.next().await {
-        println!("Received update: {:?}", msg?);
+        // The stream gives us base64-encoded binary messages. We need to decode, parse, and verify them.
+        match msg? {
+            Response::StreamUpdated(update) => {
+                if let Some(evm_data) = update.payload.evm {
+                    // Decode binary data
+                    let binary_data =
+                        base64::engine::general_purpose::STANDARD.decode(&evm_data.data)?;
+                    let evm_message = EvmMessage::deserialize_slice(&binary_data)?;
+
+                    // Parse and verify the EVM message
+                    let payload = parse_and_verify_evm_message(&evm_message);
+                    println!("EVM payload: {payload:?}\n");
+                }
+
+                if let Some(solana_data) = update.payload.solana {
+                    // Decode binary data
+                    let binary_data =
+                        base64::engine::general_purpose::STANDARD.decode(&solana_data.data)?;
+                    let solana_message = SolanaMessage::deserialize_slice(&binary_data)?;
+
+                    // Parse and verify the Solana message
+                    let payload = parse_and_verify_solana_message(&solana_message);
+                    println!("Solana payload: {payload:?}\n");
+                }
+
+                if let Some(parsed) = update.payload.parsed {
+                    // Parsed payloads (`parsed: true`) are already decoded and ready to use
+                    for feed in parsed.price_feeds {
+                        println!(
+                            "Parsed payload: {:?}: {:?} at {:?}\n",
+                            feed.price_feed_id, feed, parsed.timestamp_us
+                        );
+                    }
+                }
+            }
+            _ => println!("Received non-update message"),
+        }
+
         count += 1;
         if count >= 50 {
             break;
@@ -92,3 +132,27 @@ async fn main() -> anyhow::Result<()> {
     client.close().await?;
     Ok(())
 }
+
+fn parse_and_verify_solana_message(solana_message: &SolanaMessage) -> anyhow::Result<PayloadData> {
+    // Verify signature using the pubkey
+    let public_key = ed25519_dalek::VerifyingKey::from_bytes(&solana_message.public_key)?;
+    public_key.verify_strict(
+        &solana_message.payload,
+        &ed25519_dalek::Signature::from_bytes(&solana_message.signature),
+    )?;
+
+    let payload = PayloadData::deserialize_slice_le(&solana_message.payload)?;
+    Ok(payload)
+}
+
+fn parse_and_verify_evm_message(evm_message: &EvmMessage) -> anyhow::Result<PayloadData> {
+    // Recover pubkey from message
+    libsecp256k1::recover(
+        &libsecp256k1::Message::parse(&alloy_primitives::keccak256(&evm_message.payload)),
+        &libsecp256k1::Signature::parse_standard(&evm_message.signature)?,
+        &libsecp256k1::RecoveryId::parse(evm_message.recovery_id)?,
+    )?;
+
+    let payload = PayloadData::deserialize_slice_be(&evm_message.payload)?;
+    Ok(payload)
+}

lazer/sdk/rust/client/examples/subscribe_price_feeds.rs

@@ -3,7 +3,6 @@
 
 use {
     crate::{
-        message::{EvmMessage, SolanaMessage},
         payload::{
             BINARY_UPDATE_FORMAT_MAGIC, EVM_FORMAT_MAGIC, PARSED_FORMAT_MAGIC,
             SOLANA_FORMAT_MAGIC_BE,
@@ -79,10 +78,18 @@ impl Response {
 
             match magic {
                 EVM_FORMAT_MAGIC => {
-                    evm = Some(EvmMessage::deserialize_slice(&data[pos..pos + len])?);
+                    evm = Some(JsonBinaryData {
+                        encoding: JsonBinaryEncoding::Base64,
+                        data: base64::engine::general_purpose::STANDARD
+                            .encode(&data[pos..pos + len]),
+                    });
                 }
                 SOLANA_FORMAT_MAGIC_BE => {
-                    solana = Some(SolanaMessage::deserialize_slice(&data[pos..pos + len])?);
+                    solana = Some(JsonBinaryData {
+                        encoding: JsonBinaryEncoding::Base64,
+                        data: base64::engine::general_purpose::STANDARD
+                            .encode(&data[pos..pos + len]),
+                    });
                 }
                 PARSED_FORMAT_MAGIC => {
                     parsed = Some(serde_json::from_slice(&data[pos + 4..pos + len])?);
@@ -95,19 +102,14 @@ impl Response {
         Ok(Response::StreamUpdated(StreamUpdatedResponse {
             subscription_id,
             payload: JsonUpdate {
-                evm: evm.map(|m| JsonBinaryData {
-                    encoding: JsonBinaryEncoding::Base64,
-                    data: base64::engine::general_purpose::STANDARD.encode(&m.payload),
-                }),
-                solana: solana.map(|m| JsonBinaryData {
-                    encoding: JsonBinaryEncoding::Base64,
-                    data: base64::engine::general_purpose::STANDARD.encode(&m.payload),
-                }),
+                evm,
+                solana,
                 parsed,
             },
         }))
     }
 }
+
 /// Sent from the server after a successul subscription.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]