[improve][client] Support load RSA PKCS#8 private key (apache#24582)

Author: gaoran10

pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java

@@ -2854,6 +2854,11 @@ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMe
                 .addEncryptionKey("client-rsa.pem").cryptoKeyReader(new EncKeyReader()).create();
         Producer<byte[]> producer2 = pulsarClient.newProducer().topic("persistent://my-property/my-ns/myrsa-topic1")
                 .addEncryptionKey("client-rsa.pem").cryptoKeyReader(new EncKeyReader()).create();
+        Producer<byte[]> producer3 = pulsarClient.newProducer()
+                .topic("persistent://my-property/my-ns/myrsa-topic1")
+                .addEncryptionKey("client-rsa-pkcs8.pem")
+                .cryptoKeyReader(new EncKeyReader())
+                .create();
 
         for (int i = 0; i < totalMsg; i++) {
             String message = "my-message-" + i;
@@ -2863,20 +2868,24 @@ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMe
             String message = "my-message-" + i;
             producer2.send(message.getBytes());
         }
+        for (int i = totalMsg * 2; i < totalMsg * 3; i++) {
+            String message = "my-message-" + i;
+            producer3.send(message.getBytes());
+        }
 
         MessageImpl<byte[]> msg;
 
         msg = (MessageImpl<byte[]>) normalConsumer.receive(RECEIVE_TIMEOUT_MEDIUM_MILLIS, TimeUnit.MILLISECONDS);
         // should not able to read message using normal message.
         assertNull(msg);
 
-        for (int i = 0; i < totalMsg * 2; i++) {
+        for (int i = 0; i < totalMsg * 3; i++) {
             msg = (MessageImpl<byte[]>) consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
             // verify that encrypted message contains encryption-context
             msg.getEncryptionCtx()
                     .orElseThrow(() -> new IllegalStateException("encryption-ctx not present for encrypted message"));
             String receivedMessage = new String(msg.getData());
-            log.debug("Received message: [{}]", receivedMessage);
+            log.info("Received message: [{}]", receivedMessage);
             String expectedMessage = "my-message-" + i;
             testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage);
         }

