This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 03e4280fafc4 CAMEL-23847: Camel-PQC - stream sign/verify for large
payloads (#24644)
03e4280fafc4 is described below
commit 03e4280fafc4ffac0157bfdc5dbe37c2a4f8a67d
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Jul 13 12:24:12 2026 +0200
CAMEL-23847: Camel-PQC - stream sign/verify for large payloads (#24644)
CAMEL-23847: Camel-PQC stream sign/verify for large payloads
The sign/verify operations previously materialised the whole message body
as a
String (getMandatoryBody(String.class)) before feeding it to
Signature.update(),
loading everything into memory and corrupting non-UTF-8 binary payloads.
Feed the signature directly from the body instead:
- a byte[] body is used as-is;
- an InputStream body is read in chunks (and reset afterwards when it is a
re-readable StreamCache, so downstream processors still see the payload);
- any other body type falls back to its String representation (charset
handling
is CAMEL-23843).
The hybrid sign/verify variants now obtain the payload as a byte[] without a
String round-trip. Large payloads (files, large messages) can now be signed
and
verified without loading them fully into memory.
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../camel-pqc/src/main/docs/pqc-component.adoc | 2 +
.../apache/camel/component/pqc/PQCProducer.java | 76 +++++++++---
.../component/pqc/PQCStreamingSignatureTest.java | 130 +++++++++++++++++++++
3 files changed, 195 insertions(+), 13 deletions(-)
diff --git a/components/camel-pqc/src/main/docs/pqc-component.adoc
b/components/camel-pqc/src/main/docs/pqc-component.adoc
index f5c0edb443eb..838534e7111c 100644
--- a/components/camel-pqc/src/main/docs/pqc-component.adoc
+++ b/components/camel-pqc/src/main/docs/pqc-component.adoc
@@ -143,6 +143,8 @@ This will be true for standardized algorithms and for
experimental ones.
TIP: To combine a classical signature (ECDSA, Ed25519, RSA) with a PQC
signature for defense-in-depth during the post-quantum transition, use the
hybrid signature operations described in xref:others:pqc-hybrid.adoc[Hybrid
Cryptography].
+NOTE: The sign and verify operations stream the message body. A `byte[]` or
`InputStream` body is fed to the signature in chunks without being materialised
as a String, so large payloads (files, large messages) can be signed and
verified without loading them fully into memory. When the body is a re-readable
`StreamCache` (for example with stream caching enabled) it is reset after
reading so downstream processors still see the payload.
+
== Examples
All signature algorithms follow the same route pattern. Below is a complete
example using ML-DSA, followed by a reference table for the other algorithms.
diff --git
a/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/PQCProducer.java
b/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/PQCProducer.java
index aae685a542da..8906a5a2cfa7 100644
---
a/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/PQCProducer.java
+++
b/components/camel-pqc/src/main/java/org/apache/camel/component/pqc/PQCProducer.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.pqc;
+import java.io.IOException;
+import java.io.InputStream;
import java.security.*;
import java.security.cert.Certificate;
import java.util.Arrays;
@@ -32,7 +34,9 @@ import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.Message;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.StreamCache;
import org.apache.camel.component.pqc.crypto.hybrid.HybridKEM;
import org.apache.camel.component.pqc.crypto.hybrid.HybridSignature;
import org.apache.camel.component.pqc.lifecycle.KeyLifecycleManager;
@@ -107,6 +111,8 @@ public class PQCProducer extends DefaultProducer {
PQCKeyEncapsulationAlgorithms.SNTRUPrime.name(),
PQCKeyEncapsulationAlgorithms.KYBER.name());
+ private static final int STREAM_BUFFER_SIZE = 8192;
+
private Signature signer;
private KeyGenerator keyGenerator;
private KeyPair keyPair;
@@ -479,10 +485,8 @@ public class PQCProducer extends DefaultProducer {
throws Exception {
checkStatefulKeyBeforeSign();
- String payload = exchange.getMessage().getMandatoryBody(String.class);
-
signer.initSign(keyPair.getPrivate());
- signer.update(payload.getBytes());
+ updateSignatureFromBody(signer, exchange.getMessage());
byte[] signature = signer.sign();
exchange.getMessage().setHeader(PQCConstants.SIGNATURE, signature);
@@ -491,11 +495,9 @@ public class PQCProducer extends DefaultProducer {
}
private void verification(Exchange exchange)
- throws InvalidPayloadException, InvalidKeyException,
SignatureException {
- String payload = exchange.getMessage().getMandatoryBody(String.class);
-
+ throws InvalidPayloadException, InvalidKeyException,
SignatureException, IOException {
signer.initVerify(keyPair.getPublic());
- signer.update(payload.getBytes());
+ updateSignatureFromBody(signer, exchange.getMessage());
if
(signer.verify(exchange.getMessage().getHeader(PQCConstants.SIGNATURE,
byte[].class))) {
exchange.getMessage().setHeader(PQCConstants.VERIFY, true);
} else {
@@ -503,6 +505,56 @@ public class PQCProducer extends DefaultProducer {
}
}
+ /**
+ * Feeds the message body into the given {@link Signature} without
materialising the whole payload as a String. A
+ * {@code byte[]} body is used directly, an {@link InputStream} body is
read in chunks (and reset afterwards when it
+ * is a re-readable {@link StreamCache} so downstream processors still see
the payload), and any other body type
+ * falls back to its String representation.
+ */
+ private static void updateSignatureFromBody(Signature signature, Message
message)
+ throws InvalidPayloadException, SignatureException, IOException {
+ Object body = message.getBody();
+ if (body instanceof byte[] bytes) {
+ signature.update(bytes);
+ } else if (body instanceof InputStream stream) {
+ try {
+ byte[] buffer = new byte[STREAM_BUFFER_SIZE];
+ int read;
+ while ((read = stream.read(buffer)) != -1) {
+ signature.update(buffer, 0, read);
+ }
+ } finally {
+ if (body instanceof StreamCache streamCache) {
+ streamCache.reset();
+ }
+ }
+ } else {
+
signature.update(message.getMandatoryBody(String.class).getBytes());
+ }
+ }
+
+ /**
+ * Returns the message body as a byte array without a String round-trip
when it is already binary. Used by the
+ * hybrid operations, which need the whole payload in memory. An {@link
InputStream} is fully read (and reset
+ * afterwards when it is a re-readable {@link StreamCache}).
+ */
+ private static byte[] bodyToByteArray(Message message) throws
InvalidPayloadException, IOException {
+ Object body = message.getBody();
+ if (body instanceof byte[] bytes) {
+ return bytes;
+ } else if (body instanceof InputStream stream) {
+ try {
+ return stream.readAllBytes();
+ } finally {
+ if (body instanceof StreamCache streamCache) {
+ streamCache.reset();
+ }
+ }
+ } else {
+ return message.getMandatoryBody(String.class).getBytes();
+ }
+ }
+
private void generateEncapsulation(Exchange exchange)
throws InvalidAlgorithmParameterException {
// initialise for creating an encapsulation and shared secret.
@@ -563,11 +615,10 @@ public class PQCProducer extends DefaultProducer {
// ========== Hybrid Signature Operations ==========
private void hybridSignature(Exchange exchange)
- throws InvalidPayloadException, InvalidKeyException,
SignatureException {
+ throws InvalidPayloadException, InvalidKeyException,
SignatureException, IOException {
checkStatefulKeyBeforeSign();
- String payload = exchange.getMessage().getMandatoryBody(String.class);
- byte[] data = payload.getBytes();
+ byte[] data = bodyToByteArray(exchange.getMessage());
if (classicalSigner == null || classicalKeyPair == null) {
throw new IllegalStateException(
@@ -600,9 +651,8 @@ public class PQCProducer extends DefaultProducer {
}
private void hybridVerification(Exchange exchange)
- throws InvalidPayloadException, InvalidKeyException,
SignatureException {
- String payload = exchange.getMessage().getMandatoryBody(String.class);
- byte[] data = payload.getBytes();
+ throws InvalidPayloadException, InvalidKeyException,
SignatureException, IOException {
+ byte[] data = bodyToByteArray(exchange.getMessage());
byte[] hybridSig =
exchange.getMessage().getHeader(PQCConstants.HYBRID_SIGNATURE, byte[].class);
if (hybridSig == null) {
diff --git
a/components/camel-pqc/src/test/java/org/apache/camel/component/pqc/PQCStreamingSignatureTest.java
b/components/camel-pqc/src/test/java/org/apache/camel/component/pqc/PQCStreamingSignatureTest.java
new file mode 100644
index 000000000000..12d0f3b8ca3b
--- /dev/null
+++
b/components/camel-pqc/src/test/java/org/apache/camel/component/pqc/PQCStreamingSignatureTest.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.pqc;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.security.Security;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PQCStreamingSignatureTest extends CamelTestSupport {
+
+ @EndpointInject("mock:verify")
+ protected MockEndpoint resultVerify;
+
+ @BeforeAll
+ public static void startup() {
+ if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) {
+ Security.addProvider(new BouncyCastleProvider());
+ }
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ // Exercise the re-readable StreamCache path (sign then verify in the
same route)
+ context.setStreamCaching(true);
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:signOnly").to("pqc:sign?operation=sign&signatureAlgorithm=MLDSA");
+
from("direct:verifyOnly").to("pqc:verify?operation=verify&signatureAlgorithm=MLDSA");
+ from("direct:signVerify")
+ .to("pqc:sign?operation=sign&signatureAlgorithm=MLDSA")
+
.to("pqc:verify?operation=verify&signatureAlgorithm=MLDSA")
+ .to("mock:verify");
+ }
+ };
+ }
+
+ private byte[] sign(Object body) {
+ Exchange out = template.request("direct:signOnly", e ->
e.getMessage().setBody(body));
+ return out.getMessage().getHeader(PQCConstants.SIGNATURE,
byte[].class);
+ }
+
+ private boolean verify(Object body, byte[] signature) {
+ Exchange out = template.request("direct:verifyOnly", e -> {
+ e.getMessage().setBody(body);
+ e.getMessage().setHeader(PQCConstants.SIGNATURE, signature);
+ });
+ return out.getMessage().getHeader(PQCConstants.VERIFY, Boolean.class);
+ }
+
+ @Test
+ void testByteArrayRoundTrip() {
+ byte[] content = "hello streaming
world".getBytes(StandardCharsets.UTF_8);
+ byte[] sig = sign(content);
+ assertNotNull(sig);
+ assertTrue(verify(content, sig));
+ }
+
+ @Test
+ void testSignInputStreamVerifyBytes() {
+ byte[] content = "signed from an input
stream".getBytes(StandardCharsets.UTF_8);
+ byte[] sig = sign(new ByteArrayInputStream(content));
+ assertNotNull(sig);
+ assertTrue(verify(content, sig), "a streamed sign must verify against
the same bytes");
+ }
+
+ @Test
+ void testSignBytesVerifyInputStream() {
+ byte[] content = "verified from an input
stream".getBytes(StandardCharsets.UTF_8);
+ byte[] sig = sign(content);
+ assertTrue(verify(new ByteArrayInputStream(content), sig), "a streamed
verify must accept the matching bytes");
+ }
+
+ @Test
+ void testBinaryPayloadNotCorrupted() {
+ // Bytes that are not valid UTF-8: a String round-trip would corrupt
them and break verification
+ byte[] binary = new byte[512];
+ for (int i = 0; i < binary.length; i++) {
+ binary[i] = (byte) (i % 256);
+ }
+ assertTrue(verify(binary, sign(binary)));
+ assertTrue(verify(binary, sign(new ByteArrayInputStream(binary))));
+ }
+
+ @Test
+ void testLargeStreamSignVerifyInOneRoute() throws Exception {
+ byte[] large = new byte[1024 * 1024]; // 1 MB
+ for (int i = 0; i < large.length; i++) {
+ large[i] = (byte) i;
+ }
+ resultVerify.expectedMessageCount(1);
+ template.sendBody("direct:signVerify", new
ByteArrayInputStream(large));
+ resultVerify.assertIsSatisfied();
+
assertTrue(resultVerify.getExchanges().get(0).getMessage().getHeader(PQCConstants.VERIFY,
Boolean.class),
+ "large streamed payload should sign and verify within one
route");
+ }
+}
