Skip to content

Commit 8722b24

Browse files
siri-varmasiri-varma
committed
Add crypto
1 parent 8cb7c04 commit 8722b24

File tree

4 files changed

+446
-0
lines changed

4 files changed

+446
-0
lines changed

‎sdk/src/main/java/io/dapr/client/DaprClientImpl.java‎

Lines changed: 192 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,9 +48,11 @@
4848
import io.dapr.client.domain.ConversationTools;
4949
import io.dapr.client.domain.ConversationToolsFunction;
5050
import io.dapr.client.domain.DaprMetadata;
51+
import io.dapr.client.domain.DecryptRequestAlpha1;
5152
import io.dapr.client.domain.DeleteJobRequest;
5253
import io.dapr.client.domain.DeleteStateRequest;
5354
import io.dapr.client.domain.DropFailurePolicy;
55+
import io.dapr.client.domain.EncryptRequestAlpha1;
5456
import io.dapr.client.domain.ExecuteStateTransactionRequest;
5557
import io.dapr.client.domain.FailurePolicy;
5658
import io.dapr.client.domain.FailurePolicyType;
@@ -2071,4 +2073,194 @@ private AppConnectionPropertiesHealthMetadata getAppConnectionPropertiesHealth(
20712073
return new AppConnectionPropertiesHealthMetadata(healthCheckPath, healthProbeInterval, healthProbeTimeout,
20722074
healthThreshold);
20732075
}
2076+
2077+
/**
2078+
* {@inheritDoc}
2079+
*/
2080+
@Override
2081+
public Flux<byte[]> encrypt(EncryptRequestAlpha1 request) {
2082+
try {
2083+
if (request == null) {
2084+
throw new IllegalArgumentException("EncryptRequestAlpha1 cannot be null.");
2085+
}
2086+
if (request.getComponentName() == null || request.getComponentName().trim().isEmpty()) {
2087+
throw new IllegalArgumentException("Component name cannot be null or empty.");
2088+
}
2089+
if (request.getKeyName() == null || request.getKeyName().trim().isEmpty()) {
2090+
throw new IllegalArgumentException("Key name cannot be null or empty.");
2091+
}
2092+
if (request.getKeyWrapAlgorithm() == null || request.getKeyWrapAlgorithm().trim().isEmpty()) {
2093+
throw new IllegalArgumentException("Key wrap algorithm cannot be null or empty.");
2094+
}
2095+
if (request.getPlainTextStream() == null) {
2096+
throw new IllegalArgumentException("Plaintext stream cannot be null.");
2097+
}
2098+
2099+
return Flux.create(sink -> {
2100+
// Create response observer to receive encrypted data
2101+
StreamObserver<DaprProtos.EncryptResponse> responseObserver = new StreamObserver<DaprProtos.EncryptResponse>() {
2102+
@Override
2103+
public void onNext(DaprProtos.EncryptResponse response) {
2104+
if (response.hasPayload()) {
2105+
byte[] data = response.getPayload().getData().toByteArray();
2106+
if (data.length > 0) {
2107+
sink.next(data);
2108+
}
2109+
}
2110+
}
2111+
2112+
@Override
2113+
public void onError(Throwable t) {
2114+
sink.error(DaprException.propagate(new DaprException("ENCRYPT_ERROR",
2115+
"Error during encryption: " + t.getMessage(), t)));
2116+
}
2117+
2118+
@Override
2119+
public void onCompleted() {
2120+
sink.complete();
2121+
}
2122+
};
2123+
2124+
// Get the request stream observer from gRPC
2125+
StreamObserver<DaprProtos.EncryptRequest> requestObserver =
2126+
intercept(null, asyncStub).encryptAlpha1(responseObserver);
2127+
2128+
// Build options for the first message
2129+
DaprProtos.EncryptRequestOptions.Builder optionsBuilder = DaprProtos.EncryptRequestOptions.newBuilder()
2130+
.setComponentName(request.getComponentName())
2131+
.setKeyName(request.getKeyName())
2132+
.setKeyWrapAlgorithm(request.getKeyWrapAlgorithm());
2133+
2134+
if (request.getDataEncryptionCipher() != null && !request.getDataEncryptionCipher().isEmpty()) {
2135+
optionsBuilder.setDataEncryptionCipher(request.getDataEncryptionCipher());
2136+
}
2137+
optionsBuilder.setOmitDecryptionKeyName(request.isOmitDecryptionKeyName());
2138+
if (request.getDecryptionKeyName() != null && !request.getDecryptionKeyName().isEmpty()) {
2139+
optionsBuilder.setDecryptionKeyName(request.getDecryptionKeyName());
2140+
}
2141+
2142+
final DaprProtos.EncryptRequestOptions options = optionsBuilder.build();
2143+
final long[] sequenceNumber = {0};
2144+
final boolean[] firstMessage = {true};
2145+
2146+
// Subscribe to the plaintext stream and send chunks
2147+
request.getPlainTextStream()
2148+
.doOnNext(chunk -> {
2149+
DaprProtos.EncryptRequest.Builder reqBuilder = DaprProtos.EncryptRequest.newBuilder()
2150+
.setPayload(CommonProtos.StreamPayload.newBuilder()
2151+
.setData(ByteString.copyFrom(chunk))
2152+
.setSeq(sequenceNumber[0]++)
2153+
.build());
2154+
2155+
// Include options only in the first message
2156+
if (firstMessage[0]) {
2157+
reqBuilder.setOptions(options);
2158+
firstMessage[0] = false;
2159+
}
2160+
2161+
requestObserver.onNext(reqBuilder.build());
2162+
})
2163+
.doOnError(error -> {
2164+
requestObserver.onError(error);
2165+
sink.error(DaprException.propagate(new DaprException("ENCRYPT_ERROR",
2166+
"Error reading plaintext stream: " + error.getMessage(), error)));
2167+
})
2168+
.doOnComplete(() -> {
2169+
requestObserver.onCompleted();
2170+
})
2171+
.subscribe();
2172+
});
2173+
} catch (Exception ex) {
2174+
return DaprException.wrapFlux(ex);
2175+
}
2176+
}
2177+
2178+
/**
2179+
* {@inheritDoc}
2180+
*/
2181+
@Override
2182+
public Flux<byte[]> decrypt(DecryptRequestAlpha1 request) {
2183+
try {
2184+
if (request == null) {
2185+
throw new IllegalArgumentException("DecryptRequestAlpha1 cannot be null.");
2186+
}
2187+
if (request.getComponentName() == null || request.getComponentName().trim().isEmpty()) {
2188+
throw new IllegalArgumentException("Component name cannot be null or empty.");
2189+
}
2190+
if (request.getCipherTextStream() == null) {
2191+
throw new IllegalArgumentException("Ciphertext stream cannot be null.");
2192+
}
2193+
2194+
return Flux.create(sink -> {
2195+
// Create response observer to receive decrypted data
2196+
StreamObserver<DaprProtos.DecryptResponse> responseObserver = new StreamObserver<DaprProtos.DecryptResponse>() {
2197+
@Override
2198+
public void onNext(DaprProtos.DecryptResponse response) {
2199+
if (response.hasPayload()) {
2200+
byte[] data = response.getPayload().getData().toByteArray();
2201+
if (data.length > 0) {
2202+
sink.next(data);
2203+
}
2204+
}
2205+
}
2206+
2207+
@Override
2208+
public void onError(Throwable t) {
2209+
sink.error(DaprException.propagate(new DaprException("DECRYPT_ERROR",
2210+
"Error during decryption: " + t.getMessage(), t)));
2211+
}
2212+
2213+
@Override
2214+
public void onCompleted() {
2215+
sink.complete();
2216+
}
2217+
};
2218+
2219+
// Get the request stream observer from gRPC
2220+
StreamObserver<DaprProtos.DecryptRequest> requestObserver =
2221+
intercept(null, asyncStub).decryptAlpha1(responseObserver);
2222+
2223+
// Build options for the first message
2224+
DaprProtos.DecryptRequestOptions.Builder optionsBuilder = DaprProtos.DecryptRequestOptions.newBuilder()
2225+
.setComponentName(request.getComponentName());
2226+
2227+
if (request.getKeyName() != null && !request.getKeyName().isEmpty()) {
2228+
optionsBuilder.setKeyName(request.getKeyName());
2229+
}
2230+
2231+
final DaprProtos.DecryptRequestOptions options = optionsBuilder.build();
2232+
final long[] sequenceNumber = {0};
2233+
final boolean[] firstMessage = {true};
2234+
2235+
// Subscribe to the ciphertext stream and send chunks
2236+
request.getCipherTextStream()
2237+
.doOnNext(chunk -> {
2238+
DaprProtos.DecryptRequest.Builder reqBuilder = DaprProtos.DecryptRequest.newBuilder()
2239+
.setPayload(CommonProtos.StreamPayload.newBuilder()
2240+
.setData(ByteString.copyFrom(chunk))
2241+
.setSeq(sequenceNumber[0]++)
2242+
.build());
2243+
2244+
// Include options only in the first message
2245+
if (firstMessage[0]) {
2246+
reqBuilder.setOptions(options);
2247+
firstMessage[0] = false;
2248+
}
2249+
2250+
requestObserver.onNext(reqBuilder.build());
2251+
})
2252+
.doOnError(error -> {
2253+
requestObserver.onError(error);
2254+
sink.error(DaprException.propagate(new DaprException("DECRYPT_ERROR",
2255+
"Error reading ciphertext stream: " + error.getMessage(), error)));
2256+
})
2257+
.doOnComplete(() -> {
2258+
requestObserver.onCompleted();
2259+
})
2260+
.subscribe();
2261+
});
2262+
} catch (Exception ex) {
2263+
return DaprException.wrapFlux(ex);
2264+
}
2265+
}
20742266
}

