|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one |
| 3 | + * or more contributor license agreements. See the NOTICE file |
| 4 | + * distributed with this work for additional information |
| 5 | + * regarding copyright ownership. The ASF licenses this file |
| 6 | + * to you under the Apache License, Version 2.0 (the |
| 7 | + * "License"); you may not use this file except in compliance |
| 8 | + * with the License. You may obtain a copy of the License at |
| 9 | + * |
| 10 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | + * |
| 12 | + * Unless required by applicable law or agreed to in writing, |
| 13 | + * software distributed under the License is distributed on an |
| 14 | + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | + * KIND, either express or implied. See the License for the |
| 16 | + * specific language governing permissions and limitations |
| 17 | + * under the License. |
| 18 | + */ |
| 19 | +package org.apache.pulsar.client.api.v5; |
| 20 | + |
| 21 | +import static org.testng.Assert.assertEquals; |
| 22 | +import static org.testng.Assert.assertNotNull; |
| 23 | +import static org.testng.Assert.assertNull; |
| 24 | +import static org.testng.Assert.assertTrue; |
| 25 | +import java.nio.file.Path; |
| 26 | +import java.time.Duration; |
| 27 | +import java.util.HashSet; |
| 28 | +import java.util.Set; |
| 29 | +import lombok.Cleanup; |
| 30 | +import org.apache.pulsar.client.api.v5.auth.ConsumerCryptoFailureAction; |
| 31 | +import org.apache.pulsar.client.api.v5.auth.PemFileKeyProvider; |
| 32 | +import org.apache.pulsar.client.api.v5.config.BatchingPolicy; |
| 33 | +import org.apache.pulsar.client.api.v5.config.ConsumerEncryptionPolicy; |
| 34 | +import org.apache.pulsar.client.api.v5.config.ProducerEncryptionPolicy; |
| 35 | +import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition; |
| 36 | +import org.apache.pulsar.client.api.v5.schema.Schema; |
| 37 | +import org.testng.annotations.Test; |
| 38 | + |
| 39 | +/** |
| 40 | + * End-to-end coverage for V5 message encryption: produce → broker → consume |
| 41 | + * round-trip, with payloads encrypted on the producer side and decrypted on the |
| 42 | + * consumer side. Reuses the test PEM keys under {@code certificate/} that the v4 |
| 43 | + * tests already use. |
| 44 | + * |
| 45 | + * <p>Wiring under test: |
| 46 | + * <ul> |
| 47 | + * <li>{@link PemFileKeyProvider} loads PEM bytes from disk on each side.</li> |
| 48 | + * <li>{@link ProducerEncryptionPolicy} / {@link ConsumerEncryptionPolicy} carry |
| 49 | + * the providers + key names + failure actions through the V5 builders.</li> |
| 50 | + * <li>{@link org.apache.pulsar.client.impl.v5.CryptoKeyReaderAdapter} bridges |
| 51 | + * to the v4 {@code ConsumerImpl} / {@code ProducerImpl} crypto paths.</li> |
| 52 | + * </ul> |
| 53 | + */ |
| 54 | +public class V5EncryptionTest extends V5ClientBaseTest { |
| 55 | + |
| 56 | + private static final String KEY_NAME = "client-rsa"; |
| 57 | + private static final Path PUB_KEY = |
| 58 | + Path.of("./src/test/resources/certificate/public-key.client-rsa.pem"); |
| 59 | + private static final Path PRIV_KEY = |
| 60 | + Path.of("./src/test/resources/certificate/private-key.client-rsa.pem"); |
| 61 | + |
| 62 | + private static PemFileKeyProvider producerKeys() { |
| 63 | + return PemFileKeyProvider.builder() |
| 64 | + .publicKey(KEY_NAME, PUB_KEY) |
| 65 | + .build(); |
| 66 | + } |
| 67 | + |
| 68 | + private static PemFileKeyProvider consumerKeys() { |
| 69 | + return PemFileKeyProvider.builder() |
| 70 | + .privateKey(KEY_NAME, PRIV_KEY) |
| 71 | + .build(); |
| 72 | + } |
| 73 | + |
| 74 | + private static ProducerEncryptionPolicy producerPolicy() { |
| 75 | + return ProducerEncryptionPolicy.builder() |
| 76 | + .publicKeyProvider(producerKeys()) |
| 77 | + .keyName(KEY_NAME) |
| 78 | + .build(); |
| 79 | + } |
| 80 | + |
| 81 | + private static ConsumerEncryptionPolicy consumerPolicy() { |
| 82 | + return ConsumerEncryptionPolicy.builder() |
| 83 | + .privateKeyProvider(consumerKeys()) |
| 84 | + .failureAction(ConsumerCryptoFailureAction.FAIL) |
| 85 | + .build(); |
| 86 | + } |
| 87 | + |
| 88 | + /** Single-segment round trip: producer encrypts, consumer decrypts, payload matches. */ |
| 89 | + @Test |
| 90 | + public void testProducerConsumerRoundTrip() throws Exception { |
| 91 | + String topic = newScalableTopic(1); |
| 92 | + |
| 93 | + @Cleanup |
| 94 | + Producer<String> producer = v5Client.newProducer(Schema.string()) |
| 95 | + .topic(topic) |
| 96 | + .encryptionPolicy(producerPolicy()) |
| 97 | + .create(); |
| 98 | + @Cleanup |
| 99 | + QueueConsumer<String> consumer = v5Client.newQueueConsumer(Schema.string()) |
| 100 | + .topic(topic) |
| 101 | + .subscriptionName("crypto-sub") |
| 102 | + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) |
| 103 | + .encryptionPolicy(consumerPolicy()) |
| 104 | + .subscribe(); |
| 105 | + |
| 106 | + producer.newMessage().value("hello-encrypted").send(); |
| 107 | + |
| 108 | + Message<String> msg = consumer.receive(Duration.ofSeconds(5)); |
| 109 | + assertNotNull(msg, "consumer must receive the encrypted-then-decrypted message"); |
| 110 | + assertEquals(msg.value(), "hello-encrypted"); |
| 111 | + consumer.acknowledge(msg.id()); |
| 112 | + } |
| 113 | + |
| 114 | + /** |
| 115 | + * Multi-segment scalable topic: messages spread across segments by key, each |
| 116 | + * segment's per-segment v4 producer/consumer carries the same crypto config, |
| 117 | + * so every message decrypts correctly regardless of which segment it landed on. |
| 118 | + */ |
| 119 | + @Test |
| 120 | + public void testEncryptionAcrossMultipleSegments() throws Exception { |
| 121 | + String topic = newScalableTopic(3); |
| 122 | + |
| 123 | + @Cleanup |
| 124 | + Producer<String> producer = v5Client.newProducer(Schema.string()) |
| 125 | + .topic(topic) |
| 126 | + .encryptionPolicy(producerPolicy()) |
| 127 | + .create(); |
| 128 | + @Cleanup |
| 129 | + QueueConsumer<String> consumer = v5Client.newQueueConsumer(Schema.string()) |
| 130 | + .topic(topic) |
| 131 | + .subscriptionName("crypto-multi-sub") |
| 132 | + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) |
| 133 | + .encryptionPolicy(consumerPolicy()) |
| 134 | + .subscribe(); |
| 135 | + |
| 136 | + int n = 30; |
| 137 | + Set<String> sent = new HashSet<>(); |
| 138 | + for (int i = 0; i < n; i++) { |
| 139 | + String value = "msg-" + i; |
| 140 | + producer.newMessage().key("k-" + i).value(value).send(); |
| 141 | + sent.add(value); |
| 142 | + } |
| 143 | + |
| 144 | + Set<String> received = new HashSet<>(); |
| 145 | + for (int i = 0; i < n; i++) { |
| 146 | + Message<String> msg = consumer.receive(Duration.ofSeconds(5)); |
| 147 | + assertNotNull(msg, "expected message #" + (i + 1)); |
| 148 | + received.add(msg.value()); |
| 149 | + consumer.acknowledge(msg.id()); |
| 150 | + } |
| 151 | + assertEquals(received, sent, "every encrypted message must decrypt to its original value"); |
| 152 | + } |
| 153 | + |
| 154 | + /** |
| 155 | + * Consumer with {@link ConsumerCryptoFailureAction#CONSUME} and no |
| 156 | + * {@link org.apache.pulsar.client.api.v5.auth.PrivateKeyProvider} configured |
| 157 | + * sees the still-encrypted payload, demonstrating the "I don't decrypt; just |
| 158 | + * give me the bytes" mode. |
| 159 | + * |
| 160 | + * <p>Batching disabled on the producer: v4 drops batched encrypted messages |
| 161 | + * even under CONSUME because it can't reframe a batch envelope it can't open. |
| 162 | + */ |
| 163 | + @Test |
| 164 | + public void testConsumerWithoutProviderAndConsumeAction() throws Exception { |
| 165 | + String topic = newScalableTopic(1); |
| 166 | + |
| 167 | + @Cleanup |
| 168 | + Producer<String> producer = v5Client.newProducer(Schema.string()) |
| 169 | + .topic(topic) |
| 170 | + .batchingPolicy(BatchingPolicy.ofDisabled()) |
| 171 | + .encryptionPolicy(producerPolicy()) |
| 172 | + .create(); |
| 173 | + |
| 174 | + @Cleanup |
| 175 | + QueueConsumer<byte[]> consumer = v5Client.newQueueConsumer(Schema.bytes()) |
| 176 | + .topic(topic) |
| 177 | + .subscriptionName("crypto-consume-sub") |
| 178 | + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) |
| 179 | + .encryptionPolicy(ConsumerEncryptionPolicy.builder() |
| 180 | + .failureAction(ConsumerCryptoFailureAction.CONSUME) |
| 181 | + .build()) |
| 182 | + .subscribe(); |
| 183 | + |
| 184 | + producer.newMessage().value("plaintext-marker").send(); |
| 185 | + |
| 186 | + Message<byte[]> msg = consumer.receive(Duration.ofSeconds(5)); |
| 187 | + assertNotNull(msg, "CONSUME must deliver the message even without a private key"); |
| 188 | + // Payload is still encrypted — must not contain the plaintext marker. |
| 189 | + String body = new String(msg.value()); |
| 190 | + assertTrue(!body.contains("plaintext-marker"), |
| 191 | + "payload should still be encrypted, got: " + body); |
| 192 | + consumer.acknowledge(msg.id()); |
| 193 | + } |
| 194 | + |
| 195 | + /** |
| 196 | + * Consumer with {@link ConsumerCryptoFailureAction#DISCARD} and no provider |
| 197 | + * silently drops undecryptable messages (cursor advances) — the application |
| 198 | + * never sees them. |
| 199 | + */ |
| 200 | + @Test |
| 201 | + public void testConsumerWithoutProviderAndDiscardAction() throws Exception { |
| 202 | + String topic = newScalableTopic(1); |
| 203 | + |
| 204 | + @Cleanup |
| 205 | + Producer<String> producer = v5Client.newProducer(Schema.string()) |
| 206 | + .topic(topic) |
| 207 | + .batchingPolicy(BatchingPolicy.ofDisabled()) |
| 208 | + .encryptionPolicy(producerPolicy()) |
| 209 | + .create(); |
| 210 | + |
| 211 | + @Cleanup |
| 212 | + QueueConsumer<String> consumer = v5Client.newQueueConsumer(Schema.string()) |
| 213 | + .topic(topic) |
| 214 | + .subscriptionName("crypto-discard-sub") |
| 215 | + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) |
| 216 | + .encryptionPolicy(ConsumerEncryptionPolicy.builder() |
| 217 | + .failureAction(ConsumerCryptoFailureAction.DISCARD) |
| 218 | + .build()) |
| 219 | + .subscribe(); |
| 220 | + |
| 221 | + producer.newMessage().value("classified").send(); |
| 222 | + |
| 223 | + Message<String> msg = consumer.receive(Duration.ofMillis(500)); |
| 224 | + assertNull(msg, "DISCARD must drop the undecryptable message before delivery"); |
| 225 | + } |
| 226 | +} |
0 commit comments