Implement Cryptography API

.github/workflows/validate-examples.yml

@@ -22,7 +22,7 @@ on:
         required: false
         default: ""
   repository_dispatch:
-    types: [validate-examples]
+    types: [ validate-examples ]
   merge_group:
 jobs:
   setup:
@@ -144,7 +144,7 @@ jobs:
       fail-fast: false
       matrix:
         examples:
-          ["actors", "client", "configuration", "invoke/grpc", "invoke/grpc-proxying", "pubsub", "secrets-bulk"]
+          [ "actors", "client", "configuration", "crypto", "invoke/grpc", "invoke/grpc-proxying", "pubsub", "secrets-bulk" ]
     steps:
       - name: Check out code
         uses: actions/checkout@v4

Cargo.toml

@@ -23,6 +23,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 axum = "0.7.4"
 tokio = { version = "1.29", features = ["sync"] }
+tokio-util = { version = "0.7.10", features = ["io"] }
 chrono = "0.4.24"
 
 [build-dependencies]
@@ -45,13 +46,17 @@ path = "examples/actors/client.rs"
 name = "actor-server"
 path = "examples/actors/server.rs"
 
+[[example]]
+name = "client"
+path = "examples/client/client.rs"
+
 [[example]]
 name = "configuration"
 path = "examples/configuration/main.rs"
 
 [[example]]
-name = "client"
-path = "examples/client/client.rs"
+name = "crypto"
+path = "examples/crypto/main.rs"
 
 [[example]]
 name = "invoke-grpc-client"

examples/crypto/README.md

@@ -0,0 +1,48 @@
+# Crypto Example
+
+This is a simple example that demonstrates Dapr's Cryptography capabilities.
+
+> **Note:** Make sure to use latest version of proto bindings.
+
+## Running
+
+To run this example:
+
+1. Generate keys in examples/crypto/keys directory:
+<!-- STEP
+name: Generate keys
+background: false
+sleep: 5
+timeout_seconds: 30
+-->
+```bash
+mkdir -p keys
+# Generate a private RSA key, 4096-bit keys
+openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:4096 -out keys/rsa-private-key.pem
+# Generate a 256-bit key for AES
+openssl rand -out keys/symmetric-key-256 32
+```
+
+<!-- END_STEP -->
+
+2. Run the multi-app run template:
+
+<!-- STEP
+name: Run multi-app
+output_match_mode: substring
+match_order: none
+expected_stdout_lines:
+  - '== APP - crypto-example == Successfully Decrypted String'
+  - '== APP - crypto-example == Successfully Decrypted Image'
+background: true
+sleep: 30
+timeout_seconds: 90
+-->
+
+```bash
+dapr run -f .
+```
+
+<!-- END_STEP -->
+
+2. Stop with `ctrl + c`

examples/crypto/components/local-storage.yml

@@ -0,0 +1,11 @@
+apiVersion: dapr.io/v1alpha1
+kind: Component
+metadata:
+  name: localstorage
+spec:
+  type: crypto.dapr.localstorage
+  version: v1
+  metadata:
+    - name: path
+      # Path is relative to the folder where the example is located
+      value: ./keys

examples/crypto/dapr.yaml

@@ -0,0 +1,10 @@
+version: 1
+common:
+  daprdLogDestination: console
+apps:
+  - appID: crypto-example
+    appDirPath: ./
+    daprGRPCPort: 35002
+    logLevel: debug
+    command: [ "cargo", "run", "--example", "crypto" ]
+    resourcesPath: ./components

examples/crypto/main.rs