‎sdk/src/main/java/io/dapr/client/DaprPreviewClient.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,9 @@
2121
import io.dapr.client.domain.ConversationRequestAlpha2;
2222
import io.dapr.client.domain.ConversationResponse;
2323
import io.dapr.client.domain.ConversationResponseAlpha2;
24+
import io.dapr.client.domain.DecryptRequestAlpha1;
2425
import io.dapr.client.domain.DeleteJobRequest;
26+
import io.dapr.client.domain.EncryptRequestAlpha1;
2527
import io.dapr.client.domain.GetJobRequest;
2628
import io.dapr.client.domain.GetJobResponse;
2729
import io.dapr.client.domain.LockRequest;
@@ -32,6 +34,7 @@
3234
import io.dapr.client.domain.UnlockResponseStatus;
3335
import io.dapr.client.domain.query.Query;
3436
import io.dapr.utils.TypeRef;
37+
import reactor.core.publisher.Flux;
3538
import reactor.core.publisher.Mono;
3639

3740
import java.util.List;
@@ -325,4 +328,24 @@ <T> Subscription subscribeToEvents(
325328
* @return {@link ConversationResponseAlpha2}.
326329
*/
327330
public Mono<ConversationResponseAlpha2> converseAlpha2(ConversationRequestAlpha2 conversationRequestAlpha2);
331+
332+
/**
333+
* Encrypt data using the Dapr cryptography building block.
334+
* This method uses streaming to handle large payloads efficiently.
335+
*
336+
* @param request The encryption request containing component name, key information, and plaintext stream.
337+
* @return A Flux of encrypted byte arrays (ciphertext chunks).
338+
* @throws IllegalArgumentException if required parameters are missing.
339+
*/
340+
Flux<byte[]> encrypt(EncryptRequestAlpha1 request);
341+
342+
/**
343+
* Decrypt data using the Dapr cryptography building block.
344+
* This method uses streaming to handle large payloads efficiently.
345+
*
346+
* @param request The decryption request containing component name, optional key name, and ciphertext stream.
347+
* @return A Flux of decrypted byte arrays (plaintext chunks).
348+
* @throws IllegalArgumentException if required parameters are missing.
349+
*/
350+
Flux<byte[]> decrypt(DecryptRequestAlpha1 request);
328351
}

