chore: remove sha256 digests (#254)

src/lib.rs

@@ -42,6 +42,7 @@ pub mod services;
 pub mod spec;
 pub mod state;
 pub mod types;
+pub mod utils;
 
 build_info::build_info!(fn build_info);
 

src/notify_keys.rs

@@ -1,5 +1,8 @@
 use {
-    crate::error::{Error, Result},
+    crate::{
+        error::{Error, Result},
+        utils::topic_from_key,
+    },
     rand_chacha::{
         rand_core::{RngCore, SeedableRng},
         ChaCha20Rng,
@@ -42,7 +45,7 @@ impl NotifyKeys {
             domain,
             key_agreement_secret,
             key_agreement_public,
-            key_agreement_topic: Topic::from(sha256::digest(key_agreement_public.as_bytes())),
+            key_agreement_topic: topic_from_key(key_agreement_public.as_bytes()),
             authentication_secret,
             authentication_public,
         })

src/services/public_http_server/handlers/subscribe_topic.rs

@@ -5,13 +5,14 @@ use {
         rate_limit,
         registry::{extractor::AuthedProjectId, storage::redis::Redis},
         state::AppState,
+        utils::topic_from_key,
     },
     axum::{self, extract::State, response::IntoResponse, Json},
     chacha20poly1305::aead::OsRng,
     hyper::StatusCode,
     once_cell::sync::Lazy,
     regex::Regex,
-    relay_rpc::domain::{ProjectId, Topic},
+    relay_rpc::domain::ProjectId,
     serde::{Deserialize, Serialize},
     serde_json::json,
     std::sync::Arc,
@@ -69,7 +70,7 @@ pub async fn handler(
 
     let subscribe_key = StaticSecret::random_from_rng(OsRng);
     let signing_public = PublicKey::from(&subscribe_key);
-    let topic: Topic = sha256::digest(signing_public.as_bytes()).into();
+    let topic = topic_from_key(signing_public.as_bytes());
 
     let authentication_key = ed25519_dalek::SigningKey::generate(&mut OsRng);
 

src/services/publisher_service/mod.rs

@@ -9,13 +9,14 @@ use {
         services::websocket_server::decode_key,
         spec::{NOTIFY_MESSAGE_TAG, NOTIFY_MESSAGE_TTL},
         types::{Envelope, EnvelopeType0},
+        utils::topic_from_key,
     },
     base64::Engine,
     chrono::{DateTime, Utc},
     helpers::{dead_letter_give_up_check, update_message_processing_status},
     relay_client::http::Client,
     relay_rpc::{
-        domain::{DecodedClientId, Topic},
+        domain::DecodedClientId,
         rpc::{msg_id::MsgId, Publish, JSON_RPC_VERSION_STR},
     },
     sqlx::{postgres::PgListener, PgPool},
@@ -314,7 +315,7 @@ async fn process_notification(
     let sym_key = decode_key(&notification.subscriber_sym_key)?;
     let envelope = Envelope::<EnvelopeType0>::new(&sym_key, &message)?;
     let base64_notification = base64::engine::general_purpose::STANDARD.encode(envelope.to_bytes());
-    let topic = Topic::new(sha256::digest(&sym_key).into());
+    let topic = topic_from_key(&sym_key);
 
     let publish = Publish {
         topic,

src/services/websocket_server/handlers/notify_delete.rs

@@ -18,6 +18,7 @@ use {
         spec::{NOTIFY_DELETE_RESPONSE_TAG, NOTIFY_DELETE_RESPONSE_TTL},
         state::{AppState, WebhookNotificationEvent},
         types::{Envelope, EnvelopeType0},
+        utils::topic_from_key,
         Result,
     },
     base64::Engine,
@@ -154,12 +155,12 @@ pub async fn handle(msg: PublishedMessage, state: &AppState, client: &Client) ->
 
     let base64_notification = base64::engine::general_purpose::STANDARD.encode(envelope.to_bytes());
 
-    let response_topic = sha256::digest(&sym_key);
+    let response_topic = topic_from_key(&sym_key);
 
     publish_relay_message(
         &state.relay_http_client,
         &Publish {
-            topic: response_topic.into(),
+            topic: response_topic,
             message: base64_notification.into(),
             tag: NOTIFY_DELETE_RESPONSE_TAG,
             ttl_secs: NOTIFY_DELETE_RESPONSE_TTL.as_secs() as u32,

src/services/websocket_server/handlers/notify_subscribe.rs

@@ -18,6 +18,7 @@ use {
         spec::{NOTIFY_NOOP_TAG, NOTIFY_SUBSCRIBE_RESPONSE_TAG, NOTIFY_SUBSCRIBE_RESPONSE_TTL},
         state::{AppState, WebhookNotificationEvent},
         types::{parse_scope, Envelope, EnvelopeType0, EnvelopeType1},
+        utils::topic_from_key,
         Result,
     },
     base64::Engine,
@@ -66,7 +67,7 @@ pub async fn handle(msg: PublishedMessage, state: &AppState) -> Result<()> {
         &client_public_key,
         &x25519_dalek::StaticSecret::from(decode_key(&project.subscribe_private_key)?),
     )?;
-    let response_topic = sha256::digest(&sym_key);
+    let response_topic = topic_from_key(&sym_key);
     info!("response_topic: {response_topic}");
 
     let msg: NotifyRequest<NotifySubscribe> = decrypt_message(envelope, &sym_key)?;
@@ -149,7 +150,7 @@ pub async fn handle(msg: PublishedMessage, state: &AppState) -> Result<()> {
 
     let scope = parse_scope(&sub_auth.scp)?;
 
-    let notify_topic: Topic = sha256::digest(&notify_key).into();
+    let notify_topic = topic_from_key(&notify_key);
 
     let project_id = project.project_id;
     info!(
@@ -222,7 +223,7 @@ pub async fn handle(msg: PublishedMessage, state: &AppState) -> Result<()> {
     publish_relay_message(
         &state.relay_http_client,
         &Publish {
-            topic: response_topic.into(),
+            topic: response_topic,
             message: base64_notification.into(),
             tag: NOTIFY_SUBSCRIBE_RESPONSE_TAG,
             ttl_secs: NOTIFY_SUBSCRIBE_RESPONSE_TTL.as_secs() as u32,

src/services/websocket_server/handlers/notify_update.rs

@@ -18,6 +18,7 @@ use {
         spec::{NOTIFY_UPDATE_RESPONSE_TAG, NOTIFY_UPDATE_RESPONSE_TTL},
         state::AppState,
         types::{parse_scope, Envelope, EnvelopeType0},
+        utils::topic_from_key,
         Result,
     },
     base64::Engine,
@@ -159,12 +160,12 @@ pub async fn handle(msg: PublishedMessage, state: &AppState) -> Result<()> {
 
     let base64_notification = base64::engine::general_purpose::STANDARD.encode(envelope.to_bytes());
 
-    let response_topic = sha256::digest(&sym_key);
+    let response_topic = topic_from_key(&sym_key);
 
     publish_relay_message(
         &state.relay_http_client,
         &Publish {
-            topic: response_topic.into(),
+            topic: response_topic,
             message: base64_notification.into(),
             tag: NOTIFY_UPDATE_RESPONSE_TAG,
             ttl_secs: NOTIFY_UPDATE_RESPONSE_TTL.as_secs() as u32,

src/services/websocket_server/handlers/notify_watch_subscriptions.rs

@@ -29,13 +29,14 @@ use {
         },
         state::AppState,
         types::{Envelope, EnvelopeType0, EnvelopeType1},
+        utils::topic_from_key,
         Result,
     },
     base64::Engine,
     chrono::{Duration, Utc},
     relay_client::websocket::PublishedMessage,
     relay_rpc::{
-        domain::{DecodedClientId, Topic},
+        domain::DecodedClientId,
         rpc::{Publish, JSON_RPC_VERSION_STR},
     },
     serde_json::{json, Value},
@@ -63,7 +64,7 @@ pub async fn handle(msg: PublishedMessage, state: &AppState) -> Result<()> {
     }
 
     let response_sym_key = derive_key(&client_public_key, &state.notify_keys.key_agreement_secret)?;
-    let response_topic = sha256::digest(&response_sym_key);
+    let response_topic = topic_from_key(&response_sym_key);
 
     let msg: NotifyRequest<NotifyWatchSubscriptions> =
         decrypt_message(envelope, &response_sym_key)?;
@@ -171,7 +172,7 @@ pub async fn handle(msg: PublishedMessage, state: &AppState) -> Result<()> {
         publish_relay_message(
             &state.relay_http_client,
             &Publish {
-                topic: response_topic.into(),
+                topic: response_topic,
                 message: base64_notification.into(),
                 tag: NOTIFY_WATCH_SUBSCRIPTIONS_RESPONSE_TAG,
                 ttl_secs: NOTIFY_WATCH_SUBSCRIPTIONS_RESPONSE_TTL.as_secs() as u32,
@@ -303,7 +304,7 @@ pub async fn update_subscription_watchers(
         let base64_notification =
             base64::engine::general_purpose::STANDARD.encode(envelope.to_bytes());
 
-        let topic = Topic::from(sha256::digest(&sym_key));
+        let topic = topic_from_key(&sym_key);
         publish_relay_message(
             http_client,
             &Publish {

src/services/websocket_server/mod.rs

@@ -17,7 +17,7 @@ use {
     relay_client::websocket::{Client, PublishedMessage},
     relay_rpc::{
         domain::{MessageId, Topic},
-        rpc::{JSON_RPC_VERSION_STR, MAX_SUBSCRIPTION_BATCH_SIZE},
+        rpc::{msg_id::get_message_id, JSON_RPC_VERSION_STR, MAX_SUBSCRIPTION_BATCH_SIZE},
     },
     relay_ws_client::RelayClientEvent,
     serde::{Deserialize, Serialize},
@@ -98,7 +98,7 @@ pub async fn start(
     }
 }
 
-#[instrument(skip_all, fields(topic = %msg.topic, tag = %msg.tag, message_id = %sha256::digest(msg.message.as_bytes())))]
+#[instrument(skip_all, fields(topic = %msg.topic, tag = %msg.tag, message_id = %get_message_id(&msg.message)))]
 async fn handle_msg(msg: PublishedMessage, state: &AppState, client: &Client) {
     let start = Instant::now();
     let topic = msg.topic.clone();

src/utils.rs

@@ -0,0 +1,6 @@
+use relay_rpc::domain::Topic;
+
+// TODO consider using the key object directly instead of a byte slice
+pub fn topic_from_key(key: &[u8]) -> Topic {
+    sha256::digest(key).into()
+}

tests/deployment.rs

@@ -41,6 +41,7 @@ use {
             NOTIFY_WATCH_SUBSCRIPTIONS_TTL,
         },
         types::{Envelope, EnvelopeType0, EnvelopeType1, Notification},
+        utils::topic_from_key,
     },
     rand::{rngs::StdRng, SeedableRng},
     relay_rpc::{
@@ -49,6 +50,7 @@ use {
             ed25519_dalek::Keypair,
         },
         domain::DecodedClientId,
+        rpc::msg_id::get_message_id,
     },
     serde_json::json,
     sha2::Digest,
@@ -255,7 +257,7 @@ async fn watch_subscriptions(
 
     let response_topic_key =
         derive_key(&x25519_dalek::PublicKey::from(key_agreement_key), &secret).unwrap();
-    let response_topic = sha256::digest(&response_topic_key);
+    let response_topic = topic_from_key(&response_topic_key);
 
     let envelope = Envelope::<EnvelopeType1>::new(
         &response_topic_key,
@@ -265,10 +267,10 @@ async fn watch_subscriptions(
     .unwrap();
     let message = base64::engine::general_purpose::STANDARD.encode(envelope.to_bytes());
 
-    let watch_subscriptions_topic = sha256::digest(&key_agreement_key);
+    let watch_subscriptions_topic = topic_from_key(&key_agreement_key);
     relay_ws_client
         .publish(
-            watch_subscriptions_topic.into(),
+            watch_subscriptions_topic,
             message,
             NOTIFY_WATCH_SUBSCRIPTIONS_TAG,
             NOTIFY_WATCH_SUBSCRIPTIONS_TTL,
@@ -278,7 +280,7 @@ async fn watch_subscriptions(
         .unwrap();
 
     relay_ws_client
-        .subscribe(response_topic.clone().into())
+        .subscribe(response_topic.clone())
         .await
         .unwrap();
 
@@ -304,7 +306,7 @@ async fn watch_subscriptions(
     println!(
         "received watch_subscriptions_response with id msg.id {} and message_id {} and RPC ID {}",
         msg.message_id,
-        sha256::digest(msg.message.as_ref()),
+        get_message_id(&msg.message),
         response.id,
     );
 
@@ -469,7 +471,7 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
     .to_did_key();
 
     // Get subscribe topic for dapp
-    let subscribe_topic = sha256::digest(hex::decode(app_subscribe_public_key).unwrap().as_slice());
+    let subscribe_topic = topic_from_key(hex::decode(app_subscribe_public_key).unwrap().as_slice());
 
     // ----------------------------------------------------
     // SUBSCRIBE WALLET CLIENT TO DAPP THROUGHT NOTIFY
@@ -524,19 +526,19 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
     let message = base64::engine::general_purpose::STANDARD.encode(envelope.to_bytes());
 
     // Get response topic for wallet client and notify communication
-    let response_topic = sha256::digest(&response_topic_key);
+    let response_topic = topic_from_key(&response_topic_key);
     println!("subscription response_topic: {response_topic}");
 
     // Subscribe to the topic and listen for response
     relay_ws_client
-        .subscribe(response_topic.clone().into())
+        .subscribe(response_topic.clone())
         .await
         .unwrap();
 
     // Send subscription request to notify
     relay_ws_client
         .publish(
-            subscribe_topic.into(),
+            subscribe_topic,
             message,
             NOTIFY_SUBSCRIBE_TAG,
             NOTIFY_SUBSCRIBE_TTL,
@@ -557,7 +559,7 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
             "got additional message with unexpected tag {} msg.id {} and message_id {}",
             msg.tag,
             msg.message_id,
-            sha256::digest(msg.message.as_ref()),
+            get_message_id(&msg.message),
         );
         let Envelope::<EnvelopeType0> { sealbox, iv, .. } = Envelope::<EnvelopeType0>::from_bytes(
             base64::engine::general_purpose::STANDARD
@@ -574,7 +576,7 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
             "warn: got additional message with unexpected tag {} msg.id {} and message_id {} RPC ID {}",
             msg.tag,
             msg.message_id,
-            sha256::digest(msg.message.as_ref()),
+            get_message_id(&msg.message),
             response.id,
         );
 
@@ -658,10 +660,10 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
         decode_key(&sub.sym_key).unwrap()
     };
 
-    let notify_topic = sha256::digest(&notify_key);
+    let notify_topic = topic_from_key(&notify_key);
 
     relay_ws_client
-        .subscribe(notify_topic.clone().into())
+        .subscribe(notify_topic.clone())
         .await
         .unwrap();
 
@@ -785,7 +787,7 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
 
     relay_ws_client
         .publish(
-            notify_topic.clone().into(),
+            notify_topic.clone(),
             encoded_message,
             NOTIFY_UPDATE_TAG,
             NOTIFY_UPDATE_TTL,
@@ -886,7 +888,6 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
 
     // Encode the subscription auth
     let delete_auth = encode_auth(&delete_auth, &identity_signing_key);
-    let _delete_auth_hash = sha256::digest(&*delete_auth.clone());
 
     let sub_auth = json!({ "deleteAuth": delete_auth });
 
@@ -898,7 +899,7 @@ async fn run_test(statement: String, watch_subscriptions_all_domains: bool) {
 
     relay_ws_client
         .publish(
-            notify_topic.into(),
+            notify_topic,
             encoded_message,
             NOTIFY_DELETE_TAG,
             NOTIFY_DELETE_TTL,