@@ -0,0 +1,82 @@
+use std::fs;
+
+use tokio::fs::File;
+use tokio::time::sleep;
+
+use dapr::client::ReaderStream;
+use dapr::dapr::dapr::proto::runtime::v1::{DecryptRequestOptions, EncryptRequestOptions};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    sleep(std::time::Duration::new(2, 0)).await;
+    let port: u16 = std::env::var("DAPR_GRPC_PORT")?.parse()?;
+    let addr = format!("https://127.0.0.1:{}", port);
+
+    let mut client = dapr::Client::<dapr::client::TonicClient>::connect(addr).await?;
+
+    let encrypted = client
+        .encrypt(
+            ReaderStream::new("Test".as_bytes()),
+            EncryptRequestOptions {
+                component_name: "localstorage".to_string(),
+                key_name: "rsa-private-key.pem".to_string(),
+                key_wrap_algorithm: "RSA".to_string(),
+                data_encryption_cipher: "aes-gcm".to_string(),
+                omit_decryption_key_name: false,
+                decryption_key_name: "rsa-private-key.pem".to_string(),
+            },
+        )
+        .await
+        .unwrap();
+
+    let decrypted = client
+        .decrypt(
+            encrypted,
+            DecryptRequestOptions {
+                component_name: "localstorage".to_string(),
+                key_name: "rsa-private-key.pem".to_string(),
+            },
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(String::from_utf8(decrypted).unwrap().as_str(), "Test");
+
+    println!("Successfully Decrypted String");
+
+    let image = File::open("./image.png").await.unwrap();
+
+    let encrypted = client
+        .encrypt(
+            ReaderStream::new(image),
+            EncryptRequestOptions {
+                component_name: "localstorage".to_string(),
+                key_name: "rsa-private-key.pem".to_string(),
+                key_wrap_algorithm: "RSA".to_string(),
+                data_encryption_cipher: "aes-gcm".to_string(),
+                omit_decryption_key_name: false,
+                decryption_key_name: "rsa-private-key.pem".to_string(),
+            },
+        )
+        .await
+        .unwrap();
+
+    let decrypted = client
+        .decrypt(
+            encrypted,
+            DecryptRequestOptions {
+                component_name: "localstorage".to_string(),
+                key_name: "rsa-private-key.pem".to_string(),
+            },
+        )
+        .await
+        .unwrap();
+
+    let image = fs::read("./image.png").unwrap();
+
+    assert_eq!(decrypted, image);
+
+    println!("Successfully Decrypted Image");
+
+    Ok(())
+}

src/client.rs

@@ -1,12 +1,16 @@
-use crate::dapr::dapr::proto::{common::v1 as common_v1, runtime::v1 as dapr_v1};
-use prost_types::Any;
 use std::collections::HashMap;
-use tonic::Streaming;
-use tonic::{transport::Channel as TonicChannel, Request};
 
-use crate::error::Error;
 use async_trait::async_trait;
+use futures::StreamExt;
+use prost_types::Any;
 use serde::{Deserialize, Serialize};
+use tokio::io::AsyncRead;
+use tonic::codegen::tokio_stream;
+use tonic::{transport::Channel as TonicChannel, Request};
+use tonic::{Status, Streaming};
+
+use crate::dapr::dapr::proto::{common::v1 as common_v1, runtime::v1 as dapr_v1};
+use crate::error::Error;
 
 #[derive(Clone)]
 pub struct Client<T>(T);
@@ -379,6 +383,78 @@ impl<T: DaprInterface> Client<T> {
         };
         self.0.unsubscribe_configuration(request).await
     }
+
+    /// Encrypt binary data using Dapr. returns Vec<StreamPayload> to be used in decrypt method
+    ///
+    /// # Arguments
+    ///
+    /// * `payload` - ReaderStream to the data to encrypt
+    /// * `request_option` - Encryption request options.
+    pub async fn encrypt<R>(
+        &mut self,
+        payload: ReaderStream<R>,
+        request_options: EncryptRequestOptions,
+    ) -> Result<Vec<StreamPayload>, Status>
+    where
+        R: AsyncRead + Send,
+    {
+        // have to have it as a reference for the async move below
+        let request_options = &Some(request_options);
+        let requested_items: Vec<EncryptRequest> = payload
+            .0
+            .enumerate()
+            .fold(vec![], |mut init, (i, bytes)| async move {
+                let stream_payload = StreamPayload {
+                    data: bytes.unwrap().to_vec(),
+                    seq: 0,
+                };
+                if i == 0 {
+                    init.push(EncryptRequest {
+                        options: request_options.clone(),
+                        payload: Some(stream_payload),
+                    });
+                } else {
+                    init.push(EncryptRequest {
+                        options: None,
+                        payload: Some(stream_payload),
+                    });
+                }
+                init
+            })
+            .await;
+        self.0.encrypt(requested_items).await
+    }
+
+    /// Decrypt binary data using Dapr. returns Vec<u8>.
+    ///
+    /// # Arguments
+    ///
+    /// * `encrypted` - Encrypted data usually returned from encrypted, Vec<StreamPayload>
+    /// * `options` - Decryption request options.
+    pub async fn decrypt(
+        &mut self,
+        encrypted: Vec<StreamPayload>,
+        options: DecryptRequestOptions,
+    ) -> Result<Vec<u8>, Status> {
+        let requested_items: Vec<DecryptRequest> = encrypted
+            .iter()
+            .enumerate()
+            .map(|(i, item)| {
+                if i == 0 {
+                    DecryptRequest {
+                        options: Some(options.clone()),
+                        payload: Some(item.clone()),
+                    }
+                } else {
+                    DecryptRequest {
+                        options: None,
+                        payload: Some(item.clone()),
+                    }
+                }
+            })
+            .collect();
+        self.0.decrypt(requested_items).await
+    }
 }
 
 #[async_trait]