‎sdk/src/main/java/io/dapr/client/domain/DecryptRequestAlpha1.java‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
/*
2+
* Copyright 2024 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package io.dapr.client.domain;
15+
16+
import reactor.core.publisher.Flux;
17+
18+
/**
19+
* Request to decrypt data using the Dapr Cryptography building block.
20+
* Uses streaming to handle large payloads efficiently.
21+
*/
22+
public class DecryptRequestAlpha1 {
23+
24+
private final String componentName;
25+
private final Flux<byte[]> cipherTextStream;
26+
private String keyName;
27+
28+
/**
29+
* Constructor for DecryptRequestAlpha1.
30+
*
31+
* @param componentName Name of the cryptography component. Required.
32+
* @param cipherTextStream Stream of ciphertext data to decrypt. Required.
33+
*/
34+
public DecryptRequestAlpha1(String componentName, Flux<byte[]> cipherTextStream) {
35+
this.componentName = componentName;
36+
this.cipherTextStream = cipherTextStream;
37+
}
38+
39+
/**
40+
* Gets the cryptography component name.
41+
*
42+
* @return the component name
43+
*/
44+
public String getComponentName() {
45+
return componentName;
46+
}
47+
48+
/**
49+
* Gets the ciphertext data stream to decrypt.
50+
*
51+
* @return the ciphertext stream as Flux of byte arrays
52+
*/
53+
public Flux<byte[]> getCipherTextStream() {
54+
return cipherTextStream;
55+
}
56+
57+
/**
58+
* Gets the key name (or name/version) to use for decryption.
59+
*
60+
* @return the key name, or null if using the key embedded in the ciphertext
61+
*/
62+
public String getKeyName() {
63+
return keyName;
64+
}
65+
66+
/**
67+
* Sets the key name (or name/version) to decrypt the message.
68+
* This overrides any key reference included in the message if present.
69+
* This is required if the message doesn't include a key reference
70+
* (i.e., was created with omitDecryptionKeyName set to true).
71+
*
72+
* @param keyName the key name to use for decryption
73+
* @return this request instance for method chaining
74+
*/
75+
public DecryptRequestAlpha1 setKeyName(String keyName) {
76+
this.keyName = keyName;
77+
return this;
78+
}
79+
}

0 commit comments

Comments
 (0)