pulsar-broker/src/test/resources/certificate/private-key.client-rsa-pkcs8.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC3xJ0urDHj3X4n
+Ul7j8ovzHMUK74hmRIy+n1VuPbUcfaHYXx6lpebAsrFzgvqvNUywXGWEXZ+zlPpD
+DfMupZ0KW2rGAlOQhGWI8384PtBINi06jY0Wa1Apc2EJ43rP9lHR/XkZRHaPDRNR
+Fer3uwlb1hhxyliNRH1GZrfryX3FW1/z9chG+ouc9AvP+/NJYO2Un8klPxsswEDO
+krTgZbdAXwA0jYafViIv+ea7xses+YgOaZ6cv8JpD1rBF0TwO1U2anQRyZ7jVnlZ
+eQMEM4TimCm3i6egHpUnbw0BsSvoUKOrkpXnurCTpfx05sy3yFumFCm8ytditHB5
+3mBEtdd3AgMBAAECggEAPJJEYulkCmDr34ches3r/iaJ9hhM47SNlZlMHemeBDWO
+vpRftCbSP8rGfbX67fxJ+kmCNpZreNROhwKWmNYCxLKsJ3cF1zkkhHcKBF6D5YFh
++A5HJlAZ3IfIwOsnhFuCzY3YOBFeaLldGBl62MvgAlK3vO3+sRWVmzppPgEsgPuU
+KEGFB4EX2JRY001rQcfQ4Oh2VcR5st13/JAXw6aMfJAtYAHAt5rpiYQZyH/t25+a
+aqVTp5Q/oyx/+hNajU0gasvAaRtUGi1HipUkcZ80RTTMFKsMn/gM1kyPJTQMejcC
+o74/xhyGdGQ1hLFijXLMn8JclifIOUDBrVySZagJ1QKBgQD8kwvzhlehBZ2dQkjJ
+T+S2suuuE6UPJDSJiq+7gjy4F11p+NAZv2QhLzIGRlX0WmR0BU0Nt1X50uWIUnwM
+NJuz5JWIWIxcyBoCRD3BFHzq1yGBbP03iHyiYW864+VZy2INJmz0Oww4kC55kzdH
+aB9IVQRjCyC5po38zwHdKJAVXQKBgQC6QqrgATeVWDwKbabbIJY5uqsK8VVphCTi
+4ezRhA5XldaTA3zODg5fo+oa3Yge0CHx7moM5odCqjm8dBPdE0D9vaZ/dEsIVqcZ
+H6oEV+xuLePTXsFDcX80EwtIB/DJs5sqFIcqUYyrPBPHrY9dsfXnohBGlYysd+T5
+yLwgSxoe4wKBgQCmjBQyZeC8piZTf1TdolgPfRXZuMx7moUmORQ+EdgWEBoh+dbH
+U6QkGanj51dghIZ+1swDgmQoE0AVz0ZLChmt2sobmUAdIjm8SuFpwyfH90n5vTOP
+/ekpbhRZgdNHjcw/HZp0xcb+OVzcAA5QgME4Ag+de8X9hW1VSe9RtSClZQKBgHPO
+98eHejPVRJ+HRfOA82a5dZe8LcsJapXWuxQZ84DpnRyTUoobmzxEAk7p5rSh3Dk8
+b5vNydPdW0CdO6x9SC98N4WPnQd2OTcZxPAmas6mobzOcYOpkdmi7aS/qyr+DT/A
+u7nIMppVJEdG1pj/CVIYJd59JZYmhav471900qzZAoGBAKP5f1vSQtcN9PNMUidy
+c3rNF3jVnGmV6LXIeMk0dwlMMANT2X3CzXTUU9y80AXThHEltJtdHD5xjJjRTODv
+rWxAAZRnLDCfv/4meD+3YRWXw8oy6hG81vz4v81ypE5jD37/kwTIfbpSVGxo699J
+c3KEpMHolMtHSTl9WWz3SJwM
+-----END PRIVATE KEY-----

pulsar-broker/src/test/resources/certificate/public-key.client-rsa-pkcs8.pem

@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAt8SdLqwx491+J1Je4/KL
+8xzFCu+IZkSMvp9Vbj21HH2h2F8epaXmwLKxc4L6rzVMsFxlhF2fs5T6Qw3zLqWd
+CltqxgJTkIRliPN/OD7QSDYtOo2NFmtQKXNhCeN6z/ZR0f15GUR2jw0TURXq97sJ
+W9YYccpYjUR9Rma368l9xVtf8/XIRvqLnPQLz/vzSWDtlJ/JJT8bLMBAzpK04GW3
+QF8ANI2Gn1YiL/nmu8bHrPmIDmmenL/CaQ9awRdE8DtVNmp0Ecme41Z5WXkDBDOE
+4pgpt4unoB6VJ28NAbEr6FCjq5KV57qwk6X8dObMt8hbphQpvMrXYrRwed5gRLXX
+dwIDAQAB
+-----END PUBLIC KEY-----

pulsar-client-messagecrypto-bc/src/main/java/org/apache/pulsar/client/impl/crypto/MessageCryptoBc.java

@@ -273,6 +273,9 @@ private PrivateKey loadPrivateKey(byte[] keyBytes) throws Exception {
                 JcaPEMKeyConverter pemConverter = new JcaPEMKeyConverter();
                 privateKey = pemConverter.getPrivateKey(pKeyInfo);
 
+            } else if (pemObj instanceof PrivateKeyInfo) {
+                JcaPEMKeyConverter pemConverter = new JcaPEMKeyConverter();
+                privateKey = pemConverter.getPrivateKey((PrivateKeyInfo) pemObj);
             }
 
             // if our private key is EC type and we have parameters specified