@@ -420,6 +496,11 @@ pub trait DaprInterface: Sized {
         &mut self,
         request: UnsubscribeConfigurationRequest,
     ) -> Result<UnsubscribeConfigurationResponse, Error>;
+
+    async fn encrypt(&mut self, payload: Vec<EncryptRequest>)
+        -> Result<Vec<StreamPayload>, Status>;
+
+    async fn decrypt(&mut self, payload: Vec<DecryptRequest>) -> Result<Vec<u8>, Status>;
 }
 
 #[async_trait]
@@ -535,6 +616,51 @@ impl DaprInterface for dapr_v1::dapr_client::DaprClient<TonicChannel> {
             .await?
             .into_inner())
     }
+
+    /// Encrypt binary data using Dapr. returns Vec<StreamPayload> to be used in decrypt method
+    ///
+    /// # Arguments
+    ///
+    /// * `payload` - ReaderStream to the data to encrypt
+    /// * `request_option` - Encryption request options.
+    async fn encrypt(
+        &mut self,
+        request: Vec<EncryptRequest>,
+    ) -> Result<Vec<StreamPayload>, Status> {
+        let request = Request::new(tokio_stream::iter(request));
+        let stream = self.encrypt_alpha1(request).await?;
+        let mut stream = stream.into_inner();
+        let mut return_data = vec![];
+        while let Some(resp) = stream.next().await {
+            if let Ok(resp) = resp {
+                if let Some(data) = resp.payload {
+                    return_data.push(data)
+                }
+            }
+        }
+        Ok(return_data)
+    }
+
+    /// Decrypt binary data using Dapr. returns Vec<u8>.
+    ///
+    /// # Arguments
+    ///
+    /// * `encrypted` - Encrypted data usually returned from encrypted, Vec<StreamPayload>
+    /// * `options` - Decryption request options.
+    async fn decrypt(&mut self, request: Vec<DecryptRequest>) -> Result<Vec<u8>, Status> {
+        let request = Request::new(tokio_stream::iter(request));
+        let stream = self.decrypt_alpha1(request).await?;
+        let mut stream = stream.into_inner();
+        let mut data = vec![];
+        while let Some(resp) = stream.next().await {
+            if let Ok(resp) = resp {
+                if let Some(mut payload) = resp.payload {
+                    data.append(payload.data.as_mut())
+                }
+            }
+        }
+        Ok(data)
+    }
 }
 
 /// A request from invoking a service
@@ -614,6 +740,19 @@ pub type UnsubscribeConfigurationResponse = dapr_v1::UnsubscribeConfigurationRes
 /// A tonic based gRPC client
 pub type TonicClient = dapr_v1::dapr_client::DaprClient<TonicChannel>;
 
+/// Encryption gRPC request
+type EncryptRequest = crate::dapr::dapr::proto::runtime::v1::EncryptRequest;
+
+/// Decrypt gRPC request
+type DecryptRequest = crate::dapr::dapr::proto::runtime::v1::DecryptRequest;
+
+/// Encryption request options
+type EncryptRequestOptions = crate::dapr::dapr::proto::runtime::v1::EncryptRequestOptions;
+
+/// Decryption request options
+type DecryptRequestOptions = crate::dapr::dapr::proto::runtime::v1::DecryptRequestOptions;
+
+type StreamPayload = crate::dapr::dapr::proto::common::v1::StreamPayload;
 impl<K> From<(K, Vec<u8>)> for common_v1::StateItem
 where
     K: Into<String>,
@@ -626,3 +765,11 @@ where
         }
     }
 }
+
+pub struct ReaderStream<T>(tokio_util::io::ReaderStream<T>);
+
+impl<T: AsyncRead> ReaderStream<T> {
+    pub fn new(data: T) -> Self {
+        ReaderStream(tokio_util::io::ReaderStream::new(data))
+    }
+}